Get started with Unstructured Transform for Pydantic AI

pip install -U pydantic-ai

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

import os

from pydantic_ai.mcp import MCPToolset

transform = MCPToolset(
    "https://mcp.transform.unstructured.io",
    auth=os.environ["UNSTRUCTURED_API_KEY"],
)

import asyncio
import mimetypes
import os
from pathlib import Path

import httpx
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset

transform = MCPToolset(
    "https://mcp.transform.unstructured.io",
    auth=os.environ["UNSTRUCTURED_API_KEY"],
)

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    toolsets=[transform],
    system_prompt=(
        "You partition documents with the Unstructured Transform tools. "
        "To partition a local file: call file_info to get its name, MIME type, "
        "and size; pass those to request_file_upload_url; call put_file to "
        "upload the bytes to the returned signed URL; then call "
        "start_transform_job with the returned file_ref. Poll "
        "check_job_status until the job is COMPLETED, then call "
        "get_job_results and report a short summary."
    ),
)


@agent.tool_plain
def file_info(local_path: str) -> dict:
    """Return the filename, MIME type, and byte size of a local file."""
    path = Path(local_path)
    content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return {
        "filename": path.name,
        "content_type": content_type,
        "size_bytes": path.stat().st_size,
    }


@agent.tool_plain
def put_file(local_path: str, upload_url: str, content_type: str) -> str:
    """Upload the bytes of a local file to a signed upload URL."""
    data = Path(local_path).read_bytes()
    response = httpx.put(
        upload_url,
        content=data,
        headers={"Content-Type": content_type},
        timeout=60,
    )
    response.raise_for_status()
    return f"Uploaded {len(data)} bytes."


async def main() -> None:
    async with agent:
        result = await agent.run(
            "Partition the file at ./report.pdf and give me a short summary of what it contains."
        )
        print(result.output)


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

from dataclasses import dataclass

from pydantic_ai import Agent, RunContext
from pydantic_ai.mcp import MCPToolset


@dataclass
class UserDeps:
    unstructured_api_key: str


agent = Agent("anthropic:claude-sonnet-4-6", deps_type=UserDeps)


@agent.toolset
def transform_toolset(ctx: RunContext[UserDeps]) -> MCPToolset:
    return MCPToolset(
        "https://mcp.transform.unstructured.io",
        auth=ctx.deps.unstructured_api_key,
    )