高级 MCP 配置

Supported in ADKPython v0.1.0

本指南介绍 ADK 中模型上下文协议 (MCP) 的高级集成模式。它提供了用于动态按用户身份验证、人工介入审批、长时间运行的进度跟踪、自定义运行时执行和企业级云部署的生产模式。

高级配置模式

使用下面的矩阵为你的生产工作负载选择合适的配置机制：

开发者需求	推荐机制	主要 API / 参数	典型场景
注入按用户的凭据或动态会话 token	动态标头提供器	header_provider=...	多租户应用、按用户的 JWT/OAuth token
在危险工具调用前要求审批	工具确认	require_confirmation=...	数据库变更、破坏性 shell/文件操作
为长时间操作流式传输实时进度	进度回调与工厂	progress_callback=...	重型 SQL 查询、网络抓取、数据索引
在 FastAPI / 后端服务中运行智能体而不使用 adk web	编程式运行器生命周期	Runner + await toolset.close()	自定义微服务、CLI 工具、工作队列
解决跨服务器的工具命名冲突	工具命名空间与过滤	tool_name_prefix、tool_filter	聚合多个 MCP 服务器（数据库 + GitHub）
处理服务器发起的采样或身份验证质询	双向协议回调	sampling_callback、elicitation_callback	服务器发起的 LLM 生成与身份验证提示
检查原始 STDERR 诊断流	诊断流日志	errlog=sys.stderr	排查 MCP 子进程崩溃
在聊天中渲染丰富的交互式可视化小部件	实验性 UI 渲染	meta.ui.resourceUri	地图、图表、天气卡片或自定义表单

---

动态身份验证与按用户标头（header_provider）

在多租户或面向用户的系统中，将凭据硬编码到连接参数中是不安全的。McpToolset 支持 header_provider，它是一个异步或同步的可调用对象，接收活动的 ReadonlyContext，以便在每次工具调用时动态构造身份验证标头。

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

async def extract_per_user_headers(context: ReadonlyContext) -> dict[str, str]:
    """在每一轮对话中动态提取会话状态或按用户的 token。"""
    user_token = context.state.get("user_access_token", "ANONYMOUS_TOKEN")
    return {
        "Authorization": f"Bearer {user_token}",
        "X-User-ID": context.user_id,
        "X-Session-ID": context.session.id,  # 已修正：访问 session.id
    }

toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://mcp-server.example.com/mcp",
        timeout=5,
        sse_read_timeout=300,
    ),
    header_provider=extract_per_user_headers,
)

---

人工介入与工具确认（require_confirmation）

MCP 服务器可以暴露高影响能力，例如：修改数据库架构或删除记录。你可以在工具集中对所有工具全局强制确认，也可以通过一个检查工具参数的谓词函数来有条件地强制确认。

from typing import Any
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

def should_require_approval(query: str = "", **kwargs) -> bool:
    query_str = str(query).lower()
    destructive_keywords = ["drop", "delete", "truncate", "alter", "update"]
    return any(keyword in query_str for keyword in destructive_keywords)

toolset = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/db"],
        ),
        timeout=5,
    ),
    require_confirmation=should_require_approval,  # 也可以是一个布尔值（True）
)

root_agent = LlmAgent(
    model="gemini-flash-latest",
    name="db_administrator",
    instruction="安全地执行数据库查询，对变更操作需明确批准。",
    tools=[toolset],
)

---

实时进度跟踪（progress_callback）

长时间运行的 MCP 操作（例如抓取大型网站或训练任务）会通过 notifications/progress 通道发送中间进度通知。

选项 A：全局回调函数

为简单的日志记录或进度上报指定一个共享回调：

async def on_mcp_progress(progress: float, total: float | None, message: str | None) -> None:
    percentage = (progress / total * 100) if total else progress
    print(f"[MCP Progress] {percentage:.1f}% complete: {message or 'Working...'}")

toolset = McpToolset(
    connection_params=...,
    progress_callback=on_mcp_progress,
)

选项 B：按工具的回调工厂（会话感知）

使用 ProgressCallbackFactory 注入具有 ToolContext.state 写入权限的工具专属回调：

from google.adk.tools.tool_context import ToolContext

def create_tool_progress_tracker(tool_name: str, callback_context: ToolContext, **kwargs):
    """生成自定义进度处理器，并更新活动的智能体会话状态。"""
    async def progress_handler(progress: float, total: float | None, message: str | None):
        callback_context.state[f"{tool_name}_status"] = message
        callback_context.state[f"{tool_name}_progress"] = progress
    return progress_handler

toolset = McpToolset(
    connection_params=...,
    progress_callback=create_tool_progress_tracker,
)

---

独立运行器执行（在 adk web 之外）

当把 ADK 智能体嵌入自定义 FastAPI 应用、后台工作进程或独立 CLI 脚本时，请实例化 Runner，并通过 await toolset.close() 显式管理生命周期的销毁。

import asyncio
import os
from google.genai import types
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

