Get started with Unstructured Transform for DSPy

pip install "dspy==3.2.1" "mcp==1.28.1" httpx

export UNSTRUCTURED_API_KEY="<your-unstructured-api-key>"

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]

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())

export OPENAI_API_KEY="<your-openai-api-key>"

python transform_agent.py

Markdown output (first 500 characters):

# Dummy PDF file

Element count: 1

class ExtractStructuredData(dspy.Signature):
    """Extract structured data from a local document with the Unstructured Transform
    tools. Follow this exact sequence:

    1. Upload and partition 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()

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)