从零开始学A2A二 : A2A 协议的技术架构与实现

A2A 协议的技术架构与实现

学习目标

技术架构掌握
- 深入理解 A2A 协议的分层架构设计
- 掌握各层次的功能和职责
- 理解协议的工作原理和数据流
实现能力培养
- 能够搭建基本的 A2A 服务端
- 掌握客户端开发方法
- 实现智能体间的有效通信
架构设计理解
- 理解与 MCP 的本质区别
- 掌握多智能体协作模式
- 学习分布式系统设计

一、A2A 的技术架构

1. 通信架构概述

A2A 协议采用现代化的三层架构设计：
在这里插入图片描述

HTTP/HTTPS 层：基础通信层
JSON-RPC 层：远程调用层
核心服务层：业务功能层

2. 各层详细说明

2.1 HTTP/HTTPS 层实现

class A2AAgent:def __init__(self, agent_id: str):self.agent_id = agent_idself.capabilities = set()self.comm = AgentCommunication()async def handle_request(self, request: Request):if not self.can_handle(request):other_agent = await self.discover_capable_agent(request)return await self.delegate_request(other_agent, request)return await self.process_request(request)

2.2 JSON-RPC 层实现

class MessageRouter:def __init__(self):self.routes = {}async def route_message(self, message: Message):if message.target in self.routes:handler = self.routes[message.target]await handler(message)

2.3 核心服务层实现

class ServiceRegistry:def __init__(self):self.services = {}def register(self, service_id: str, capabilities: List[str]):self.services[service_id] = {"capabilities": capabilities,"status": "active","registered_at": datetime.now()}

二、A2A 任务生命周期

1. 任务流程

在这里插入图片描述

任务创建
- 生成唯一标识
- 设定任务参数
- 确定执行要求
状态更新
- 任务分配状态
- 执行进度更新
- 异常状态处理
结果返回
- 数据格式化
- 结果验证
- 回调处理

2. 实现示例

class TaskLifecycle:def __init__(self):self.task_store = {}async def create_task(self, spec: Dict) -> str:task_id = str(uuid4())self.task_store[task_id] = {"spec": spec,"status": "created","created_at": datetime.now()}return task_idasync def update_status(self, task_id: str, status: str):if task_id in self.task_store:self.task_store[task_id]["status"] = statusself.task_store[task_id]["updated_at"] = datetime.now()async def complete_task(self, task_id: str, result: Any):if task_id in self.task_store:self.task_store[task_id].update({"status": "completed","result": result,"completed_at": datetime.now()})

三、与 MCP 的架构差异

1. 核心设计理念对比

特性	MCP	A2A
架构重点	单体智能体增强	多智能体协作
上下文管理	完整上下文传递	任务相关上下文
通信模式	工具调用	对等通信
扩展方式	垂直扩展	水平扩展

在这里插入图片描述

架构差异说明：

MCP：单体智能体通过工具调用扩展能力，垂直扩展方式
A2A：多智能体通过对等通信协作，水平扩展方式

2. 实现差异示例

MCP 实现

class MCPAgent:def __init__(self):self.tools = {}self.context = Context()def execute_tool(self, tool_name: str, params: Dict):if tool_name in self.tools:return self.tools[tool_name].execute(params, self.context)

A2A 实现

class A2AAgent:def __init__(self, agent_id: str):self.agent_id = agent_idself.capabilities = set()self.peers = {}async def collaborate(self, task: Task):if task.requires_capability not in self.capabilities:peer = self.find_capable_peer(task.requires_capability)return await peer.handle_task(task)return await self.process_task(task)