模型上下文协议 (MCP) 工具¶
模型上下文协议 (MCP) 是一项用于将生成式 AI 模型连接到外部数据源、工具和系统的开放标准。可以把它理解为一种通用连接机制，简化了 LLM 获取上下文、执行操作以及与各类系统交互的方式。
核心架构与概念¶
MCP 遵循客户端 - 服务器架构，定义了数据或资源、交互式模板或提示词，以及可执行的函数或工具如何由 MCP 服务器暴露，并由 MCP 客户端消费；这里的客户端可以是 LLM 宿主应用或 AI 智能体。在 ADK 中，你使用 McpToolset 类作为 MCP 服务器与 ADK 智能体之间的接口。你也可以将 ADK 服务器配置为 MCP 服务器，供其他客户端系统使用。
sequenceDiagram
autonumber
participant Agent as ADK LlmAgent（客户端）
participant Toolset as McpToolset
participant Server as MCP 服务器
Agent->>Toolset: 初始化连接
Toolset->>Server: 协议握手与工具发现 (list_tools)
Server-->>Toolset: 可用工具模式
Toolset-->>Agent: 适配后的 ADK 工具
Agent->>Toolset: 调用工具（参数）
Toolset->>Server: 执行工具（通过 Stdio/HTTP 调用 call_tool）
Server-->>Toolset: 执行结果（文本/JSON）
Toolset-->>Agent: 将结果返回给 LLM前提条件与设置规则¶
在开始之前，请确认已完成以下设置：
- 已安装 ADK：在你的项目环境中完成标准 ADK 设置，并安装 MCP 扩展：
pip install "google-adk[mcp]"。 - 运行时要求：Python 3.10+ 或 Java 17+。
- Node.js 与
npx（仅 Python/TS）：运行以 npm 打包的社区 MCP 服务器时需要。 - 验证安装：在已激活的虚拟环境中，确认
adk和npx已在你的 PATH 中：
部署规则
部署到生产环境的智能体必须在 agent.py 中同步定义 McpToolset。动态异步智能体初始化仅支持用于本地调试或自定义独立运行器。
理解用途与集成方式¶
主要有三种集成模式。直接集成在本页介绍，其他实现方式则在其各自的页面中介绍。
- 直接 MCP 工具集成：ADK 智能体使用
McpToolset作为 MCP 客户端。 - 智能体暴露的 MCP 服务器：你使用
to_mcp_server构建一个封装 ADK 工具的 MCP 服务器。 - 专业化子智能体委派：智能体使用
AgentTool将任务委派给子智能体。
MCP 实现选项¶
当你开始使用模型上下文协议 (MCP) 和 ADK 进行构建时，以下这些关键的架构差异将帮助你设计出更稳定、更高效的智能体。下表作为一份对比指南，帮助你构建这些智能体。
| 维度 | 直接 MCP 工具集成 (McpToolset) |
智能体暴露的 MCP 服务器 (to_mcp_server) |
专业化子智能体委派 (AgentTool) |
|---|---|---|---|
| 架构 | 外部服务器进程或远程服务提供确定性端点，并被适配到主 LlmAgent 的工具列表中。 |
一个自治的 ADK 智能体被编译成 MCP 服务器，可由外部客户端（Claude Code、IDE、外部宿主）调用。 | 进程内的分层智能体封装，父智能体将子 LlmAgent 作为可调用的工具来调用。 |
| 上下文窗口影响 | 高上下文膨胀：每个工具定义和原始输出（例如：数据库行或文件二进制块）都会进入主智能体的历史记录。 | 隔离：外部调用方只会收到最终聚合后的响应文本/块。 | 零上下文膨胀：中间的探索性推理、失败的工具调用和大量原始输出都隔离在子智能体循环中。 |
| AI 模型负载与分层 | 单一模型必须同时理解所有工具模式、验证约束和工作流状态。 | 独立的模型推理，专门用于被封装的任务。 | 支持模型分层，例如：用 gemini-2.5-pro 作为编排器，用 gemini-2.5-flash 配合专用系统指令执行子智能体工具。 |
| 延迟与 token 成本 | 成本更低、延迟可预测：1 次 LLM 轮次 + 1 次确定性工具调用 + 1 次响应生成轮次。 | 由客户端驱动；延迟取决于内部智能体的执行深度。 | 成本更高、延迟波动：在返回父智能体之前有多次 LLM 调用和子智能体推理轮次。 |
| 理想用例 |
|
|
|
状态恢复
虽然 ADK 智能体在生命周期事件期间会保留会话状态，但恢复时不会自动重新建立活动的 MCP 连接。智能体会根据需要重新初始化连接。
直接 MCP 工具集成 (McpToolset)¶
McpToolset 类可以直接添加到你的智能体工具列表中；这个类能让你无缝连接到 MCP 服务器、发现其工具，并让这些工具可供你的智能体使用。在初始化时，McpToolset 会建立并管理与 MCP 服务器的连接。当智能体或进程终止时，它还会处理连接的优雅关闭。
使用 McpToolset 将外部 MCP 服务器中的工具导入到你的 ADK LlmAgent 中。
示例：本地 Stdio 传输（FileSystem MCP）¶
本示例搭建了一个连接到本地 MCP 文件系统服务器的 ADK 智能体；它在智能体的工具列表中直接实例化 McpToolset，以启用文件管理能力。
步骤 1. 使用 McpToolset 定义你的智能体：
import os
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
TARGET_FOLDER = os.path.abspath("./accessible_files")
root_agent = LlmAgent(
model="gemini-flash-latest",
name="filesystem_assistant",
instruction="帮助用户管理本地文件。",
tools=[
McpToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command="npx",
args=["-y", "@modelcontextprotocol/server-filesystem", TARGET_FOLDER],
),
),
# 可选：选择暴露给智能体的特定工具
tool_filter=["list_directory", "read_file"],
)
],
)
步骤 2：打包并运行你的智能体，让 ADK 能够发现你的智能体并开始与之交互，请遵循以下工作流：
- 初始化你的包：在与你的 agent.py 相同的目录下创建一个
__init__.py文件。这一步是 ADK 识别你的智能体所必需的。 -
启动 Web 界面：
-
与智能体交互：从下拉菜单中选择
filesystem_assistant，并向智能体发出提示：列出当前目录中的文件 或 another_file.md 的内容是什么？
```
import { LlmAgent, MCPToolset } from "@google/adk";
import path from "path";
const TARGET_FOLDER = path.resolve("./accessible_files");
export const rootAgent = new LlmAgent({
model: "gemini-flash-latest",
name: "filesystem_assistant",
instruction: "Help users manage local files.",
tools: [
new MCPToolset({
type: "StdioConnectionParams",
serverParams: {
command: "npx",
args: ["-y", "@modelcontextprotocol/server-filesystem", TARGET_FOLDER],
},
}, ["list_directory", "read_file"]) // 可选的工具过滤数组
],
});
package agents;
import com.google.adk.agents.LlmAgent;
import com.google.adk.tools.mcp.McpToolset;
import com.google.adk.tools.mcp.StdioServerParameters;
import java.util.List;
public class FileSystemAgentCreator {
public static void main(String[] args) throws Exception {
StdioServerParameters serverParams = StdioServerParameters.builder()
.command("npx")
.args(List.of("-y", "@modelcontextprotocol/server-filesystem", "/absolute/path/to/folder"))
.build();
try (McpToolset toolset = new McpToolset(serverParams.toServerParameters())) {
LlmAgent agent = LlmAgent.builder()
.model("gemini-flash-latest")
.name("filesystem_assistant")
.instruction("Help users access their file systems.")
.tools(toolset)
.build();
System.out.println("Agent initialized: " + agent.name());
}
}
}
package main
import (
"context"
"fmt"
"os/exec"
"github.com/modelcontextprotocol/go-sdk/mcp"
"google.golang.org/adk/v2/agent"
"google.golang.org/adk/v2/agent/llmagent"
"google.golang.org/adk/v2/model/gemini"
"google.golang.org/adk/v2/tool"
"google.golang.org/adk/v2/tool/mcptoolset"
)
func createFilesystemAgent(ctx context.Context) (agent.Agent, error) {
// 1. 使用 CommandTransport 和 AllowedToolsPredicate 初始化 MCP 工具集
mcpTools, err := mcptoolset.New(mcptoolset.Config{
Transport: &mcp.CommandTransport{
Command: exec.Command("npx", "-y", "@modelcontextprotocol/server-filesystem", "./accessible_files"),
},
ToolFilter: tool.AllowedToolsPredicate([]string{"list_directory", "read_file"}),
})
if err != nil {
return nil, fmt.Errorf("failed to create mcp toolset: %w", err)
}
// 2. 初始化 Gemini 模型的 model.LLM 实例
llm, err := gemini.NewModel(ctx, "gemini-2.0-flash", nil)
if err != nil {
return nil, fmt.Errorf("failed to create gemini model: %w", err)
}
// 3. 创建返回 agent.Agent 的 LLM 智能体
return llmagent.New(llmagent.Config{
Name: "filesystem_assistant",
Model: llm,
Instruction: "帮助用户管理本地文件。",
Toolsets: []tool.Toolset{mcpTools},
})
}
func main() {
ctx := context.Background()
ag, err := createFilesystemAgent(ctx)
if err != nil {
panic(err)
}
fmt.Printf("Successfully created agent: %s\n", ag.Name())
}
示例：远程 HTTP / SSE 传输（Google Maps Grounding Lite）¶
开始之前，请按照 Google Maps Grounding Lite 的说明在你的 Google Cloud 项目上启用该服务，并生成你的 Maps Platform API 密钥。 与前面的本地进程示例不同，此模式使用服务器发送事件 (SSE) 将你的智能体连接到远程的、云托管的 MCP 服务器。它使用 Google Maps Grounding Lite 服务来演示如何将身份验证标头（例如 API 密钥）传递给可扩展的端点。
步骤 1：使用 McpToolset 定义你的智能体。
import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
API_KEY = os.getenv("GOOGLE_MAPS_API_KEY")
root_agent = LlmAgent(
model="gemini-flash-latest",
name="travel_planner",
instruction="使用 Google Maps 规划旅行路线并搜索地点。",
tools=[
McpToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://mapstools.googleapis.com/mcp",
headers={
"X-Goog-Api-Key": API_KEY,
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
},
timeout=5,
sse_read_timeout=300
)
)
],
)
adk web 之前设置环境变量，在你的终端中设置你的 Google API 密钥
步骤 3：运行 adk web：导航到 mcp_agent 的父目录并启动 Web 界面。
步骤 4：与 UI 交互：
- 从下拉菜单中选择 travel_planner。
- 尝试诸如以下提示：我明天会到旧金山。天气怎么样 或 找一下金门公园附近的咖啡店
```typescript import { LlmAgent, MCPToolset } from "@google/adk";
export const rootAgent = new LlmAgent({ model: "gemini-flash-latest", name: "travel_planner", instruction: "Plan travel routes and search locations using Google Maps.", tools: [ new MCPToolset({ type: "StreamableHTTPConnectionParams", url: "https://mapstools.googleapis.com/mcp", transportOptions: { requestInit: { headers: { "X-Goog-Api-Key": process.env.GOOGLE_MAPS_API_KEY!, "Content-Type": "application/json", "Accept": "application/json, text/event-stream", }, }, }, timeout: 5, sseReadTimeout: 300, }), ], }); ```
远程 MCP 身份验证与资源访问¶
本节向你展示如何使用身份验证连接到远程 MCP 服务器，以及如何读取 MCP 服务器暴露的数据资源。当 MCP 服务器需要身份验证时（例如通过服务器发送事件 SseConnectionParams 或可流式传输的 HTTP），McpToolset 会自动处理凭据注入和 token 管理。
关键身份验证参数¶
| 参数 | 类型 | 描述 |
|---|---|---|
auth_scheme |
AuthScheme |
身份验证策略（例如 Bearer、Basic、APIKey、OAuth2）。 |
auth_credential |
AuthCredential |
机密凭据负载，例如 API token、OAuth 访问 token、用户名/密码。 |
ADK 会自动构建所需的 Authorization HTTP 标头，并在客户端请求期间管理 OAuth 2.0 token 刷新。
配置身份验证¶
当 MCP 服务器需要身份验证时，McpToolset 会自动处理凭据注入和 token 管理。请使用原生的 auth_scheme 和 auth_credential 参数，而不是手动注入 HTTP 标头。
关于通用的 ADK 身份验证模式，请参阅我们的自定义工具身份验证指南
import os
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import SseConnectionParams
# 通过标头配置 Bearer Token 身份验证
toolset = McpToolset(
connection_params=SseConnectionParams(
url="https://mcp-server.example.com/sse",
headers={"Authorization": f"Bearer {os.getenv('MCP_AUTH_TOKEN')}"},
timeout=5,
)
)
import { MCPToolset } from "@google/adk";
// 通过标头配置 Bearer Token 身份验证
const toolset = new MCPToolset({
type: "StreamableHTTPConnectionParams",
url: "https://mcp-server.example.com/sse",
transportOptions: {
requestInit: {
headers: {
"Authorization": `Bearer ${process.env.MCP_AUTH_TOKEN}`,
},
},
},
timeout: 5,
});
访问 MCP 资源¶
除了可执行的工具之外，MCP 服务器还可以暴露资源数据文件、数据库记录或 API 上下文二进制块。
McpToolset 提供了两个核心方法来发现和读取这些数据资源：
核心方法¶
list_resources()：返回 MCP 服务器暴露的所有可用数据资源列表。read_resource(name)：按名称或 URI 检索特定资源的原始内容块（文本或二进制数据）。use_mcp_resources=True（配置）：在初始化McpToolset时，设置此标志会自动为 LLM 智能体配备LoadMcpResourceTool。这使智能体能够自行动态发现和读取资源，而无需手动以编程方式获取。
试一试¶
import asyncio
async def fetch_mcp_data(toolset):
# 1. 发现服务器上可用的资源
resources = await toolset.list_resources()
print("Available Resources:", resources)
# 2. 从特定资源读取内容
if resources:
resource_name = resources[0]
content_blocks = await toolset.read_resource(name=resource_name)
print(f"Content of {resource_name}:", content_blocks)
async function fetchMcpData(toolset: any) {
// 1. 发现可用资源
const resources = await toolset.listResources();
console.log("Available Resources:", resources);
// 2. 读取特定资源
if (resources.length > 0) {
const content = await toolset.readResource(resources[0]);
console.log(`Content of ${resources[0]}:`, content);
}
}
故障排查与最佳实践清单¶
- 安全与作用域：始终在
McpToolset中提供tool_filter=[...]，只向你的 LLM 暴露必要的操作。 - 超时：在
StdioConnectionParams(timeout=5)上配置显式超时，防止子进程挂起。 - 生命周期清理：在非
adk web运行器中，调用await toolset.close()或使用异步上下文管理器来优雅地关闭子进程。 - 环境检测：根据环境变量动态选择连接类型，例如：用
K_SERVICE区分 Cloud Run 与本地开发的 Stdio。
配置环境感知连接¶
import os
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import (
StdioConnectionParams,
StreamableHTTPConnectionParams,
)
from mcp import StdioServerParameters
if os.getenv("K_SERVICE"):
# 在生产环境（Cloud Run）中运行
# 使用可流式传输的 HTTP 实现无状态可扩展性，并使用 bearer 认证标头
mcp_toolset = McpToolset(
connection_params=StreamableHTTPConnectionParams(
url=os.getenv("REMOTE_MCP_URL"),
headers={"Authorization": f"Bearer {os.getenv('MCP_AUTH_TOKEN')}"},
timeout=5,
sse_read_timeout=300,
)
)
else:
# 在本地开发环境中运行
# 使用 Stdio 子进程 IPC 实现零网络延迟测试
mcp_toolset = McpToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command="npx",
args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
),
timeout=5,
)
)
延伸资源¶
掌握了基础知识之后，可以探索高级用例，了解复杂的实现和自定义集成。