MCP + A2A 融合：协议层已就绪，信任层才是硬仗

MCP + A2A 融合：协议层已就绪，信任层才是硬仗

1. 从基础说起：MCP 与 A2A 是什么？在探讨融合之前，我们先明确两个核心概念。MCP（Model Context Protocol）是一种模型上下文协议，它定义了 AI 模型（如大型语言模型）如何与外部工具、数据源或服务进行交互。简单来说，MCP 让 AI 模型能“调用”外部函数，获取实时信息或执行操作。例如，一个聊天机器人可以通过 MCP 调用天气 API 来回答“今天北京天气如何？”。A2A（Agent-to-Agent Protocol）则是一种代理间通信协议，它规范了多个 AI 代理（Agent）如何相互协作。每个代理可能负责不同的任务（如搜索、计算、决策），A2A 让它们能共享数据、协调行动，形成更复杂的智能系统。为什么需要融合？MCP 解决的是“模型与外部世界”的交互，而 A2A 解决的是“代理与代理”的交互。当我们需要构建一个多代理系统，其中每个代理都通过 MCP 连接外部工具时，两者必须无缝协作。例如，一个“旅行规划代理”需要调用天气 MCP 和酒店预订 MCP，同时与“财务代理”通过 A2A 协商预算。从协议层看，MCP 和 A2A 的接口设计已经成熟，可以轻松整合。但真正的挑战在于信任层——如何确保代理之间的交互是安全、可靠且可审计的？这正是本文的核心。## 2. 协议层的融合：基础代码示例让我们用一个简单的 Python 示例，展示 MCP 和 A2A 如何协同工作。假设我们有两个代理：WeatherAgent负责查询天气（通过 MCP），TravelAdvisorAgent负责整合信息（通过 A2A 调用 WeatherAgent）。### 代码示例 1：MCP 基础实现python# 模拟 MCP 协议：定义一个天气查询工具class WeatherMCP: """MCP 工具：提供天气查询功能""" def __init__(self): self.name = "weather_tool" self.description = "查询指定城市的天气" def execute(self, city: str) -> str: # 模拟 API 调用 weather_data = { "北京": "晴，25°C", "上海": "多云，28°C", "广州": "雨，30°C" } return weather_data.get(city, "未知城市")# 定义 WeatherAgent，它通过 MCP 与外部数据交互class WeatherAgent: """一个简单的 AI 代理，使用 MCP 查询天气""" def __init__(self): self.mcp_tool = WeatherMCP() def handle_query(self, city: str) -> str: # 模拟模型通过 MCP 调用工具 print(f"[WeatherAgent] 通过 MCP 调用 {self.mcp_tool.name}") result = self.mcp_tool.execute(city) return f"{city} 的天气是：{result}"# 测试if __name__ == "__main__": agent = WeatherAgent() print(agent.handle_query("北京"))输出：[WeatherAgent] 通过 MCP 调用 weather_tool北京 的天气是：晴，25°C### 代码示例 2：A2A 通信与信任验证现在，我们引入 A2A 协议，让TravelAdvisorAgent通过 A2A 调用WeatherAgent。为了强调信任层，我们会添加一个简单的身份验证和日志记录机制。pythonimport jsonfrom datetime import datetime# 模拟 A2A 协议的消息格式class A2AMessage: """A2A 协议消息，包含身份验证字段""" def __init__(self, sender: str, receiver: str, payload: dict, token: str = None): self.sender = sender self.receiver = receiver self.payload = payload self.token = token # 信任凭证 self.timestamp = datetime.now().isoformat() def to_dict(self): return { "sender": self.sender, "receiver": self.receiver, "payload": self.payload, "token": self.token, "timestamp": self.timestamp }# 增强版 WeatherAgent，支持 A2A 通信class SecureWeatherAgent(WeatherAgent): """带信任验证的 WeatherAgent""" def __init__(self): super().__init__() self.trusted_tokens = {"travel_advisor": "token_12345"} # 信任白名单 def handle_a2a_request(self, message: A2AMessage) -> str: # 信任层验证：检查发送者是否有有效令牌 if message.sender not in self.trusted_tokens: return "错误：未授权的请求（信任验证失败）" if message.token != self.trusted_tokens[message.sender]: return "错误：令牌无效（信任验证失败）" # 记录日志（信任审计） print(f"[信任日志] {message.timestamp}: {message.sender} -> {message.receiver}, 令牌验证通过") # 处理 A2A 消息中的 payload city = message.payload.get("city") return super().handle_query(city)# TravelAdvisorAgent，通过 A2A 调用其他代理class TravelAdvisorAgent: """旅行建议代理，通过 A2A 协议协调其他代理""" def __init__(self): self.name = "travel_advisor" self.token = "token_12345" # 自己的凭证 def plan_trip(self, city: str, weather_agent: SecureWeatherAgent) -> str: # 构建 A2A 消息 a2a_msg = A2AMessage( sender=self.name, receiver="weather_agent", payload={"city": city}, token=self.token ) # 通过 A2A 发送请求 print(f"[A2A] {self.name} 发送请求给 weather_agent") response = weather_agent.handle_a2a_request(a2a_msg) return f"旅行建议：{response}"# 测试信任层if __name__ == "__main__": weather_agent = SecureWeatherAgent() travel_agent = TravelAdvisorAgent() # 正常请求（带有效令牌） print(travel_agent.plan_trip("上海", weather_agent)) print("\n--- 测试未授权请求 ---") # 模拟恶意代理 malicious_msg = A2AMessage( sender="hacker", receiver="weather_agent", payload={"city": "广州"}, token="fake_token" ) print(weather_agent.handle_a2a_request(malicious_msg))输出：[A2A] travel_advisor 发送请求给 weather_agent[信任日志] 2025-04-10T15:30:00: travel_advisor -> weather_agent, 令牌验证通过[WeatherAgent] 通过 MCP 调用 weather_tool旅行建议：上海 的天气是：多云，28°C--- 测试未授权请求 ---错误：未授权的请求（信任验证失败）## 3. 信任层的深层挑战：从代码到现实上述代码示例展示了信任层的基本实现：通过令牌验证确保只有授权的代理才能调用服务。但在真实系统中，信任层远比这复杂：### 3.1 身份与凭证管理-动态信任：令牌可能过期，需要实时刷新。例如，在 A2A 通信中，代理可能需要从证书颁发机构获取短期令牌。-多级信任：一个代理可能代表用户，用户身份需要向下传递（即“代理链”信任）。这要求协议支持“委托式验证”。### 3.2 数据完整性- 即使身份验证通过，消息在传输中可能被篡改。因此，A2A 消息通常需要数字签名（如 HMAC 或 JWT）。### 3.3 审计与可追溯性- 在示例中，我们记录了日志，但真实系统需要不可篡改的审计链（如区块链或中心化日志服务），以便在事故后追溯。### 3.4 安全沙箱- 通过 MCP 调用的外部工具可能包含恶意代码（如 SQL 注入）。信任层必须确保代理只能执行白名单内的操作，且返回值是安全的。## 4. 信任层的进阶实践：JWT 签名与审计下面是一个更高级的示例，使用 JSON Web Token (JWT) 来加强信任：pythonimport jwtimport datetime# 密钥（实际生产环境应从安全存储获取）SECRET_KEY = "my_secret_key_123"class SecureA2AMessage: """使用 JWT 签名的 A2A 消息""" def __init__(self, sender: str, receiver: str, payload: dict): self.sender = sender self.receiver = receiver self.payload = payload self.token = self._generate_token() def _generate_token(self) -> str: # 创建 JWT，包含发送者、接收者和过期时间 payload = { "sender": self.sender, "receiver": self.receiver, "data": self.payload, "exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=5) } return jwt.encode(payload, SECRET_KEY, algorithm="HS256")def verify_a2a_message(message: SecureA2AMessage) -> bool: """验证消息的签名和有效性""" try: decoded = jwt.decode(message.token, SECRET_KEY, algorithms=["HS256"]) # 检查接收者是否匹配 if decoded["receiver"] != message.receiver: return False return True except jwt.ExpiredSignatureError: print("令牌已过期") return False except jwt.InvalidTokenError: print("无效令牌") return False# 测试msg = SecureA2AMessage("travel_advisor", "weather_agent", {"city": "北京"})print(f"原始消息令牌：{msg.token}")print(f"验证结果：{verify_a2a_message(msg)}")输出：原始消息令牌：eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...验证结果：True## 5. 总结MCP 和 A2A 的融合在协议层面已经万事俱备——MCP 提供了模型与工具的桥梁，A2A 构建了代理间的协作网络。但正如我们通过代码示例所揭示的，信任层才是决定这一架构能否落地的硬仗。从基础的令牌验证，到 JWT 签名、动态凭证和审计链，信任层需要解决身份、完整性、可追溯性和安全沙箱四大问题。对于开发者而言，这不仅是技术实现，更是一种设计思维：在构建智能系统时，必须将信任作为第一性原理，而非事后修补。未来，随着 AI 代理越来越自主，信任层将成为类似“网络防火墙”的基础设施。作为编程讲师，我建议你在学习 MCP 和 A2A 时，始终带着“信任视角”去思考：谁可以调用谁？数据是否被篡改？行为能否被审计？只有这样，我们才能构建真正可靠、可用的多代理系统。