部署带 MCP 工具的智能体¶
将使用 MCP 工具的 ADK 智能体部署到 Cloud Run、GKE 或 Agent Runtime 等生产环境时，你需要考虑 MCP 连接在容器化和分布式环境中如何工作。
关键部署要求：同步定义智能体¶
Warning
部署带 MCP 工具的智能体时，智能体及其 McpToolset 必须在你的 agent.py 文件中同步定义。虽然 adk web 允许异步创建智能体，但部署环境要求同步实例化。
# 正确：用于部署的同步智能体定义
import os
from google.adk.agents.llm_agent 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
_allowed_path = os.path.dirname(os.path.abspath(__file__))
root_agent = LlmAgent(
model='gemini-flash-latest',
name='enterprise_assistant',
instruction=f'帮助用户访问其文件系统。允许的目录：{_allowed_path}',
tools=[
McpToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command='npx',
args=['-y', '@modelcontextprotocol/server-filesystem', _allowed_path],
),
timeout=5, # 配置合适的超时时间
),
# 在生产环境中过滤工具以确保安全
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])
快速部署命令¶
Agent Runtime¶
uv run adk deploy agent_engine \
--project=<your-gcp-project-id> \
--region=<your-gcp-region> \
--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 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()
# 创建无状态模式的会话管理器以实现可扩展性
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 的智能体配置：
import java.util.Map;
import com.google.adk.tools.mcp.StreamableHttpServerParameters;
import com.google.adk.tools.mcp.McpToolset;
// 你的 ADK 智能体通过可流式传输的 HTTP 连接到远程 MCP 服务
StreamableHttpServerParameters streamableParams = StreamableHttpServerParameters.builder()
.url("https://your-mcp-server-url.run.app/mcp")
.headers(Map.of("Authorization", "Bearer your-auth-token"))
.build();
McpToolset toolset = new McpToolset(streamableParams);
import com.google.adk.kt.tools.mcp.McpConnectionParameters
import com.google.adk.kt.tools.mcp.McpToolset
// 你的 ADK 智能体通过可流式传输的 HTTP 连接到远程 MCP 服务
// headerProvider 是 suspend 的，所以 fetchToken() 可以在每个请求等待一个新 token；
// 它还会禁用会话复用，因此请使用 StreamableHttp(headers = ...) 来固定一个 token。
val toolset =
McpToolset.McpToolsetConfig(
streamableHttpConnectionParams =
McpConnectionParameters.StreamableHttp(
url = "https://your-mcp-server-url.run.app/mcp",
),
).toToolset(headerProvider = { mapOf("Authorization" to "Bearer ${fetchToken()}") })
模式 3：Sidecar MCP 服务器（GKE）¶
在 Kubernetes 环境中，你可以把 MCP 服务器部署为边车 (Sidecar) 容器：
# deployment.yaml - 带 MCP 边车的 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 服务器边车
- name: mcp-server
image: your-mcp-server:latest
ports:
- containerPort: 8081
连接管理注意事项¶
根据你的扩展性和基础设施需求选择连接类型。
Stdio 连接¶
- 优点： 设置简单、进程隔离，在容器中表现良好。
- 缺点： 有进程开销；不适合大规模部署。
- 最适合： 开发、单租户部署和简单的 MCP 服务器。
SSE/HTTP 连接¶
- 优点： 基于网络、可扩展，能处理多个客户端。
- 缺点： 需要网络基础设施，且身份验证较复杂。
- 最适合： 生产部署、多租户系统、外部 MCP 服务和大流量场景。
生产部署指南¶
将带 MCP 工具的智能体部署到生产环境时，请遵循以下核心指南。
连接生命周期¶
- 使用标准的退出栈模式正确清理 MCP 连接。
- 为连接建立和请求配置合适的超时时间。
- 实现重试逻辑，以优雅地处理瞬时连接故障。
资源管理¶
- 密切监控 Stdio MCP 服务器的内存使用情况，因为每个连接都会生成一个新进程。
- 为 MCP 服务器进程配置合适的 CPU 和内存限制。
- 为远程 MCP 服务器实现连接池，以优化资源消耗。
安全性¶
安全最佳实践
- 为所有远程 MCP 连接使用严格的身份验证标头。
- 严格限制 ADK 智能体与 MCP 服务器之间的网络访问。
- 使用
tool_filter过滤 MCP 工具，严格限制暴露的功能。 - 验证所有 MCP 工具输入，防止提示词注入或命令注入攻击。
- 为文件系统 MCP 服务器使用严格限制的绝对文件路径（例如
os.path.dirname(os.path.abspath(__file__))）。 - 在生产环境中尽可能使用只读工具过滤器。
监控与可观测性¶
- 记录所有 MCP 连接建立和销毁事件。
- 监控 MCP 工具的执行时间和总体成功率。
- 为反复出现的 MCP 连接故障设置自动告警。