Skip to content

面向 ADK 的 Unstructured Transform MCP 工具

在 ADK 中受支持Python

Unstructured Transform MCP Server 将你的 ADK 智能体连接到 Unstructured——一个将原始文件转换为结构化、AI 就绪数据的文档处理平台。该集成使你的智能体能够使用自然语言解析 PDF、Office 文档、电子邮件、图像和扫描文件（共支持 40 多种文件格式），并输出经过分区、富化、分块和嵌入的结果。Transform 是托管的远程 MCP 服务器，无需在本地安装或运行任何内容。

使用场景

  • RAG 数据摄取：将异构文档集合解析为干净的、分块的、可用于嵌入的输出，供向量存储和检索流水线使用。

  • 文档问答智能体：让智能体按需获取并解析合同、报告或论文，然后基于解析内容回答问题。

  • 格式标准化：将混合输入（扫描 PDF、电子表格、演示文稿、邮件线程）转换为一致的结构化表示。

  • 智能体运行时 OCR：在更大的智能体工作流中，从图像和扫描文档中提取文本和结构。

  • Structured data extraction: Pull named fields out of forms, invoices, and contracts as JSON matching a schema, either one you supply or one the server drafts from the document.

Prerequisites

安装

安装带有 mcp 扩展的 ADK。该扩展是必需的；没有它，ADK 的 MCP 类将无法导入：

pip install "google-adk[mcp]"

与智能体一起使用

将你的 API 密钥设置为环境变量：

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

服务器在每个请求（包括初始握手）中使用你的 Unstructured API 密钥作为 Bearer 令牌进行身份验证。wait_seconds 辅助函数让智能体在状态检查之间暂停，因为解析作业是异步运行的：

import asyncio
import os

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


async def wait_seconds(seconds: int) -> dict:
    """在下一次状态检查前暂停。除非另有说明，否则使用 30 秒。

    Args:
        seconds: 等待时长。

    Returns:
        dict 确认等待完成。
    """
    seconds = max(1, min(int(seconds), 120))
    await asyncio.sleep(seconds)
    return {"waited_seconds": seconds}


root_agent = Agent(
    model="gemini-flash-latest",
    name="transform_agent",
    instruction=(
        "You parse documents with the Unstructured Transform MCP server. "
        "Pass public https:// file URLs straight to start_transform_job. It "
        "returns a job_id; poll with check_job_status, calling "
        "wait_seconds(30) between checks (jobs take 30 seconds to a few "
        "minutes). When the job completes, call get_job_results and "
        "report the parsed content back to the user. start_transform_job "
        "accepts an optional stages config; it auto-selects a parse "
        "strategy by default, but if the output looks low quality "
        "(garbled text or lost tables), re-run the file with a hi_res "
        "partition strategy for a cleaner result. If the user wants "
        "specific fields rather than the whole document, extract "
        "instead of just parsing. The extraction tools read the element "
        "JSON a parse produces, so parse the file first and keep the "
        "output_ref that get_job_results returns for it. Call "
        "suggest_extraction_schema_for_file with that output_ref when "
        "you need a schema, then start_extraction_job with "
        "element_json_refs set to the output_refs and schema_to_extract "
        "set to a JSON Schema passed as a JSON string. Poll and read an "
        "extraction job with check_job_status and get_job_results like "
        "any other job; its results come back inline, wrapped with the "
        "source filename, so report that filename with each object. If "
        "asked to parse a local file, explain that this requires the "
        "upload helper from the Unstructured ADK guide."
    ),
    tools=[
        wait_seconds,
        McpToolset(
            connection_params=StreamableHTTPConnectionParams(
                url="https://mcp.transform.unstructured.io",  # 根 URL；不要追加 /mcp
                headers={
                    "Authorization": f"Bearer {os.environ['UNSTRUCTURED_API_KEY']}",
                },
                timeout=30.0,  # ADK 默认的 5 秒对于远程握手来说太短
                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",
            ],
        )
    ],
)

Note

Transforming a document is asynchronous: start_transform_job starts a job, the agent polls check_job_status, and get_job_results returns pre-signed download URLs for the output. Instruct your agent to pause between status checks, as shown above, so a polling loop does not burn through model rate limits.

Structured-data extraction is a second asynchronous job that runs on the element JSON of a completed parse, identified by the output_ref that get_job_results returns for each file. A prompt that parses and then extracts therefore runs two polling loops, so allow for the extra time and model steps.

To parse local files, the agent also needs a plain function tool that HTTP PUTs the file bytes to the pre-signed URL returned by request_file_upload_url (this upload is not an MCP call, and it must not send the Authorization header). A complete agent with the upload and wait helpers is in the Unstructured Transform ADK guide.

可用工具

工具 描述
request_file_upload_url Returns a pre-signed upload URL and file reference for a local file.
start_transform_job Starts a parsing job for uploaded files or public HTTP(S) URLs; returns a job_id.
suggest_extraction_schema_for_file Drafts a JSON Schema from one parsed document's element JSON, for when you do not have a schema yet.
start_extraction_job Starts a structured-data extraction job over parsed element JSON against a JSON Schema; returns a job_id.
check_job_status Reports whether a job is SCHEDULED, IN_PROGRESS, or COMPLETED. Serves both parsing and extraction jobs.
get_job_results Returns a completed job's output: pre-signed download URLs for a parsing job, or the extracted data inline for an extraction job.

资源