Skip to content

面向 ADK 的 Unstructured Transform MCP 工具

在 ADK 中受支持Python

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

使用场景

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

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

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

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

前置条件

安装

安装带有 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=(
        "你使用 Unstructured Transform MCP 服务器解析文档。"
        "将公开的 https:// 文件 URL 直接传递给 transform_files。"
        "它会返回一个 job_id；使用 check_transform_status 进行轮询，"
        "每次检查之间调用 wait_seconds(30)（作业需要 30 秒到几分钟）。"
        "作业完成后，调用 get_transform_results 并将解析内容报告给用户。"
        "transform_files 接受可选的 stages 配置；默认情况下会自动选择解析策略，"
        "但如果输出质量较低（文本乱码或表格丢失），请使用 hi_res 分区策略重新运行文件"
        "以获得更清晰的结果。如果要求解析本地文件，请说明这需要 Unstructured ADK 指南中的"
        "上传辅助函数。"
    ),
    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",
                "transform_files",
                "check_transform_status",
                "get_transform_results",
            ],
        )
    ],
)

Note

文档转换是异步的：transform_files 启动一个作业，智能体轮询 check_transform_status，而 get_transform_results 返回输出的预签名下载 URL。如上所示，指示你的智能体在状态检查之间暂停，这样轮询循环不会耗尽模型的速率限制。

要解析本地文件，智能体还需要一个普通函数工具，该工具通过 HTTP PUT 将文件字节发送到 request_file_upload_url 返回的预签名 URL（此上传不是 MCP 调用，且不得发送 Authorization 头）。一个包含上传和等待辅助函数的完整智能体示例，请参见 Unstructured Transform ADK 指南。

可用工具

工具 描述
request_file_upload_url 返回本地文件的预签名上传 URL 和文件引用。
transform_files 为已上传的文件或公开 HTTP(S) URL 启动解析作业；返回 job_id。
check_transform_status 报告作业状态为 SCHEDULED（已计划）、IN_PROGRESS（进行中）或 COMPLETED（已完成）。
get_transform_results 返回已完成作业的解析输出和预签名下载 URL。

资源