Unstructured Transform MCP server installation for Google ADK

Learn how to connect the Unstructured Transform MCP server to a Google Agent Development Kit (ADK) agent. 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.

Google Agent Development Kit (ADK) is Google’s open-source framework for building AI agents. Unlike end-user AI tools that add the Transform MCP server through a settings screen, ADK connects to it from agent code: you declare the Transform MCP server as a toolset and hand its tools to an agent. Because Transform is a hosted remote MCP server, there is nothing to install or run locally. You point the toolset at the server URL and authenticate with your Unstructured API key.

ADK ships two building blocks that make this work out of the box, so you do not need a custom transport:

• The McpToolset class, which connects to a remote MCP server and exposes its tools to an agent.
• The StreamableHTTPConnectionParams connection settings, which carry your Unstructured API key as an Authorization: Bearer header on every request, including the initial initialize handshake.

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.
• A Gemini API key for the agent’s underlying model. Get an API key.

Install the packages

In your terminal, install ADK with the mcp extra:

pip install "google-adk[mcp]"

The [mcp] extra is required. A plain pip install google-adk does not install the MCP client dependency, and ADK then hides its MCP classes silently: from google.adk.tools.mcp_tool import McpToolset fails with an ImportError even though ADK itself installed correctly.

ADK is in active development and its APIs change between releases. This guide was verified against version 2.4.0. If you install a different version, check the ADK release notes for changes.

Connect to the Transform MCP server

Store your API keys in environment variables rather than hard-coding them:

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

In your agent code, declare the Transform MCP server as an McpToolset. The headers setting attaches the Authorization: Bearer header to every request, so the handshake and the tool calls are both authenticated:

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",
        "check_job_status",
        "get_job_results",
    ],
)

Use the server URL exactly as shown. Do not add /mcp or /sse to the end of the URL.

Also, set timeout explicitly. ADK’s default of 5 seconds is too short for a remote server handshake and causes intermittent connection failures.

The tools in the tool_filter list are the tools the transform job lifecycle uses. The filter exposes only these tools to the agent and ignores any others the server advertises.

Understand the transform job lifecycle

Transforming a document is an asynchronous, multi-step flow. Four of the steps are MCP tool calls, and two are plain HTTP transfers that are not MCP calls:

1. request_file_upload_url: Returns a pre-signed upload_url and a file_ref for a local file.
2. An HTTP PUT of the raw file bytes to the pre-signed upload_url. This is not an MCP call.
3. start_transform_job: Starts a transform job for one or more file_ref values (or public HTTP(S) URLs) and returns a job_id.
4. check_job_status: Reports whether the job is SCHEDULED, IN_PROGRESS, or COMPLETED.
5. get_job_results: Returns a pre-signed download_url for the Markdown output of each transformed file.
6. An HTTP GET of the Markdown from the pre-signed download_url. This is not an MCP call.

An agent cannot perform the two raw HTTP transfers through MCP tools alone, and an unpaced polling loop wastes model calls. Give the agent two small helper functions alongside the toolset: one that performs the PUT upload, and one that waits between status checks. Both helpers must be async: ADK runs tools on its event loop, so a helper that blocks (for example, with time.sleep or a synchronous HTTP call) freezes every other session the agent is serving, such as concurrent sessions under adk web.

The pre-signed upload_url and download_url carry their own credentials in the URL itself. The PUT and GET must not send the Authorization header, or the storage service rejects them.

Build the agent

Create an agent package with the standard ADK layout:

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

In __init__.py:

from . import agent

In agent.py (the httpx library it imports is installed with ADK):

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",
        "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"
        "Limits: files up to 50 MB, 10 files per request, 5 concurrent requests."
    ),
    tools=[transform_toolset, upload_local_file, wait_seconds],
)

The upload_local_file tool is a trust boundary: the agent chooses its arguments, so a maliciously crafted document or prompt could try to direct it at sensitive files or attacker-controlled URLs. The guards in the example restrict it to files inside the current working directory and to https:// destinations. Keep those guards, run the agent in a working directory that contains only the files you intend to upload, and treat any loosening of the checks as a deliberate security decision.

On the Gemini API free tier, some models allow as few as 5 requests per minute, and an agent spends one model request per step of the workflow. The wait_seconds(30) pacing in the instruction keeps a polling loop from hitting that limit; if you still see 429 RESOURCE_EXHAUSTED errors, increase the wait or switch to a model with a higher quota. Also, some older model IDs are not available to newly created API keys. If a model returns a 404 NOT_FOUND error, choose a current one from the Gemini model list.

Run the agent

From the directory that contains the transform_agent package, start the agent:

adk run transform_agent

Then ask it to parse a file:

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.

The agent starts the transform job, polls at a measured pace, fetches the results, and answers. The following abbreviated trace shows the tool calls and final answer from a successful run; the live adk run console prints the same activity with more verbose formatting:

→ 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

If you do not use Google Cloud, ADK prints a warning at startup that it “Failed to configure mTLS” because default credentials were not found. The warning is harmless: the Transform MCP server authenticates with your Unstructured API key, not with Google Cloud credentials.

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.