MCP（模型上下文协议）工具集成¶
Supported in ADKPython v0.1.0Java v0.1.0
本指南将引导你了解将模型上下文协议 (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 服务器的连接。这可以是本地服务器进程（使用StdioConnectionParams通过标准输入/输出进行通信）或远程服务器（使用SseConnectionParams进行服务器发送事件）。工具集还在智能体或应用程序终止时处理此连接的优雅关闭。 - 工具发现和适配： 连接后，
MCPToolset查询 MCP 服务器的可用工具（通过list_toolsMCP 方法）。然后将这些发现的 MCP 工具的架构转换为 ADK 兼容的BaseTool实例。 - 暴露给智能体： 这些适配的工具然后像原生 ADK 工具一样提供给你的
LlmAgent。 - 代理工具调用： 当你的
LlmAgent决定使用这些工具之一时，MCPToolset透明地将调用代理（使用call_toolMCP 方法）到 MCP 服务器，发送必要的参数，并将服务器的响应返回给智能体。 - 过滤（可选）： 你可以在创建
MCPToolset时使用tool_filter参数来选择 MCP 服务器的特定工具子集，而不是将所有工具暴露给你的智能体。
以下示例演示了如何在 adk web 开发环境中使用 MCPToolset。如果你需要对 MCP 连接生命周期有更细致的控制，或不使用 adk web，请参阅本页后续"在 adk web 之外的自己的智能体中使用 MCP 工具"部分。
示例 1：文件系统 MCP 服务器¶
这个 Python 示例演示了连接到提供文件系统操作的本地 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
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import 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=StdioConnectionParams(
server_params = StdioServerParameters(
command='npx',
args=[
"-y", # npx 自动确认安装的参数
"@modelcontextprotocol/server-filesystem",
# 重要：这必须是 npx 进程可以访问的文件夹的绝对路径。
# 替换为你系统上的有效绝对路径。
# 例如："/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 服务器¶
对于 Java，请参考以下示例来定义初始化 MCPToolset 的智能体：
package agents;
import com.google.adk.JsonBaseModel;
import com.google.adk.agents.LlmAgent;
import com.google.adk.agents.RunConfig;
import com.google.adk.runner.InMemoryRunner;
import com.google.adk.tools.mcp.McpTool;
import com.google.adk.tools.mcp.McpToolset;
import com.google.adk.tools.mcp.McpToolset.McpToolsAndToolsetResult;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
import io.modelcontextprotocol.client.transport.ServerParameters;
import java.util.List;
import java.util.concurrent.CompletableFuture;
public class McpAgentCreator {
/**
* 初始化 McpToolset，使用 stdio 从 MCP 服务器检索工具，
* 创建带有这些工具的 LlmAgent，向智能体发送提示，
* 并确保工具集被关闭。
* @param args 命令行参数（未使用）。
*/
public static void main(String[] args) {
//注意：如果文件夹在 home 目录外，你可能会有权限问题
String yourFolderPath = "~/path/to/folder";
ServerParameters connectionParams = ServerParameters.builder("npx")
.args(List.of(
"-y",
"@modelcontextprotocol/server-filesystem",
yourFolderPath
))
.build();
try {
CompletableFuture<McpToolsAndToolsetResult> futureResult =
McpToolset.fromServer(connectionParams, JsonBaseModel.getMapper());
McpToolsAndToolsetResult result = futureResult.join();
try (McpToolset toolset = result.getToolset()) {
List<McpTool> tools = result.getTools();
LlmAgent agent = LlmAgent.builder()
.model("gemini-2.0-flash")
.name("enterprise_assistant")
.description("帮助用户访问其文件系统的智能体")
.instruction(
"帮助用户访问其文件系统。你可以列出目录中的文件。"
)
.tools(tools)
.build();
System.out.println("智能体已创建：" + agent.name());
InMemoryRunner runner = new InMemoryRunner(agent);
String userId = "user123";
String sessionId = "1234";
String promptText = "此目录中有哪些文件 - " + yourFolderPath + "?";
// 首先显式创建会话
try {
// appName for InMemoryRunner defaults to agent.name() if not specified in constructor
runner.sessionService().createSession(runner.appName(), userId, null, sessionId).blockingGet();
System.out.println("会话已创建：" + sessionId + " 用户：" + userId);
} catch (Exception sessionCreationException) {
System.err.println("创建会话失败：" + sessionCreationException.getMessage());
sessionCreationException.printStackTrace();
return;
}
Content promptContent = Content.fromParts(Part.fromText(promptText));
System.out.println("\n向智能体发送提示：\"" + promptText + "\"...\n");
runner.runAsync(userId, sessionId, promptContent, RunConfig.builder().build())
.blockingForEach(event -> {
System.out.println("收到事件：" + event.toJson());
});
}
} catch (Exception e) {
System.err.println("发生错误：" + e.getMessage());
e.printStackTrace();
}
}
}
假设一个包含三个名为 first、second 和 third 文件的文件夹，成功响应将如下所示：
收到事件: {"id":"163a449e-691a-48a2-9e38-8cadb6d1f136","invocationId":"e-c2458c56-e57a-45b2-97de-ae7292e505ef","author":"enterprise_assistant","content":{"parts":[{"functionCall":{"id":"adk-388b4ac2-d40e-4f6a-bda6-f051110c6498","args":{"path":"~/home-test"},"name":"list_directory"}}],"role":"model"},"actions":{"stateDelta":{},"artifactDelta":{},"requestedAuthConfigs":{}},"timestamp":1747377543788}
收到事件: {"id":"8728380b-bfad-4d14-8421-fa98d09364f1","invocationId":"e-c2458c56-e57a-45b2-97de-ae7292e505ef","author":"enterprise_assistant","content":{"parts":[{"functionResponse":{"id":"adk-388b4ac2-d40e-4f6a-bda6-f051110c6498","name":"list_directory","response":{"text_output":[{"text":"[FILE] first\n[FILE] second\n[FILE] third"}]}}}],"role":"user"},"actions":{"stateDelta":{},"artifactDelta":{},"requestedAuthConfigs":{}},"timestamp":1747377544679}
收到事件: {"id":"8fe7e594-3e47-4254-8b57-9106ad8463cb","invocationId":"e-c2458c56-e57a-45b2-97de-ae7292e505ef","author":"enterprise_assistant","content":{"parts":[{"text":"目录中有三个文件：first、second 和 third。"}],"role":"model"},"actions":{"stateDelta":{},"artifactDelta":{},"requestedAuthConfigs":{}},"timestamp":1747377544689}
示例 2：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
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import 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=StdioConnectionParams(
server_params = StdioServerParameters(
command='npx',
args=[
"-y",
"@modelcontextprotocol/server-google-maps",
],
# 将 API key 作为环境变量传递给 npx 进程
# 这是 Google Maps 的 MCP 服务器期望密钥的方式。
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 并交互¶
-
设置环境变量（推荐）： 在运行
adk web前，最好在终端设置 Google Maps API key：替换为你的实际 key。
-
运行
adk web： 切换到mcp_agent的父目录（如adk_agent_samples）并运行： -
在 UI 交互：
- 选择
maps_assistant_agent。 - 尝试如下提示：
- "获取从 GooglePlex 到 SFO 的路线。"
- "在金门公园附近找咖啡店。"
- "从法国巴黎到德国柏林的路线是什么？"
- 选择
你应该能看到智能体使用 Google Maps MCP 工具提供路线或地理信息。
对于 Java，请参考以下示例来定义初始化 MCPToolset 的智能体：
package agents;
import com.google.adk.JsonBaseModel;
import com.google.adk.agents.LlmAgent;
import com.google.adk.agents.RunConfig;
import com.google.adk.runner.InMemoryRunner;
import com.google.adk.tools.mcp.McpTool;
import com.google.adk.tools.mcp.McpToolset;
import com.google.adk.tools.mcp.McpToolset.McpToolsAndToolsetResult;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
import io.modelcontextprotocol.client.transport.ServerParameters;
import java.util.List;
import java.util.Map;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.CompletableFuture;
import java.util.Arrays;
public class MapsAgentCreator {
/**
* 为 Google Maps 初始化 McpToolset，检索工具，
* 创建 LlmAgent，发送地图相关提示，并关闭工具集。
* @param args 命令行参数（未使用）。
*/
public static void main(String[] args) {
// TODO: 替换为你的实际 Google Maps API key，在启用了 Places API 的项目上。
String googleMapsApiKey = "YOUR_GOOGLE_MAPS_API_KEY";
Map<String, String> envVariables = new HashMap<>();
envVariables.put("GOOGLE_MAPS_API_KEY", googleMapsApiKey);
ServerParameters connectionParams = ServerParameters.builder("npx")
.args(List.of(
"-y",
"@modelcontextprotocol/server-google-maps"
))
.env(Collections.unmodifiableMap(envVariables))
.build();
try {
CompletableFuture<McpToolsAndToolsetResult> futureResult =
McpToolset.fromServer(connectionParams, JsonBaseModel.getMapper());
McpToolsAndToolsetResult result = futureResult.join();
try (McpToolset toolset = result.getToolset()) {
List<McpTool> tools = result.getTools();
LlmAgent agent = LlmAgent.builder()
.model("gemini-2.0-flash")
.name("maps_assistant")
.description("地图助手")
.instruction("使用可用工具帮助用户进行地图和导航。")
.tools(tools)
.build();
System.out.println("智能体已创建：" + agent.name());
InMemoryRunner runner = new InMemoryRunner(agent);
String userId = "maps-user-" + System.currentTimeMillis();
String sessionId = "maps-session-" + System.currentTimeMillis();
String promptText = "请给我到麦迪逊广场花园最近药店的路线。";
try {
runner.sessionService().createSession(runner.appName(), userId, null, sessionId).blockingGet();
System.out.println("会话已创建：" + sessionId + " 用户：" + userId);
} catch (Exception sessionCreationException) {
System.err.println("创建会话失败：" + sessionCreationException.getMessage());
sessionCreationException.printStackTrace();
return;
}
Content promptContent = Content.fromParts(Part.fromText(promptText));
System.out.println("\n向智能体发送提示：\"" + promptText + "\"...\n");
runner.runAsync(userId, sessionId, promptContent, RunConfig.builder().build())
.blockingForEach(event -> {
System.out.println("收到事件：" + event.toJson());
});
}
} catch (Exception e) {
System.err.println("发生错误：" + e.getMessage());
e.printStackTrace();
}
}
}
成功响应将如下所示：
收到事件: {"id":"1a4deb46-c496-4158-bd41-72702c773368","invocationId":"e-48994aa0-531c-47be-8c57-65215c3e0319","author":"maps_assistant","content":{"parts":[{"text":"好的。我看到几个选项。最近的是位于纽约州纽约市 10001 宾夕法尼亚广场 5 号的 CVS 药店。你想要路线吗？\n"}],"role":"model"},"actions":{"stateDelta":{},"artifactDelta":{},"requestedAuthConfigs":{}},"timestamp":1747380026642}
2. 使用 ADK 工具构建 MCP 服务器（MCP 服务器暴露 ADK）¶
步骤概述¶
你将使用 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.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import 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=StdioConnectionParams(
server_params = StdioServerParameters(
command='python3', # 运行你的 MCP 服务器脚本的命令
args=[PATH_TO_YOUR_MCP_SERVER_SCRIPT], # 参数是脚本的路径
)
)
# tool_filter=['load_web_page'] # 可选：仅加载特定工具
)
],
)
同目录下创建 __init__.py：
运行测试：
-
启动自定义 MCP 服务器（可选，便于观察）： 你可以在一个终端直接运行
my_adk_mcp_server.py以查看日志：它将打印"正在启动 MCP 服务器..."并等待。如果
StdioConnectionParams中的command设置为执行它，ADK 智能体（通过adk web运行）将连接到这个过程。 （或者，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 # Optional
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import 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(
# 使用 StdioConnectionParams 进行本地进程通信
connection_params=StdioConnectionParams(
server_params = StdioServerParameters(
command='npx', # 运行服务器的命令
args=["-y", # 命令的参数
"@modelcontextprotocol/server-filesystem",
TARGET_FOLDER_PATH],
),
),
tool_filter=['read_file', 'list_directory'] # 可选：过滤特定工具
# 对于远程服务器，你会使用 SseConnectionParams：
# connection_params=SseConnectionParams(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 Tools (BaseTool, FunctionTool, AgentTool, etc.) are Python objects designed for direct use within the ADK's LlmAgent and Runner.
- MCP Tools are capabilities exposed by an MCP Server according to the protocol's schema. MCPToolset makes these look like ADK tools to an LlmAgent.
-
有状态会话（MCP）： MCP 在客户端和服务器实例之间建立有状态的持久连接。这与典型的无状态 REST API 不同。
-
部署： 这种有状态性可能会给扩展和部署带来挑战，特别是对于处理多个用户的远程服务器。原始 MCP 设计通常假定客户端和服务器位于同一位置。管理这些持久连接需要仔细的基础设施考虑（例如，负载均衡，会话亲和性）。
- ADK MCPToolset： 管理这种连接生命周期。示例中显示的 exit_stack 模式对于确保连接（可能还有服务器进程）在 ADK 智能体完成时正确终止至关重要。
进一步资源¶
使用 MCP 工具部署智能体¶
当将使用 MCP 工具的 ADK 智能体部署到生产环境（如 Cloud Run、GKE 或 Vertex AI Agent Engine）时，你需要考虑 MCP 连接在容器化和分布式环境中如何工作。
关键部署要求：同步智能体定义¶
⚠️ 重要： 当部署带有 MCP 工具的智能体时，智能体及其 MCPToolset 必须在你的 agent.py 文件中同步定义。虽然 adk web 允许异步智能体创建，但部署环境需要同步实例化。
# ✅ 正确：用于部署的同步智能体定义
import os
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool import StdioConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from mcp import StdioServerParameters
_allowed_path = os.path.dirname(os.path.abspath(__file__))
root_agent = LlmAgent(
model='gemini-2.0-flash',
name='enterprise_assistant',
instruction=f'帮助用户访问其文件系统。允许的目录：{_allowed_path}',
tools=[
MCPToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command='npx',
args=['-y', '@modelcontextprotocol/server-filesystem', _allowed_path],
),
),
# 在生产环境中过滤工具以确保安全
tool_filter=[
'read_file', 'read_multiple_files', 'list_directory',
'directory_tree', 'search_files', 'get_file_info',
'list_allowed_directories',
],
)
],
)
# ❌ 错误：异步模式在部署中不起作用
async def get_agent(): # 这对部署不起作用
toolset = await create_mcp_toolset_async()
return LlmAgent(tools=[toolset])
快速部署命令¶
Vertex AI Agent Engine¶
uv run adk deploy agent_engine \
--project=<your-gcp-project-id> \
--region=<your-gcp-region> \
--staging_bucket="gs://<your-gcs-bucket>" \
--display_name="My MCP Agent" \
./path/to/your/agent_directory
Cloud Run¶
uv run adk deploy cloud_run \
--project=<your-gcp-project-id> \
--region=<your-gcp-region> \
--service_name=<your-service-name> \
./path/to/your/agent_directory
部署模式¶
模式 1：自包含的 Stdio MCP 服务器¶
对于可以打包为 npm 包或 Python 模块的 MCP 服务器（如 @modelcontextprotocol/server-filesystem），你可以直接将它们包含在智能体容器中：
容器要求：
# npm 基础 MCP 服务器示例
FROM python:3.13-slim
# 为 MCP 服务器安装 Node.js 和 npm
RUN apt-get update && apt-get install -y nodejs npm && rm -rf /var/lib/apt/lists/*
# 安装你的 Python 依赖
COPY requirements.txt .
RUN pip install -r requirements.txt
# 复制你的智能体代码
COPY . .
# 你的智能体现在可以使用带有 'npx' 命令的 StdioConnectionParams
CMD ["python", "main.py"]
智能体配置：
# 这在容器中有效，因为 npx 和 MCP 服务器在同一环境中运行
MCPToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command='npx',
args=["-y", "@modelcontextprotocol/server-filesystem", "/app/data"],
),
),
)
模式 2：远程 MCP 服务器（可流式 HTTP）¶
对于需要可扩展性的生产部署，将 MCP 服务器部署为单独的服务并通过可流式 HTTP 连接：
MCP 服务器部署（Cloud Run）：
# deploy_mcp_server.py - 使用可流式 HTTP 的单独 Cloud Run 服务
import contextlib
import logging
from collections.abc import AsyncIterator
from typing import Any
import anyio
import click
import mcp.types as types
from mcp.server.lowlevel import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.types import Receive, Scope, Send
logger = logging.getLogger(__name__)
def create_mcp_server():
"""创建和配置 MCP 服务器。"""
app = Server("adk-mcp-streamable-server")
@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentBlock]:
"""处理来自 MCP 客户端的工具调用。"""
# 示例工具实现 - 替换为你的实际 ADK 工具
if name == "example_tool":
result = arguments.get("input", "No input provided")
return [
types.TextContent(
type="text",
text=f"Processed: {result}"
)
]
else:
raise ValueError(f"Unknown tool: {name}")
@app.list_tools()
async def list_tools() -> list[types.Tool]:
"""列出可用工具。"""
return [
types.Tool(
name="example_tool",
description="Example tool for demonstration",
inputSchema={
"type": "object",
"properties": {
"input": {
"type": "string",
"description": "Input text to process"
}
},
"required": ["input"]
}
)
]
return app
def main(port: int = 8080, json_response: bool = False):
"""主服务器函数。"""
logging.basicConfig(level=logging.INFO)
app = create_mcp_server()
# Create session manager with stateless mode for scalability
session_manager = StreamableHTTPSessionManager(
app=app,
event_store=None,
json_response=json_response,
stateless=True, # 对 Cloud Run 可扩展性很重要
)
async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:
await session_manager.handle_request(scope, receive, send)
@contextlib.asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
"""管理会话管理器生命周期。"""
async with session_manager.run():
logger.info("MCP Streamable HTTP server started!")
try:
yield
finally:
logger.info("MCP server shutting down...")
# 创建 ASGI 应用程序
starlette_app = Starlette(
debug=False, # 生产环境设置为 False
routes=[
Mount("/mcp", app=handle_streamable_http),
],
lifespan=lifespan,
)
import uvicorn
uvicorn.run(starlette_app, host="0.0.0.0", port=port)
if __name__ == "__main__":
main()
远程 MCP 的智能体配置：
# 你的 ADK 智能体通过可流式 HTTP 连接到远程 MCP 服务
MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://your-mcp-server-url.run.app/mcp",
headers={"Authorization": "Bearer your-auth-token"}
),
)
模式 3：Sidecar MCP 服务器（GKE）¶
在 Kubernetes 环境中，你可以将 MCP 服务器部署为 sidecar 容器：
# deployment.yaml - 带有 MCP sidecar 的 GKE
apiVersion: apps/v1
kind: Deployment
metadata:
name: adk-agent-with-mcp
spec:
template:
spec:
containers:
# 主 ADK 智能体容器
- name: adk-agent
image: your-adk-agent:latest
ports:
- containerPort: 8080
env:
- name: MCP_SERVER_URL
value: "http://localhost:8081"
# MCP 服务器 sidecar
- name: mcp-server
image: your-mcp-server:latest
ports:
- containerPort: 8081
连接管理考虑事项¶
Stdio 连接¶
SSE/HTTP Connections¶
- Pros: Network-based, scalable, can handle multiple clients
- Cons: Requires network infrastructure, authentication complexity
-
Best for: Production deployments, multi-tenant systems, external MCP services
-
缺点： 进程开销，不适合高规模部署
- 最适合： 开发、单租户部署、简单的 MCP 服务器
SSE/HTTP 连接¶
-
优点： 基于网络，可扩展，可以处理多个客户端
-
缺点： 需要网络基础设施，认证复杂性
- 最适合： 生产部署、多租户系统、外部 MCP 服务
生产部署清单¶
当将带有 MCP 工具的智能体部署到生产环境时：
✅ 连接生命周期
- 使用 exit_stack 模式确保 MCP 连接的正确清理
- 为连接建立和请求配置适当的超时
- 为瞬态连接失败实现重试逻辑
✅ 资源管理
- 监控 stdio MCP 服务器的内存使用情况（每个都会产生一个进程）
- 为 MCP 服务器进程配置适当的 CPU/内存限制
- 考虑远程 MCP 服务器的连接池
✅ 安全
- 为远程 MCP 连接使用认证头
- 限制 ADK 智能体和 MCP 服务器之间的网络访问
- 使用
tool_filter过滤 MCP 工具以限制暴露的功能 - 验证 MCP 工具输入以防止注入攻击
- 为文件系统 MCP 服务器使用限制性文件路径（例如，
os.path.dirname(os.path.abspath(__file__))） - 考虑为生产环境使用只读工具过滤器
✅ 监控和可观测性
- 记录 MCP 连接建立和拆除事件
- 监控 MCP 工具执行时间和成功率
- 为 MCP 连接失败设置警报
✅ 可扩展性
- 对于高容量部署，优先选择远程 MCP 服务器而不是 stdio
- 如果使用有状态 MCP 服务器，配置会话亲和性
- 考虑 MCP 服务器连接限制并实现断路器
环境特定配置¶
Cloud Run¶
# Cloud Run 环境的 MCP 配置环境变量
import os
# 检测 Cloud Run 环境
if os.getenv('K_SERVICE'):
# 在 Cloud Run 中使用远程 MCP 服务器
mcp_connection = SseConnectionParams(
url=os.getenv('MCP_SERVER_URL'),
headers={'Authorization': f"Bearer {os.getenv('MCP_AUTH_TOKEN')}"}
)
else:
# 本地开发使用 stdio
mcp_connection = StdioConnectionParams(
server_params=StdioServerParameters(
command='npx',
args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
)
)
MCPToolset(connection_params=mcp_connection)
GKE¶
# GKE 特定的 MCP 配置
# 使用集群内 MCP 服务器的服务发现
MCPToolset(
connection_params=SseConnectionParams(
url="http://mcp-service.default.svc.cluster.local:8080/sse"
),
)
Vertex AI Agent Engine¶
# Agent Engine 托管部署
# 优先选择轻量级、自包含的 MCP 服务器或外部服务
MCPToolset(
connection_params=SseConnectionParams(
url="https://your-managed-mcp-service.googleapis.com/sse",
headers={'Authorization': 'Bearer $(gcloud auth print-access-token)'}
),
)
故障排除部署问题¶
常见的 MCP 部署问题：
- Stdio 进程启动失败
# 调试 stdio 连接问题
MCPToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command='npx',
args=["-y", "@modelcontextprotocol/server-filesystem", "/app/data"],
# 添加环境调试
env={'DEBUG': '1'}
),
),
)
- 网络连接问题
# 测试远程 MCP 连接
import aiohttp
async def test_mcp_connection():
async with aiohttp.ClientSession() as session:
async with session.get('https://your-mcp-server.com/health') as resp:
print(f"MCP Server Health: {resp.status}")
- 资源耗尽
- 使用 stdio MCP 服务器时监控容器内存使用情况
- 在 Kubernetes 部署中设置适当的限制
- 对资源密集型操作使用远程 MCP 服务器