> ## Documentation Index
> Fetch the complete documentation index at: https://docs.unstructured.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Get started with Unstructured Transform for DSPy

> Learn how to connect the Unstructured Transform MCP server to DSPy agents. Your agents can then point to your files and have Transform start producing partitioned, enriched, chunked, and embedded data based on your files in minutes.

[DSPy](https://dspy.ai) is an open-source Python framework for programming, rather than prompting, language models.
Unlike end-user AI tools that add the Transform MCP server through a settings screen, DSPy connects to it *from code*:
you load the Transform MCP server's tools through the official MCP Python SDK and hand them to a DSPy agent such as
`dspy.ReAct`. Because Transform is a hosted *remote* MCP server, there is nothing to install or run locally: you point
the client at the server URL and authenticate with your Unstructured API key.

## Requirements

Before you begin, you must have the following:

* Python 3.10 or later on your local development machine. To check, in your terminal, run `python --version` or `python3 --version`. [Install Python](https://www.python.org/downloads/).
* An Unstructured API key, which the Transform MCP server uses as a bearer token. [Get an API key](/transform/code#get-your-unstructured-api-key-and-url).
* An OpenAI API key, or another model provider that DSPy supports, for the agent's underlying chat model.

## Install the packages

In your terminal, install DSPy, the MCP Python SDK, and `httpx` for the pre-signed upload and download requests:

```bash theme={null}
pip install "dspy==3.2.1" "mcp==1.28.1" httpx
```

This guide was verified against the pinned versions shown. Newer versions typically work, but check the
[DSPy release notes](https://github.com/stanfordnlp/dspy/releases) if an API has changed.

## Connect to the Transform MCP server

The Transform MCP server speaks the streamable HTTP transport and authenticates with your Unstructured API key sent as
a bearer token. Store your key in an environment variable rather than hard-coding it:

```bash theme={null}
export UNSTRUCTURED_API_KEY="<your-unstructured-api-key>"
```

Open a session with the MCP Python SDK, then convert the server's tools into DSPy tools with
`dspy.Tool.from_mcp_tool`:

```python theme={null}
import os

import dspy
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client(
    "https://mcp.transform.unstructured.io",
    headers={"Authorization": f"Bearer {os.environ['UNSTRUCTURED_API_KEY']}"},
) as (read_stream, write_stream, _):
    async with ClientSession(read_stream, write_stream) as session:
        await session.initialize()
        listing = await session.list_tools()

        # Loads the Transform tools: request_file_upload_url, start_transform_job,
        # check_job_status, and get_job_results.
        tools = [dspy.Tool.from_mcp_tool(session, tool) for tool in listing.tools]
```

<Tip>
  The converted tools call back into the MCP session, so build and run your agent inside the two `async with` blocks.
  Tools used after the session closes fail with a closed-connection error.
</Tip>

## Add local file helpers for uploads, downloads, and waiting

Transform's job lifecycle mixes MCP tool calls with two plain HTTP requests that are not MCP calls: an HTTP `PUT` of
the raw file bytes to a pre-signed `upload_url`, and an HTTP `GET` of the finished output from a pre-signed
`download_url`. An agent cannot perform raw HTTP requests through MCP tools alone. The full example in the next
section registers four small local functions as additional agent tools:

* `describe_local_file`: Returns a file's filename, content type, and exact size in bytes, which
  `request_file_upload_url` requires.
* `upload_local_file`: Sends the file bytes to the pre-signed `upload_url` with an HTTP `PUT`.
* `download_text`: Retrieves the finished Markdown from the pre-signed `download_url` with an HTTP `GET`.
* `wait_seconds`: Pauses between status checks. Each `dspy.ReAct` step consumes one of the agent's `max_iters`
  iterations, so pacing the polling keeps a multi-minute transform from exhausting the iteration budget.

Pre-signed URLs carry their own credentials in the URL itself. The helpers must not send the `Authorization` header on
the upload and download requests, or the storage service rejects them.

## Use the tools in a ReAct agent

The following example builds a `dspy.ReAct` agent that can drive the full Transform job lifecycle. The agent requests
an upload URL, uploads the file through the local helper, starts the transform, polls for status with paced waits, and
downloads the Markdown output:

```python theme={null}
import asyncio
import mimetypes
import os
from pathlib import Path

import dspy
import httpx
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

TRANSFORM_MCP_URL = "https://mcp.transform.unstructured.io"


def describe_local_file(path: str) -> dict:
    """Get the filename, content type, and exact size in bytes of a local file."""
    file = Path(path).expanduser()
    content_type = mimetypes.guess_type(file.name)[0] or "application/octet-stream"
    return {
        "filename": file.name,
        "content_type": content_type,
        "size_bytes": file.stat().st_size,
    }


def upload_local_file(path: str, upload_url: str, content_type: str) -> str:
    """Upload a local file's raw bytes to a pre-signed upload_url with an HTTP PUT."""
    # The URL is pre-signed: do not send an Authorization header here.
    file = Path(path).expanduser()
    response = httpx.put(
        upload_url,
        content=file.read_bytes(),
        headers={"Content-Type": content_type},
        timeout=120.0,
    )
    response.raise_for_status()
    return f"Uploaded {file.name}, status {response.status_code}."


def download_text(url: str) -> str:
    """Download the transformed output from a pre-signed download_url with an HTTP GET."""
    # The URL is pre-signed: do not send an Authorization header here.
    response = httpx.get(url, timeout=120.0)
    response.raise_for_status()
    return response.text


async def wait_seconds(seconds: int) -> str:
    """Wait the given number of seconds before checking the job status again."""
    await asyncio.sleep(seconds)
    return f"Waited {seconds} seconds."


class TransformDocument(dspy.Signature):
    """Transform a local document into structured Markdown with the Unstructured
    Transform tools. Follow this exact sequence:

    1. Call describe_local_file to get the file's filename, content_type, and
       size_bytes. Report files larger than 52428800 bytes (50 MB) to the user
       instead of uploading them.
    2. Call request_file_upload_url with that filename, content_type, and
       size_bytes.
    3. Call upload_local_file with the file path and the returned upload_url
       and Content-Type header value.
    4. Call start_transform_job with file_refs set to a one-item list containing
       the returned file_ref.
    5. Poll check_job_status with the returned job_id until status is
       COMPLETED. Transforms take from about 10 seconds to several minutes.
       Call wait_seconds with 10 between status checks.
    6. Call get_job_results with the job_id, then call download_text
       with the first file's download_url to retrieve the Markdown output.
    """

    request: str = dspy.InputField()
    answer: str = dspy.OutputField()


async def main() -> None:
    dspy.configure(lm=dspy.LM(os.environ.get("DSPY_MODEL", "openai/gpt-4.1-mini")))

    async with streamablehttp_client(
        TRANSFORM_MCP_URL,
        headers={"Authorization": f"Bearer {os.environ['UNSTRUCTURED_API_KEY']}"},
    ) as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            listing = await session.list_tools()

            tools = [dspy.Tool.from_mcp_tool(session, tool) for tool in listing.tools]
            tools += [
                dspy.Tool(func)
                for func in (describe_local_file, upload_local_file, download_text, wait_seconds)
            ]

            agent = dspy.ReAct(TransformDocument, tools=tools, max_iters=50)
            result = await agent.acall(
                request=(
                    "Transform the document at ./sample.pdf and show me the first "
                    "500 characters of the Markdown output, plus the element count."
                )
            )
            print(result.answer)


if __name__ == "__main__":
    asyncio.run(main())
```

Because DSPy programs compose, the transformed Markdown can feed directly into a downstream `dspy.Predict` module, for
example one with a signature that extracts structured fields from the document text. Transform can also do that
extraction server-side, against a JSON Schema, as shown in
[Extract structured data](#extract-structured-data) below.

## Run the agent

The example defaults to OpenAI as the chat model provider. Set your OpenAI key, place any PDF that is 50 MB or smaller
at `./sample.pdf`, and run the program:

```bash theme={null}
export OPENAI_API_KEY="<your-openai-api-key>"

python transform_agent.py
```

The agent works through the tool sequence and prints a summary such as the following. Transforms take from about 10
seconds to several minutes, depending on page count:

```text theme={null}
Markdown output (first 500 characters):

# Dummy PDF file

Element count: 1
```

To use a different model provider, set the `DSPY_MODEL` environment variable to any
[model string DSPy supports](https://dspy.ai/learn/programming/language_models/). For example, for Azure OpenAI, set
`DSPY_MODEL="azure/<your-deployment-name>"` along with the `AZURE_API_KEY`, `AZURE_API_BASE`, and `AZURE_API_VERSION`
environment variables.

## Extract structured data

To return named fields as JSON rather than the whole document as text, use the server's two extraction tools.
`session.list_tools()` already loads them, so the only change is a signature that drives the extraction sequence.

The tools read the element JSON that a parse produces, not the raw file, so an extraction always follows a transform
job and is keyed by the `output_ref` that `get_job_results` returns for each file:

```python theme={null}
class ExtractStructuredData(dspy.Signature):
    """Extract structured data from a local document with the Unstructured Transform
    tools. Follow this exact sequence:

    1. Upload and parse the document as in TransformDocument steps 1 to 4, passing
       stages {"partition": {"strategy": "vlm"}} to start_transform_job for an image,
       PowerPoint file, or PDF, and {"partition": {"strategy": "fast"}} otherwise.
    2. Poll check_job_status with the returned job_id until status is COMPLETED,
       calling wait_seconds with 10 between checks.
    3. Call get_job_results with the job_id and keep the first file's output_ref. Do
       not call download_text: the extraction tools read the output_ref, not the
       rendered Markdown.
    4. If the request did not supply a schema, call
       suggest_extraction_schema_for_file with that output_ref and report the drafted
       schema before extracting.
    5. Call start_extraction_job with element_json_refs set to a one-item list
       containing the output_ref, and schema_to_extract set to the JSON Schema as a
       JSON string.
    6. Poll check_job_status with the new job_id the same way, then call
       get_job_results. Extraction results are returned inline, so there is no
       download step. Report the results with their filename and do not strip the
       surrounding provenance fields.
    """

    request: str = dspy.InputField()
    extracted_json: str = dspy.OutputField()
```

Run it with the same tool list as the transform agent:

```python theme={null}
agent = dspy.ReAct(ExtractStructuredData, tools=tools, max_iters=50)
result = await agent.acall(
    request=(
        "Extract the vendor, invoice number, and total from ./invoice.pdf as JSON. "
        "Draft a schema from the document itself and show it to me first."
    )
)
print(result.extracted_json)
```

Each result is wrapped with the source filename, file type, timestamp, and the element JSON reference it was taken
from, alongside the extracted data. Keep that wrapper when you persist results, since it is what ties each record
back to its document.

<Tip>
  Extraction can only surface what the parse captured, so parse quality sets the ceiling. If a result comes back
  sparse, re-parse with `strategy: "hi_res"` plus the `image_description`, `generative_ocr`, and `table_to_html`
  enrichment steps, then extract from the new `output_ref`. For more prompt patterns, see
  [Structured data extraction](/transform/sde).
</Tip>

## Parse your source files

Parsing requests have the following limits:

* Each file must be of a [supported file type](/transform/supported-file-types).
* Each file must be 50 MB or less in size.
* Each request must have 10 files or fewer.
* Only 5 requests can be running at a time.

The Transform MCP server is designed to report these limits back to the agent through its tool responses. Because of
this, your agent should notify you whenever it encounters a file that exceeds 50 MB in size, and it should formulate
strategies to send requests that are 10 files or fewer and not cause more than 5 requests to be running at a time. You
can reinforce this behavior in the agent's signature docstring, as the example above does for the file size limit.

## Troubleshooting

* **`401 Unauthorized` or an `invalid_token` error on connect.** Confirm the `UNSTRUCTURED_API_KEY` environment
  variable is set and that the `Authorization` header is formatted as `Bearer <your-unstructured-api-key>`.
* **`Session terminated` or `404` when connecting.** Verify the server URL is exactly
  `https://mcp.transform.unstructured.io`, with no path appended.
* **A closed-connection error when a tool runs.** The MCP tools are bound to the session, so run the agent inside the
  `streamablehttp_client` and `ClientSession` context managers, as the example does.
* **The agent stops before the job finishes.** Each ReAct step consumes one of `max_iters`, and each poll cycle uses
  two steps (`wait_seconds` plus `check_job_status`), so `max_iters=50` covers roughly four minutes of
  polling. For longer transforms, raise `max_iters` or increase the wait between checks.
* **The upload or download request is rejected.** The `upload_url` and `download_url` values are pre-signed. Send those
  two requests without the `Authorization` header.

## Next steps

* [Control Transform file parsing output](/transform/output): Control how the Unstructured Transform MCP server instructs Transform to partition, enrich, chunk, and embed the data based on your files.
* [Control Transform structured data extraction](/transform/sde): Control how the Unstructured Transform MCP server extracts and formats structured data from your files.
* [Control Transform generated sample code](/transform/code): Control how the Unstructured Transform MCP server generates sample curl or Python code that demonstrates how to use Transform to partition, enrich, chunk, and embed the data based on your files.

## Questions? Need help?

* For technical support, [request support](/support/request).