ADK-Python MCP协议集成：模型上下文协议工具开发
·
ADK-Python MCP协议集成：模型上下文协议工具开发
项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python 引言：AI Agent开发的新范式
在当今AI Agent开发领域，开发者面临着一个核心挑战：如何让AI系统安全、高效地访问外部工具和数据源？传统方法往往需要为每个工具编写繁琐的集成代码，这不仅增加了开发复杂度，还限制了Agent的灵活性和扩展性。
Model Context Protocol（MCP，模型上下文协议）的出现彻底改变了这一局面。作为AI开发领域的重要标准，MCP提供了一种统一的方式来连接AI模型与外部工具和数据源。而Google的ADK-Python框架通过深度集成MCP协议，为开发者提供了构建下一代智能Agent的强大能力。
本文将深入探讨ADK-Python中MCP协议的集成机制，通过实际代码示例和架构分析，帮助你掌握这一革命性技术的核心原理和实践应用。
MCP协议核心概念解析
什么是MCP协议？
MCP（Model Context Protocol）是一个开放标准协议，旨在标准化AI模型与外部工具和数据源之间的通信。它解决了AI开发中的几个关键问题：
- 工具发现与注册：自动发现可用工具并注册到AI系统
- 统一接口：为不同类型的工具提供标准化的调用接口
- 安全访问：提供细粒度的访问控制和权限管理
- 实时通信：支持双向流式通信和实时数据交换
MCP在ADK-Python中的架构位置
ADK-Python MCP集成实战
环境准备与依赖安装
首先确保你的Python环境满足要求：
# 安装ADK核心包
pip install google-adk
# 安装MCP相关依赖（ADK会自动处理）
pip install mcp
基础MCP工具集配置
ADK-Python提供了多种MCP连接方式，下面是最常用的Stdio连接示例：
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
import os
# 设置允许访问的目录
_allowed_path = os.path.dirname(os.path.abspath(__file__))
# 创建MCP工具集
mcp_toolset = 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',
],
)
# 创建Agent并集成MCP工具
root_agent = LlmAgent(
model='gemini-2.0-flash',
name='enterprise_assistant',
instruction=f"""\
帮助用户访问文件系统。
允许访问的目录: {_allowed_path}
请谨慎操作，确保只访问授权范围内的文件。
""",
tools=[mcp_toolset]
)
配置文件方式集成MCP
ADK-Python支持通过YAML配置文件定义MCP工具，这种方式更加简洁：
# root_agent.yaml
name: notion_agent
model: gemini-2.0-flash
instruction: |
你是我的工作空间助手。使用提供的工具来读取、搜索、评论或创建
Notion页面。当不确定时要询问澄清问题。
tools:
- name: MCPToolset
args:
stdio_server_params:
command: "npx"
args:
- "-y"
- "@notionhq/notion-mcp-server"
env:
OPENAPI_MCP_HEADERS: '{"Authorization": "Bearer your_notion_token", "Notion-Version": "2022-06-28"}'
MCP连接模式详解
1. Stdio连接模式（本地进程）
Stdio模式是最常用的连接方式，适合本地运行的MCP服务器：
from google.adk.tools.mcp_tool import StdioConnectionParams
from mcp import StdioServerParameters
stdio_params = StdioConnectionParams(
server_params=StdioServerParameters(
command='python3',
args=['-m', 'my_custom_mcp_server', '--port', '8080'],
),
timeout=10
)
2. SSE连接模式（服务器端事件）
SSE模式适合需要长连接的远程MCP服务器：
from google.adk.tools.mcp_tool import SseConnectionParams
sse_params = SseConnectionParams(
url='https://api.example.com/mcp',
headers={'Authorization': 'Bearer token123'}
)
3. HTTP流式连接模式
HTTP流式连接提供更好的兼容性和稳定性：
from google.adk.tools.mcp_tool import StreamableHTTPConnectionParams
http_params = StreamableHTTPConnectionParams(
url='https://api.example.com/mcp/stream',
method='POST',
headers={'Content-Type': 'application/json'}
)
高级MCP工具管理
工具过滤与权限控制
ADK-Python提供了灵活的工具过滤机制，确保Agent只能访问授权的功能：
# 方法1：列表过滤
tool_filter=['read_file', 'search_files', 'list_directory']
# 方法2：Lambda函数过滤
tool_filter=lambda tool, ctx=None: tool.name not in [
'write_file',
'edit_file',
'create_directory',
'delete_file'
]
# 方法3：基于上下文的动态过滤
def dynamic_tool_filter(tool, context=None):
if context and context.get('user_role') == 'admin':
return True # 管理员可以访问所有工具
else:
return tool.name in ['read_file', 'search_files']
认证与安全配置
MCP工具支持多种认证方式：
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
# API密钥认证
mcp_toolset = MCPToolset(
connection_params=...,
auth_scheme='api_key',
auth_credential='your_api_key_here'
)
# OAuth2认证
mcp_toolset = MCPToolset(
connection_params=...,
auth_scheme='oauth2',
auth_credential={
'client_id': 'your_client_id',
'client_secret': 'your_client_secret',
'refresh_token': 'your_refresh_token'
}
)
实战案例：构建企业级文件管理Agent
场景需求分析
假设我们需要构建一个企业文件管理Agent，具备以下能力：
- 安全浏览指定目录的文件结构
- 搜索特定内容的文件
- 读取文件内容（仅文本文件）
- 获取文件元信息
- 严格的权限控制
完整实现代码
import os
from typing import List, Optional
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
class EnterpriseFileManager:
def __init__(self, base_path: str, allowed_extensions: Optional[List[str]] = None):
self.base_path = os.path.abspath(base_path)
self.allowed_extensions = allowed_extensions or ['.txt', '.md', '.py', '.json']
# 验证路径安全性
if not self._is_path_safe(self.base_path):
raise ValueError("指定的路径不在安全范围内")
def _is_path_safe(self, path: str) -> bool:
"""确保路径在允许的范围内"""
allowed_base = os.path.abspath('/safe/directory')
return path.startswith(allowed_base)
def create_mcp_toolset(self) -> MCPToolset:
"""创建配置好的MCP工具集"""
return MCPToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command='npx',
args=[
'-y',
'@modelcontextprotocol/server-filesystem',
self.base_path,
],
),
timeout=8,
),
tool_filter=self._get_tool_filter(),
auth_scheme='api_key',
auth_credential=os.getenv('MCP_API_KEY')
)
def _get_tool_filter(self):
"""根据文件类型动态过滤工具"""
return [
'list_directory',
'directory_tree',
'search_files',
'get_file_info',
'list_allowed_directories',
'read_file' # 谨慎使用
]
def create_agent(self) -> LlmAgent:
"""创建文件管理Agent"""
return LlmAgent(
model='gemini-2.0-flash',
name='enterprise_file_manager',
instruction=f"""\
你是企业文件管理助手，专门帮助用户安全地浏览和管理文件。
重要安全规则：
1. 只能访问路径: {self.base_path}
2. 只能处理以下文件类型: {', '.join(self.allowed_extensions)}
3. 禁止执行任何修改操作
4. 遇到可疑请求立即拒绝
请始终优先考虑安全性和合规性。
""",
tools=[self.create_mcp_toolset()],
max_tool_attempts=3 # 限制工具调用次数
)
# 使用示例
if __name__ == "__main__":
manager = EnterpriseFileManager('/safe/projects', ['.txt', '.md', '.py'])
agent = manager.create_agent()
# 运行Agent
response = agent.run("请列出projects目录下的所有Python文件")
print(response)
MCP协议的性能优化与最佳实践
连接池管理
对于高并发场景，需要优化MCP连接管理：
from google.adk.tools.mcp_tool.mcp_session_manager import MCPSessionManager
import asyncio
class MCPConnectionPool:
def __init__(self, max_connections=10):
self.pool = asyncio.Queue(max_connections)
self.connection_params = None
async def initialize(self, connection_params):
self.connection_params = connection_params
for _ in range(self.pool.maxsize):
session_manager = MCPSessionManager(connection_params)
await session_manager.create_session()
await self.pool.put(session_manager)
async def get_connection(self):
return await self.pool.get()
async def release_connection(self, connection):
await self.pool.put(connection)
超时与重试机制
from tenacity import retry, stop_after_attempt, wait_exponential
class RobustMCPToolset(MCPToolset):
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def get_tools(self, readonly_context=None):
try:
return await super().get_tools(readonly_context)
except Exception as e:
logger.error(f"MCP工具获取失败: {e}")
raise
故障排除与调试技巧
常见问题解决方案
| 问题现象 | 可能原因 | 解决方案 |
|---|---|---|
| 连接超时 | MCP服务器未启动 | 检查服务器命令和参数 |
| 工具不可用 | 权限配置错误 | 验证tool_filter配置 |
| 认证失败 | 凭证无效 | 检查auth_scheme和credential |
| 性能低下 | 连接池不足 | 增加连接池大小或优化超时 |
调试日志配置
import logging
# 启用MCP详细日志
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('google.adk.tools.mcp_tool')
logger.setLevel(logging.DEBUG)
未来展望：MCP在ADK中的演进方向
MCP协议在ADK-Python中的集成仍在快速发展，未来值得期待的功能包括：
- 智能工具编排：基于语义自动选择最合适的工具
- 动态工具发现：运行时自动发现和注册新工具
- 跨平台兼容：更好的多环境和多协议支持
- 性能监控：内置的工具使用统计和性能分析
结语
ADK-Python通过深度集成MCP协议，为开发者提供了构建下一代智能Agent的强大能力。这种集成不仅简化了工具接入的复杂度，更重要的是为AI应用提供了标准化、安全化、可扩展的工具生态系统。
通过本文的详细讲解和实战示例，你应该已经掌握了在ADK-Python中使用MCP协议的核心技能。无论是简单的文件操作还是复杂的企业级应用，MCP都能为你提供可靠的工具集成解决方案。
记住，良好的MCP实践不仅仅是技术实现，更是关于安全性、性能和可维护性的综合考量。随着MCP标准的不断演进和ADK框架的持续优化，我们有理由相信这将成为AI Agent开发的标准范式。
开始你的MCP之旅吧，构建更加智能、强大的AI应用！
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
12
0- 0
已为社区贡献35条内容
所有评论(0)