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 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.
• An Unstructured API key, which the Transform MCP server uses as a bearer token. Get an API key.
• 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:

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 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:

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:

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]

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.

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:

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.

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:

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:

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. 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.
• 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: 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: Control how the Unstructured Transform MCP server extracts and formats structured data from your files.
• Control Transform generated sample 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.