> ## Documentation Index
> Fetch the complete documentation index at: https://docs.unstructured.io/llms.txt
> Use this file to discover all available pages before exploring further.

# 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)](https://google.github.io/adk-docs/) 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](https://www.python.org/downloads/).
* An Unstructured API key, which the Transform MCP server uses as a bearer token. [Get an API key](/transform/code#get-your-unstructured-api-key-and-url).
* A Gemini API key for the agent's underlying model. [Get an API key](https://aistudio.google.com/apikey).

## Install the packages

In your terminal, install ADK with the `mcp` extra:

```bash theme={null}
pip install "google-adk[mcp]"
```

<Warning>
  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.
</Warning>

<Note>
  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](https://github.com/google/adk-python/releases) for changes.
</Note>

## Connect to the Transform MCP server

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

```bash theme={null}
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:

```python theme={null}
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",
    ],
)
```

<Warning>
  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.
</Warning>

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`.

<Note>
  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.
</Note>

## Build the agent

Create an agent package with the standard ADK layout:

```text theme={null}
transform_agent/
├── __init__.py
└── agent.py
```

In `__init__.py`:

```python theme={null}
from . import agent
```

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

```python theme={null}
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],
)
```

<Warning>
  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.
</Warning>

<Note>
  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](https://ai.google.dev/gemini-api/docs/models).
</Note>

## Run the agent

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

```bash theme={null}
adk run transform_agent
```

Then ask it to parse a file:

```text theme={null}
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:

```text theme={null}
→ 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
```

<Note>
  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.
</Note>

## Next steps

* [Control Transform file parsing output](/transform/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](/transform/sde): Control how the Unstructured Transform MCP server extracts and formats structured data from your files.
* [Control Transform generated sample code](/transform/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](/support/request).