MCP（模型上下文协议）工具集成¶
本指南将引导你了解将模型上下文协议 (MCP) 与 ADK 集成的两种方式。
什么是模型上下文协议 (MCP)？¶
模型上下文协议 (MCP) 是一种开放标准，旨在标准化大型语言模型 (LLM)（如 Gemini 和 Claude）与外部应用程序、数据源和工具的通信方式。可以将其视为一种通用连接机制，简化了 LLM 获取上下文、执行操作和与各种系统交互的方式。
MCP 遵循客户端 - 服务器架构，定义了数据（资源）、交互模板（提示）和可执行函数（工具）如何由 MCP 服务器公开并由 MCP 客户端（可能是 LLM 主机应用程序或 AI 智能体）使用。
本指南涵盖了两种主要的集成模式：
- 在 ADK 中使用现有 MCP 服务器： ADK 智能体充当 MCP 客户端，利用外部 MCP 服务器提供的工具。
- 通过 MCP 服务器暴露 ADK 工具： 构建一个包装 ADK 工具的 MCP 服务器，使其可被任何 MCP 客户端访问。
先决条件¶
在开始之前，确保你已设置以下内容：
- 设置 ADK： 按照快速入门中的标准 ADK 设置说明进行操作。
- 安装/更新 Python/Java： MCP 需要 Python 3.9 或更高版本或 Java 17+。
- 设置 Node.js 和 npx： （仅限 Python） 许多社区 MCP 服务器作为 Node.js 包分发并使用
npx运行。如果你尚未安装，请安装 Node.js（包括 npx）。详情请参阅 https://nodejs.org/en。 - 验证安装： （仅限 Python） 在激活的虚拟环境中确认
adk和npx在你的 PATH 中：
1. 在 adk web 中将 MCP 服务器与 ADK 智能体一起使用（ADK 作为 MCP 客户端）¶
本节演示如何将外部 MCP（模型上下文协议）服务器的工具集成到你的 ADK 智能体中。这是最常见的集成模式，当你的 ADK 智能体需要使用由现有服务通过 MCP 接口提供的能力时。你将看到如何将 MCPToolset 类直接添加到智能体的 tools 列表中，从而实现与 MCP 服务器的无缝连接、工具发现，并让你的智能体可以使用这些工具。这些示例主要聚焦于 adk web 开发环境中的交互。
MCPToolset 类¶
MCPToolset 类是 ADK 集成 MCP 服务器工具的主要机制。当你在智能体的 tools 列表中包含一个 MCPToolset 实例时，它会自动处理与指定 MCP 服务器的交互。其工作原理如下：
- 连接管理： 初始化时，
MCPToolset建立并管理与 MCP 服务器的连接。这可以是本地服务器进程（使用StdioServerParameters通过标准输入/输出通信）或远程服务器（使用SseServerParams进行 SSE 通信）。当智能体或应用终止时，工具集还会优雅地关闭此连接。 - 工具发现与适配： 连接后，
MCPToolset会查询 MCP 服务器的可用工具（通过list_toolsMCP 方法），并将这些工具的 schema 转换为 ADK 兼容的BaseTool实例。 - 暴露给智能体： 这些适配后的工具会像原生 ADK 工具一样提供给你的
LlmAgent。 - 代理工具调用： 当你的
LlmAgent决定使用其中一个工具时，MCPToolset会透明地将调用（通过call_toolMCP 方法）代理到 MCP 服务器，发送必要参数，并将服务器响应返回给智能体。 - 过滤（可选）： 创建
MCPToolset时可以使用tool_filter参数，仅选择 MCP 服务器中的特定工具子集暴露给智能体，而不是全部暴露。
以下示例演示了如何在 adk web 开发环境中使用 MCPToolset。如果你需要对 MCP 连接生命周期有更细致的控制，或不使用 adk web，请参阅本页后续"在 adk web 之外的自己的智能体中使用 MCP 工具"部分。
示例 1：文件系统 MCP 服务器¶
此示例演示了连接到提供文件系统操作的本地 MCP 服务器。
步骤 1：使用 MCPToolset 定义你的智能体¶
创建一个 agent.py 文件（如 ./adk_agent_samples/mcp_agent/agent.py）。MCPToolset 直接在你的 LlmAgent 的 tools 列表中实例化。
- 重要提示： 将
args列表中的"/path/to/your/folder"替换为 MCP 服务器可以访问的本地系统上实际文件夹的绝对路径。 - 重要提示： 将
.env文件放在./adk_agent_samples目录的父目录中。
# ./adk_agent_samples/mcp_agent/agent.py
import os # 用于路径操作
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
# 最好动态定义路径，或确保用户理解需要绝对路径。
# 本示例假设 '/path/to/your/folder' 与 agent.py 同目录。
# 如有需要请替换为实际绝对路径。
TARGET_FOLDER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "/path/to/your/folder")
# 确保 TARGET_FOLDER_PATH 是 MCP 服务器可访问的绝对路径。
# 如果你创建了 ./adk_agent_samples/mcp_agent/your_folder，
root_agent = LlmAgent(
model='gemini-2.0-flash',
name='filesystem_assistant_agent',
instruction='帮助用户管理他们的文件。你可以列出文件、读取文件等。',
tools=[
MCPToolset(
connection_params=StdioServerParameters(
command='npx',
args=[
"-y", # npx 自动确认安装
"@modelcontextprotocol/server-filesystem",
# 重要提示：此处必须为 MCP 服务器可访问的绝对路径。
# 例如："/Users/youruser/accessible_mcp_files"
# 或使用动态构造的绝对路径：
os.path.abspath(TARGET_FOLDER_PATH),
],
),
# 可选：仅暴露 MCP 服务器中的部分工具
# tool_filter=['list_directory', 'read_file']
)
],
)
步骤 2：创建 __init__.py 文件¶
确保与 agent.py 同目录下有 __init__.py，以便 ADK 能发现该 Python 包。
步骤 3：运行 adk web 并交互¶
在终端中切换到 mcp_agent 的父目录（如 adk_agent_samples），然后运行：
Windows 用户注意事项
当遇到 _make_subprocess_transport NotImplementedError 时，请考虑使用 adk web --no-reload 代替。
一旦 ADK Web UI 在你的浏览器中加载：
- 在智能体下拉菜单中选择
filesystem_assistant_agent。 - 尝试如下提示：
- "列出当前目录中的文件。"
- "你能读取名为 sample.txt 的文件吗？"（假设你在
TARGET_FOLDER_PATH创建了该文件） - "
another_file.md的内容是什么？"
你应该能看到智能体与 MCP 文件系统服务器交互，服务器的响应（文件列表、文件内容）通过智能体返回。adk web 控制台（你运行命令的终端）也可能显示 npx 进程的日志。
示例 2：Google Maps MCP 服务器¶
此示例演示了如何连接到 Google Maps MCP 服务器。
步骤 1：获取 API 密钥并启用 API¶
- Google Maps API Key： 按照 使用 API 密钥 获取 Google Maps API Key。
- 启用 API： 在你的 Google Cloud 项目中，确保已启用以下 API：
- Directions API
- Routes API 具体操作见 Google Maps Platform 入门 文档。
步骤 2：为 Google Maps 使用 MCPToolset 定义你的智能体¶
修改你的 agent.py 文件（如 ./adk_agent_samples/mcp_agent/agent.py）。将 YOUR_GOOGLE_MAPS_API_KEY 替换为你获取的实际 API key。
# ./adk_agent_samples/mcp_agent/agent.py
import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
# 从环境变量获取 API key，或直接插入。
# 推荐使用环境变量。
# 确保在运行 'adk web' 的终端设置了该环境变量。
# 例如：export GOOGLE_MAPS_API_KEY="YOUR_ACTUAL_KEY"
google_maps_api_key = os.environ.get("GOOGLE_MAPS_API_KEY")
if not google_maps_api_key:
# 测试用的回退方式 - 生产环境不推荐
google_maps_api_key = "YOUR_GOOGLE_MAPS_API_KEY_HERE" # 如未用环境变量请替换
if google_maps_api_key == "YOUR_GOOGLE_MAPS_API_KEY_HERE":
print("警告：GOOGLE_MAPS_API_KEY 未设置。请将其设置为环境变量或在脚本中设置。")
# 如密钥必需且未找到可报错或退出
root_agent = LlmAgent(
model='gemini-2.0-flash',
name='maps_assistant_agent',
instruction='使用 Google Maps 工具帮助用户进行地图、导航和地点查找。',
tools=[
MCPToolset(
connection_params=StdioServerParameters(
command='npx',
args=[
"-y",
"@modelcontextprotocol/server-google-maps",
],
# 以环境变量方式传递 API key 给 npx 进程
env={
"GOOGLE_MAPS_API_KEY": google_maps_api_key
}
),
# 如需可筛选特定 Maps 工具：
# tool_filter=['get_directions', 'find_place_by_id']
)
],
)
步骤 3：确保 __init__.py 存在¶
如在示例 1 已创建可跳过，否则确保 ./adk_agent_samples/mcp_agent/ 目录下有 __init__.py：
步骤 4：运行 adk web 并交互¶
-
设置环境变量（推荐）： 在运行
替换为你的实际 key。adk web前，最好在终端设置 Google Maps API key： -
运行
adk web： 切换到mcp_agent的父目录（如adk_agent_samples）并运行： -
在 UI 交互：
- 选择
maps_assistant_agent。 - 尝试如下提示：
- "获取从 GooglePlex 到 SFO 的路线。"
- "在金门公园附近找咖啡店。"
- "从法国巴黎到德国柏林的路线是什么？"
- 选择
你应该能看到智能体使用 Google Maps MCP 工具提供路线或地理信息。
2. 使用 ADK 工具构建 MCP 服务器（MCP 服务器暴露 ADK）¶
这种模式允许你包装现有 ADK 工具，并让任何标准 MCP 客户端应用都能访问它们。本节示例通过自定义 MCP 服务器暴露 ADK load_web_page 工具。
步骤概述¶
你将使用 mcp 库创建一个标准 Python MCP 服务器应用。在该服务器中：
- 实例化你要暴露的 ADK 工具（如
FunctionTool(load_web_page)）。 - 实现 MCP 服务器的
@app.list_tools()处理器，使用google.adk.tools.mcp_tool.conversion_utils的adk_to_mcp_tool_type工具将 ADK 工具定义转换为 MCP schema。 - 实现 MCP 服务器的
@app.call_tool()处理器：- 接收 MCP 客户端的工具调用请求。
- 判断请求是否针对你包装的 ADK 工具。
- 执行 ADK 工具的
.run_async()方法。 - 将 ADK 工具结果格式化为 MCP 兼容响应（如
mcp.types.TextContent）。
先决条件¶
在与 ADK 安装相同的 Python 环境中安装 MCP 服务器库：
步骤 1：创建 MCP 服务器脚本¶
为你的 MCP 服务器创建一个新的 Python 文件，如 my_adk_mcp_server.py。
步骤 2：实现服务器逻辑¶
将以下代码添加到 my_adk_mcp_server.py。该脚本设置了一个 MCP 服务器，暴露 ADK load_web_page 工具。
# my_adk_mcp_server.py
import asyncio
import json
import os
from dotenv import load_dotenv
# MCP 服务器导入
from mcp import types as mcp_types # 使用别名以避免冲突
from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio # 用于作为 stdio 服务器运行
# ADK 工具导入
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.load_web_page import load_web_page # 示例 ADK 工具
# ADK <-> MCP 转换工具
from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type
# --- 加载环境变量（如 ADK 工具需要，如 API key） ---
load_dotenv() # 如有需要，在同目录下创建 .env 文件
# --- 准备要暴露的 ADK 工具 ---
# 实例化你要暴露的 ADK 工具。
# 该工具将被 MCP 服务器包装并调用。
print("正在初始化 ADK load_web_page 工具...")
adk_tool_to_expose = FunctionTool(load_web_page)
print(f"ADK 工具 '{adk_tool_to_expose.name}' 已初始化并准备通过 MCP 暴露。")
# --- 结束 ADK 工具准备 ---
# --- MCP 服务器设置 ---
print("正在创建 MCP 服务器实例...")
# 使用 mcp.server 库创建命名的 MCP Server 实例
app = Server("adk-tool-exposing-mcp-server")
# 实现 MCP 服务器的 handler 以列出可用工具
@app.list_tools()
async def list_mcp_tools() -> list[mcp_types.Tool]:
"""MCP handler to list tools this server exposes."""
print("MCP 服务器：收到 list_tools 请求。")
# 将 ADK 工具定义转换为 MCP Tool schema 格式
mcp_tool_schema = adk_to_mcp_tool_type(adk_tool_to_expose)
print(f"MCP 服务器：广告工具：{mcp_tool_schema.name}")
return [mcp_tool_schema]
# 实现 MCP 服务器的 handler 以执行工具调用
@app.call_tool()
async def call_mcp_tool(
name: str, arguments: dict
) -> list[mcp_types.Content]: # MCP 使用 mcp_types.Content
"""MCP handler to execute a tool call requested by an MCP client."""
print(f"MCP 服务器：收到对 '{name}' 的 call_tool 请求，参数：{arguments}")
# 检查请求的工具名是否匹配我们包装的 ADK 工具
if name == adk_tool_to_expose.name:
try:
# 执行 ADK 工具的 run_async 方法。
# 注意：此处 tool_context 为 None，因为该 MCP 服务器
# 在完整 ADK Runner 调用之外运行 ADK 工具。
# 如果 ADK 工具需要 ToolContext 特性（如 state 或 auth），
# 这种直接调用可能需要更复杂的处理。
adk_tool_response = await adk_tool_to_expose.run_async(
args=arguments,
tool_context=None,
)
print(f"MCP 服务器：ADK 工具 '{name}' 已执行。响应：{adk_tool_response}")
# 将 ADK 工具的响应（通常为字典）格式化为 MCP 兼容格式。
# 这里将响应字典序列化为 JSON 字符串，放入 TextContent。
# 可根据 ADK 工具输出和客户端需求调整格式。
response_text = json.dumps(adk_tool_response, indent=2)
# MCP 期望返回 mcp_types.Content 部件的列表
return [mcp_types.TextContent(type="text", text=response_text)]
except Exception as e:
print(f"MCP 服务器：执行 ADK 工具 '{name}' 时出错：{e}")
# 以 MCP 格式返回错误信息
error_text = json.dumps({"error": f"执行工具 '{name}' 失败：{str(e)}"})
return [mcp_types.TextContent(type="text", text=error_text)]
else:
# 处理未知工具的调用
print(f"MCP 服务器：工具 '{name}' 未在此服务器中找到/暴露。")
error_text = json.dumps({"error": f"工具 '{name}' 未在此服务器中实现。"})
return [mcp_types.TextContent(type="text", text=error_text)]
# --- MCP 服务器运行器 ---
async def run_mcp_stdio_server():
"""以标准输入/输出监听连接，运行 MCP 服务器。"""
# 使用 mcp.server.stdio 库的 stdio_server 上下文管理器
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
print("MCP Stdio 服务器：开始与客户端握手...")
await app.run(
read_stream,
write_stream,
InitializationOptions(
server_name=app.name, # 使用上面定义的服务器名
server_version="0.1.0",
capabilities=app.get_capabilities(
# 定义服务器能力 - 参见 MCP 文档
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
print("MCP Stdio 服务器：运行循环完成或客户端断开连接。")
if __name__ == "__main__":
print("正在启动 MCP 服务器以通过 stdio 暴露 ADK 工具...")
try:
asyncio.run(run_mcp_stdio_server())
except KeyboardInterrupt:
print("\nMCP 服务器（stdio）被用户停止。")
except Exception as e:
print(f"MCP 服务器（stdio）遇到错误：{e}")
finally:
print("MCP 服务器（stdio）进程退出。")
# --- 结束 MCP 服务器 ---
步骤 3：使用 ADK 智能体测试你的自定义 MCP 服务器¶
现在，创建一个 ADK 智能体作为你刚刚构建的 MCP 服务器的客户端。该智能体将使用 MCPToolset 连接到你的 my_adk_mcp_server.py 脚本。
创建 agent.py（如 ./adk_agent_samples/mcp_client_agent/agent.py）：
# ./adk_agent_samples/mcp_client_agent/agent.py
import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import MCPToolset, StdioServerParameters
# 重要提示：请替换为你的 my_adk_mcp_server.py 脚本的绝对路径
PATH_TO_YOUR_MCP_SERVER_SCRIPT = "/path/to/your/my_adk_mcp_server.py" # <<< 替换
if PATH_TO_YOUR_MCP_SERVER_SCRIPT == "/path/to/your/my_adk_mcp_server.py":
print("警告：PATH_TO_YOUR_MCP_SERVER_SCRIPT 未设置。请在 agent.py 中更新它。")
# 如路径必需可报错
root_agent = LlmAgent(
model='gemini-2.0-flash',
name='web_reader_mcp_client_agent',
instruction="使用 'load_web_page' 工具从用户提供的 URL 获取内容。",
tools=[
MCPToolset(
connection_params=StdioServerParameters(
command='python3', # 运行你的 MCP 服务器脚本的命令
args=[PATH_TO_YOUR_MCP_SERVER_SCRIPT], # 参数为脚本路径
)
# tool_filter=['load_web_page'] # 可选：仅加载特定工具
)
],
)
同目录下创建 __init__.py：
运行测试：
-
启动自定义 MCP 服务器（可选，便于观察）： 你可以在一个终端直接运行
它会打印 "正在启动 MCP 服务器..." 并等待。ADK 智能体（通过my_adk_mcp_server.py以查看日志：adk web运行）会连接到该进程（如果StdioServerParameters的command设置为执行它）。 （或者，MCPToolset会在智能体初始化时自动以子进程方式启动该服务器脚本。） -
为客户端智能体运行
adk web： 切换到mcp_client_agent的父目录（如adk_agent_samples）并运行： -
在 ADK Web UI 交互：
- 选择
web_reader_mcp_client_agent。 - 尝试如 "加载 https://example.com 的内容" 的提示。
- 选择
ADK 智能体（web_reader_mcp_client_agent）会用 MCPToolset 启动并连接你的 my_adk_mcp_server.py。你的 MCP 服务器会收到 call_tool 请求，执行 ADK load_web_page 工具并返回结果。ADK 智能体会转发该信息。你应能在 ADK Web UI（及其终端）和 my_adk_mcp_server.py 终端（如单独运行）看到日志。
此示例演示了如何将 ADK 工具封装在 MCP 服务器中，使其可被更广泛的 MCP 客户端（不仅仅是 ADK 智能体）访问。
参见 文档 以尝试与 Claude Desktop 集成。
在 adk web 之外的自己的智能体中使用 MCP 工具¶
如果以下情况符合你的需求，本节内容与你相关：
- 你正在使用 ADK 开发自己的智能体
- 且你不使用
adk web - 且你通过自己的 UI 公开智能体
使用 MCP 工具需要与使用常规工具不同的设置，因为 MCP 工具的规范是从远程运行或在另一个进程中运行的 MCP 服务器异步获取的。
以下示例是从上面的"示例 1：文件系统 MCP 服务器"示例修改而来的。主要区别是：
- 你的工具和智能体是异步创建的
- 你需要正确管理退出栈，以便在与 MCP 服务器的连接关闭时正确销毁你的智能体和工具。
# agent.py（根据需要修改 get_tools_async 和其他部分）
# ./adk_agent_samples/mcp_agent/agent.py
import os
import asyncio
from dotenv import load_dotenv
from google.genai import types
from google.adk.agents.llm_agent import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService # 可选
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams, StdioServerParameters
# 从父目录的 .env 文件加载环境变量
# 在使用环境变量（如 API 密钥）之前，将此放在靠前的位置
load_dotenv('../.env')
# 确保 TARGET_FOLDER_PATH 是 MCP 服务器的绝对路径。
TARGET_FOLDER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "/path/to/your/folder")
# --- 步骤 1：智能体定义 ---
async def get_agent_async():
"""创建一个配备 MCP 服务器工具的 ADK 智能体。"""
toolset = MCPToolset(
# 使用 StdioServerParameters 进行本地进程通信
connection_params=StdioServerParameters(
command='npx', # 运行服务器的命令
args=["-y", # 命令的参数
"@modelcontextprotocol/server-filesystem",
TARGET_FOLDER_PATH],
),
tool_filter=['read_file', 'list_directory'] # 可选：过滤特定工具
# 对于远程服务器，你将使用 SseServerParams 代替：
# connection_params=SseServerParams(url="http://remote-server:port/path", headers={...})
)
# 在智能体中使用
root_agent = LlmAgent(
model='gemini-2.0-flash', # 根据可用性调整模型名称
name='enterprise_assistant',
instruction='帮助用户访问他们的文件系统',
tools=[toolset], # 为 ADK 智能体提供 MCP 工具
)
return root_agent, toolset
# --- 步骤 2：主执行逻辑 ---
async def async_main():
session_service = InMemorySessionService()
# 此示例可能不需要 Artifact 服务
artifacts_service = InMemoryArtifactService()
session = await session_service.create_session(
state={}, app_name='mcp_filesystem_app', user_id='user_fs'
)
# 提示：将查询更改为与你指定的文件夹相关的内容。
# 例如，"列出 'documents' 子文件夹中的文件"或"读取 'notes.txt' 文件"
query = "列出测试文件夹中的文件"
print(f"用户查询：'{query}'")
content = types.Content(role='user', parts=[types.Part(text=query)])
root_agent, toolset = await get_agent_async()
runner = Runner(
app_name='mcp_filesystem_app',
agent=root_agent,
artifact_service=artifacts_service, # 可选
session_service=session_service,
)
print("正在运行智能体...")
events_async = runner.run_async(
session_id=session.id, user_id=session.user_id, new_message=content
)
async for event in events_async:
print(f"收到事件：{event}")
# 清理由智能体框架自动处理
# 但如果需要，你也可以手动关闭：
print("正在关闭 MCP 服务器连接...")
await toolset.close()
print("清理完成。")
if __name__ == '__main__':
try:
asyncio.run(async_main())
except Exception as e:
print(f"发生错误：{e}")
关键考虑事项¶
使用 MCP 和 ADK 时，请记住以下几点：
-
协议与库： MCP 是一种协议规范，定义通信规则。ADK 是用于构建智能体的 Python 库/框架。MCPToolset 通过在 ADK 框架中实现 MCP 协议的客户端端来桥接这两者。相反，在 Python 中构建 MCP 服务器需要使用 model-context-protocol 库。
-
ADK 工具与 MCP 工具：
- ADK 工具（BaseTool, FunctionTool, AgentTool 等）是为在 ADK 的 LlmAgent 和 Runner 中直接使用而设计的 Python 对象。
- MCP 工具是根据协议的模式由 MCP 服务器公开的能力。MCPToolset 使这些对 LlmAgent 看起来像 ADK 工具。
- Langchain/CrewAI 工具是这些库中的特定实现，通常是简单的函数或类，缺乏 MCP 的服务器/协议结构。ADK 提供了一些互操作性的包装器（LangchainTool, CrewaiTool）。
-
异步特性： ADK 和 MCP Python 库都严重基于 Python 的 asyncio 库。工具实现和服务器处理程序通常应该是异步函数。
-
有状态会话（MCP）： MCP 在客户端和服务器实例之间建立有状态的持久连接。这与典型的无状态 REST API 不同。
- 部署： 这种有状态性可能会给扩展和部署带来挑战，特别是对于处理多个用户的远程服务器。原始 MCP 设计通常假定客户端和服务器位于同一位置。管理这些持久连接需要仔细的基础设施考虑（例如，负载均衡，会话亲和性）。
- ADK MCPToolset： 管理这种连接生命周期。示例中显示的 exit_stack 模式对于确保连接（可能还有服务器进程）在 ADK 智能体完成时正确终止至关重要。