async def run_standalone_mcp_agent():
    # 1. 同步定义 McpToolset 和智能体
    toolset = McpToolset(
        connection_params=StdioConnectionParams(
            server_params=StdioServerParameters(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-filesystem", os.path.abspath("./data")],
            ),
            timeout=5,
        ),
        tool_filter=["list_directory", "read_file"],
    )

    agent = LlmAgent(
        model="gemini-flash-latest",
        name="filesystem_assistant",
        instruction="协助用户进行文件管理。",
        tools=[toolset],
    )

    # 2. 设置会话和运行器
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="standalone_mcp_app",
        user_id="user_001",
    )

    runner = Runner(
        app_name="standalone_mcp_app",
        agent=agent,
        session_service=session_service,
    )

    try:
        # 3. 流式传输智能体执行过程
        user_message = types.Content(
            role="user",
            parts=[types.Part(text="List the files available in the directory.")],
        )

        async for event in runner.run_async(
            session_id=session.id,
            user_id=session.user_id,
            new_message=user_message,
        ):
            if event.content and event.content.parts:
                for part in event.content.parts:
                    if part.text:
                        print(part.text, end="", flush=True)
    finally:
        # 4. 优雅地终止子进程和网络连接
        print("\nTerminating MCP connection...")
        await toolset.close()

if __name__ == "__main__":
    asyncio.run(run_standalone_mcp_agent())

---

名称冲突与工具命名空间（tool_name_prefix）

当连接到多个 MCP 服务器时，query 或 search 这类工具名称可能会冲突。使用 tool_name_prefix 自动为发现的工具添加命名空间：

from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

postgres_toolset = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/db"],
        )
    ),
    tool_name_prefix="pg_",  # 生成 pg_query、pg_list_tables
)

github_toolset = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-github"],
        )
    ),
    tool_name_prefix="gh_",  # 生成 gh_search_repositories、gh_create_issue
)

---

双向协议钩子：采样与征询

模型上下文协议支持双向交互，服务器可以向客户端请求执行操作：

• 采样（sampling_callback）：允许 MCP 服务器请求 ADK 宿主生成一次 LLM 补全。
• 征询（elicitation_callback）：允许 MCP 服务器请求带外用户交互或身份验证流程。

from mcp import SamplingCapability
from google.adk.tools.mcp_tool import McpToolset

async def handle_server_sampling(params):
    """处理服务器发起的 LLM 生成请求。"""
    return {
        "role": "assistant",
        "content": {"type": "text", "text": "Generated response from ADK"},
    }

async def handle_server_elicitation(params):
    """处理来自服务器的身份验证或交互质询。"""
    print(f"Elicitation requested: {params}")
    return {"action": "approved"}

toolset = McpToolset(
    connection_params=...,
    sampling_callback=handle_server_sampling,
    sampling_capabilities=SamplingCapability(),
    elicitation_callback=handle_server_elicitation,
)

---

诊断日志与错误流（errlog）

默认情况下，MCP 子进程的错误会记录到标准错误。你可以将 STDERR 流重定向到外部文件或诊断缓冲区，以便进行根因调试：

import sys
from google.adk.tools.mcp_tool import McpToolset

error_file = open("mcp_server_errors.log", "a")
try:
    toolset = McpToolset(connection_params=..., errlog=error_file)
    # 运行智能体……
finally:
    await toolset.close()
    error_file.close()

渲染交互式 UI 小部件

标准 MCP 工具返回纯文本或 JSON 输出。此功能让 MCP 工具能够直接在聊天界面内返回丰富的交互式可视化小部件，例如地图、图表或表单。

sequenceDiagram
    autonumber
    participant Tool as MCP 服务器工具
    participant ADK as ADK 框架
    participant UI as 客户端 UI（adk web / 前端）

    Tool-->>ADK: 返回结果 + 元数据（meta.ui.resourceUri = "ui://widgets/map"）
    ADK->>ADK: 检测 meta.ui.resourceUri 注解
    ADK-->>UI: 发出带有 UI 渲染信号和资源 URI 的事件
    UI->>Tool: 从资源 URI 获取 UI 包
    UI-->>UI: 在聊天界面中渲染交互式小部件

工作原理

1. 工具注册：MCP 工具在 tools/list 期间于其模式定义元数据中声明 UI 资源链接：meta.ui.resourceUri = "ui://widgets/weather-card"。
2. ADK 检测：ADK 读取模式定义以检测 meta.ui.resourceUri，从而知道该工具支持交互式 UI。
3. 客户端显示：工具执行后，ADK 会通知 Web UI（adk web 或自定义前端）获取该 UI 资源并渲染交互式小部件，而不是显示纯文本。

from mcp import types as mcp_types
from mcp.server.lowlevel import Server

app = Server("weather-mcp-server")


@app.list_tools()
async def list_mcp_tools() -> list[mcp_types.Tool]:
  """声明该工具并附加 UI 渲染元数据。"""
  return [
      mcp_types.Tool(
          name="get_weather",
          description="Get weather forecast for a city.",
          inputSchema={
              "type": "object",
              "properties": {"city": {"type": "string"}},
              "required": ["city"],
          },
          meta={"ui": {"resourceUri": "ui://widgets/weather-card"}},
      )
  ]


@app.call_tool()
async def call_mcp_tool(name: str, arguments: dict) -> list[mcp_types.Content]:
  """执行该工具并返回标准的文本/数据内容。"""
  if name == "get_weather":
    city = arguments.get("city", "Unknown")
    return [
        mcp_types.TextContent(
            type="text", text=f"Weather in {city}: 72°F Sunny"
        )
    ]
  return [
      mcp_types.TextContent(type="text", text=f"Unknown tool: '{name}'")
  ]

---

后续步骤

• 返回模型上下文协议概述了解基础设置。
• 探索自定义函数工具，了解进程内 Python 工具。
• 阅读 ADK 部署指南，了解完整的云配置选项。