Unstructured Transform MCP server installation for Google ADK

pip install "google-adk[mcp]"

export UNSTRUCTURED_API_KEY="<your-unstructured-api-key>"
export GOOGLE_API_KEY="<your-gemini-api-key>"
export GOOGLE_GENAI_USE_VERTEXAI=FALSE

import os

from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

transform_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://mcp.transform.unstructured.io",
        headers={"Authorization": f"Bearer {os.environ['UNSTRUCTURED_API_KEY']}"},
        timeout=30.0,
        sse_read_timeout=300.0,
    ),
    tool_filter=[
        "request_file_upload_url",
        "start_transform_job",
        "suggest_extraction_schema_for_file",
        "start_extraction_job",
        "check_job_status",
        "get_job_results",
    ],
)

transform_agent/
├── __init__.py
└── agent.py

from . import agent

import asyncio
import os
import pathlib

import httpx
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

TRANSFORM_MCP_URL = "https://mcp.transform.unstructured.io"  # root URL; do not append /mcp

MAX_UPLOAD_BYTES = 50 * 1024 * 1024  # the server's per-file limit


async def upload_local_file(local_path: str, upload_url: str) -> dict:
    """Upload a local file's bytes to a pre-signed upload URL with an HTTP PUT.

    Use this after request_file_upload_url returns an upload URL for a local file.
    The pre-signed URL carries its own credentials: no Authorization header is sent.
    Only files inside the current working directory can be uploaded, and only to
    https:// destinations.

    Args:
        local_path: Path to the file, inside the current working directory.
        upload_url: The pre-signed upload URL returned by request_file_upload_url.

    Returns:
        dict with 'status_code' and 'bytes_uploaded', or 'error'.
    """
    path = pathlib.Path(local_path).resolve()
    if not path.is_relative_to(pathlib.Path.cwd()):
        return {"error": "local_path must be inside the current working directory"}
    if not path.is_file():
        return {"error": f"file not found: {local_path}"}
    if not upload_url.startswith("https://"):
        return {"error": "upload_url must be an https:// URL"}
    if path.stat().st_size > MAX_UPLOAD_BYTES:
        return {"error": "file exceeds the 50 MB per-file limit"}
    data = path.read_bytes()
    try:
        async with httpx.AsyncClient() as client:
            response = await client.put(
                upload_url,
                content=data,
                headers={"Content-Type": "application/octet-stream"},
                timeout=120.0,
            )
    except httpx.HTTPError as exc:
        return {"error": f"upload failed: {exc}"}
    if response.status_code >= 300:
        return {"error": f"upload failed with HTTP {response.status_code}"}
    return {"status_code": response.status_code, "bytes_uploaded": len(data)}


async def wait_seconds(seconds: int) -> dict:
    """Pause before the next status check. Transform jobs take 30 seconds to a few
    minutes; call this between check_job_status calls instead of polling
    in a tight loop.

    Args:
        seconds: How long to wait. Use 30 unless told otherwise.

    Returns:
        dict confirming the wait.
    """
    seconds = max(1, min(int(seconds), 120))
    await asyncio.sleep(seconds)
    return {"waited_seconds": seconds}


transform_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url=TRANSFORM_MCP_URL,
        headers={"Authorization": f"Bearer {os.environ['UNSTRUCTURED_API_KEY']}"},
        timeout=30.0,
        sse_read_timeout=300.0,
    ),
    tool_filter=[
        "request_file_upload_url",
        "start_transform_job",
        "suggest_extraction_schema_for_file",
        "start_extraction_job",
        "check_job_status",
        "get_job_results",
    ],
)

root_agent = LlmAgent(
    name="transform_agent",
    model="gemini-flash-latest",
    description="Parses documents into structured data using Unstructured Transform.",
    instruction=(
        "You parse documents with the Unstructured Transform MCP server.\n"
        "Workflow:\n"
        "1. For a file at a public https:// URL, pass the URL straight to start_transform_job.\n"
        "2. For a local file: call request_file_upload_url, then upload_local_file with the\n"
        "   returned upload URL, then start_transform_job with the returned file reference.\n"
        "3. start_transform_job returns a job_id. Poll with check_job_status, calling\n"
        "   wait_seconds(30) between checks. Jobs take 30 seconds to a few minutes,\n"
        "   and unpaced polling also burns through model rate limits.\n"
        "4. When the job is complete, call get_job_results and report the parsed\n"
        "   content (or download URLs) back to the user.\n"
        "If the user asks for specific fields rather than the whole document, extract\n"
        "instead of just parsing:\n"
        "5. Parse the file first as above, and keep the output_ref that get_job_results\n"
        "   returns for it. The extraction tools read that element JSON reference, not\n"
        "   the raw file, so do not download the parsed output before extracting.\n"
        "6. If the user gave no schema, call suggest_extraction_schema_for_file with the\n"
        "   output_ref and show the draft before extracting.\n"
        "7. Call start_extraction_job with element_json_refs set to the output_refs and\n"
        "   schema_to_extract set to a JSON Schema passed as a JSON string, then poll\n"
        "   and read it with check_job_status and get_job_results as above. Extraction\n"
        "   results come back inline, wrapped with the source filename; report that\n"
        "   filename with each object rather than stripping the wrapper.\n"
        "Limits: files up to 50 MB, 10 files per request, 5 concurrent requests."
    ),
    tools=[transform_toolset, upload_local_file, wait_seconds],
)

adk run transform_agent

Parse this PDF with Unstructured Transform: https://arxiv.org/pdf/1706.03762
When it completes, tell me the document title and the first three section headings
from the parsed output.

→ start_transform_job({'file_refs': ['https://arxiv.org/pdf/1706.03762']})
→ check_job_status({'job_id': '1b359e00-d566-4ee4-b27d-cf2f408d898c'})
→ wait_seconds({'seconds': 30})
→ check_job_status({'job_id': '1b359e00-d566-4ee4-b27d-cf2f408d898c'})
→ get_job_results({'job_id': '1b359e00-d566-4ee4-b27d-cf2f408d898c', 'output_format': 'md'})
→ get_job_results({'job_id': '1b359e00-d566-4ee4-b27d-cf2f408d898c', 'output_format': 'json'})

The document has been successfully processed! Since the output files are stored
out of band, here are the download links for you to retrieve the full parsed
content: (pre-signed download links for the Markdown and JSON output)

Document Title: Attention Is All You Need

First Three Section Headings:
1. 1 Introduction
2. 2 Background
3. 3 Model Architecture

Extract each plant's name, light needs, watering instructions, and humidity level from
https://docs.unstructured.io/img/pipelines/data-extractor/house-plant-care.png as JSON.
Draft a schema from the image itself and show it to me first.

→ start_transform_job({'file_refs': ['https://docs.unstructured.io/img/...'], 'stages': {'partition': {'strategy': 'vlm'}}})
→ check_job_status({'job_id': '72015c9b-9a34-4f29-9fb1-3d7c4bfa872a'})
→ wait_seconds({'seconds': 30})
→ get_job_results({'job_id': '72015c9b-9a34-4f29-9fb1-3d7c4bfa872a'})
→ suggest_extraction_schema_for_file({'element_json_ref': 'u10d://output/72015c9b-..._house-plant-care.json'})
→ start_extraction_job({'element_json_refs': ['u10d://output/72015c9b-..._house-plant-care.json'], 'schema_to_extract': '{"type": "object", ...}'})
→ check_job_status({'job_id': '82e19e19-4a87-4dc6-a1fc-43cbaa6c1972'})
→ get_job_results({'job_id': '82e19e19-4a87-4dc6-a1fc-43cbaa6c1972'})