> ## 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.

# Unstructured Transform MCP server installation 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, transform_files,
        # check_transform_status, and get_transform_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 transform_files with file_refs set to a one-item list containing
       the returned file_ref.
    5. Poll check_transform_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_transform_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.

## 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.

## 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_transform_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 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).