![[智能体-345]：A2A (Agent2Agent) 最小可运行示例](images/news1.jpg)
A2A (Agent2Agent) 最小可运行示例基于JSON-RPC 2.0 HTTP实现标准 A2A 通信包含Agent Card 定义、服务端 Agent、客户端 Agent、完整调用链路纯 Python、无额外复杂依赖。环境依赖仅需内置库 flask简易 HTTP 服务bash运行pip install flask一、整体架构说明Server Agent服务端对外暴露 A2A 接口提供数据查询能力自带标准Agent CardAgent Card遵循A2A 规范描述身份、能力、接口地址、支持协议Client Agent客户端拉取对方 Agent Card → 能力校验 → 发起 A2A 任务调用通信格式JSON-RPC 2.0完全对齐 A2A v0.2.x 规范二、完整代码分为服务端 客户端1. A2A 服务端 (a2a_server.py)python运行from flask import Flask, request, jsonify app Flask(__name__) # 1. 标准 A2A Agent Card (智能体名片) AGENT_CARD { a2a_version: 0.2.5, agent_id: data_agent_001, name: Data Query Agent, description: 数据查询专用智能体支持基础数据统计, endpoints: { rpc: http://127.0.0.1:8000/rpc }, capabilities: [ {name: data_query, desc: 执行数据查询任务} ], auth_schemes: [none], # 示例免认证生产用 OAuth2/mTLS interaction_modes: [sync, stream] } # 模拟业务数据 mock_data {sales: 98600, user_count: 12500, order_num: 3680} # 2. A2A 标准接口 # 1) 拉取 Agent Card 标准路由/.well-known/agent.json app.route(/.well-known/agent.json, methods[GET]) def get_agent_card(): return jsonify(AGENT_CARD) # 2) JSON-RPC 任务处理接口 (A2A 核心通信入口) app.route(/rpc, methods[POST]) def a2a_rpc(): req_data request.get_json() # 解析 JSON-RPC 标准字段 rpc_id req_data.get(id) method req_data.get(method) params req_data.get(params, {}) # A2A 任务状态枚举 task_status completed result {} # 能力分发匹配 Agent 提供的能力 if method data_query: query_key params.get(query_key, ) result[data] mock_data.get(query_key, 无对应数据) result[task_status] task_status result[context_id] params.get(context_id, ) else: return jsonify({ jsonrpc: 2.0, id: rpc_id, error: {code: -32601, message: 方法不存在} }) # 返回 JSON-RPC 标准响应 return jsonify({ jsonrpc: 2.0, id: rpc_id, result: result }) if __name__ __main__: app.run(host127.0.0.1, port8000, debugFalse)2. A2A 客户端 (a2a_client.py)python运行import requests import uuid # 服务端基础地址 SERVER_BASE http://127.0.0.1:8000 class A2AClientAgent: def __init__(self): self.agent_card None def fetch_agent_card(self): 步骤1拉取对方 Agent Card能力发现 url f{SERVER_BASE}/.well-known/agent.json resp requests.get(url, timeout5) self.agent_card resp.json() print( 成功获取对方 Agent Card ) print(self.agent_card, \n) def check_capability(self, cap_name: str) - bool: 步骤2校验对方是否具备目标能力 if not self.agent_card: return False caps [c[name] for c in self.agent_card[capabilities]] return cap_name in caps def call_a2a_task(self, method: str, params: dict): 步骤3发起 A2A 任务调用JSON-RPC rpc_url self.agent_card[endpoints][rpc] # 生成唯一任务ID、上下文IDA2A 标准字段 rpc_id str(uuid.uuid4()) context_id fctx_{uuid.uuid4()} params[context_id] context_id # 构造 JSON-RPC 2.0 请求体 rpc_body { jsonrpc: 2.0, id: rpc_id, method: method, params: params } print(f 发起A2A任务 | context_id: {context_id} ) resp requests.post(rpc_url, jsonrpc_body, timeout10) return resp.json() if __name__ __main__: # 初始化客户端智能体 client A2AClientAgent() # 1. 发现对方Agent client.fetch_agent_card() # 2. 校验能力 target_cap data_query if not client.check_capability(target_cap): print(f对方不具备 {target_cap} 能力) exit() # 3. 执行A2A跨智能体调用 rpc_result client.call_a2a_task( methoddata_query, params{query_key: sales} ) # 4. 打印最终结果 print( A2A 调用结果 ) print(rpc_result)三、运行步骤启动服务端bash运行python a2a_server.py服务会监听127.0.0.1:8000新开终端启动客户端bash运行python a2a_client.py预期输出plaintext 成功获取对方 Agent Card {...完整Agent Card信息...} 发起A2A任务 | context_id: ctx_xxxxxxx A2A 调用结果 { jsonrpc: 2.0, id: uuid字符串, result: { context_id: ctx_xxxxxxx, data: 98600, task_status: completed } }四、核心 A2A 规范要点代码对应Agent Card 标准路径固定路由/.well-known/agent.json所有 A2A Agent 必须实现用于能力自动发现。核心字段a2a_version协议版本用于兼容性校验capabilities对外暴露的能力列表客户端用来做能力匹配endpoints.rpcJSON-RPC 通信地址任务标识idJSON-RPC 请求 IDcontext_idA2A 上下文 ID多轮协作 / 关联任务靠它串联task_status任务状态pending/running/completed/failed五、生产级扩展方向增加认证对接 OAuth2 / APIKey /mTLS替换示例中auth_schemes: [none]流式交互基于 WebSocket/SSE 实现 A2A Stream 模式异步长任务增加push notificationWebhook回调注册中心接入公共 / 企业 Agent 注册中心实现大规模 Agent 发现状态持久化数据库存储task/context数据支持跨会话协作六、结合 MCP 联动补充当前示例是Agent ↔ Agent (A2A)如果想让服务端 Agent 再调用工具只需在服务端内部接入MCPClient Agent →(A2A)→ Server Agent →(MCP)→ 外部工具/数据库/接口形成完整的「智能体协作 工具调用」协议栈。
[智能体-345]：A2A (Agent2Agent) 最小可运行示例
A2A (Agent2Agent) 最小可运行示例基于JSON-RPC 2.0 HTTP实现标准 A2A 通信包含Agent Card 定义、服务端 Agent、客户端 Agent、完整调用链路纯 Python、无额外复杂依赖。环境依赖仅需内置库 flask简易 HTTP 服务bash运行pip install flask一、整体架构说明Server Agent服务端对外暴露 A2A 接口提供数据查询能力自带标准Agent CardAgent Card遵循A2A 规范描述身份、能力、接口地址、支持协议Client Agent客户端拉取对方 Agent Card → 能力校验 → 发起 A2A 任务调用通信格式JSON-RPC 2.0完全对齐 A2A v0.2.x 规范二、完整代码分为服务端 客户端1. A2A 服务端 (a2a_server.py)python运行from flask import Flask, request, jsonify app Flask(__name__) # 1. 标准 A2A Agent Card (智能体名片) AGENT_CARD { a2a_version: 0.2.5, agent_id: data_agent_001, name: Data Query Agent, description: 数据查询专用智能体支持基础数据统计, endpoints: { rpc: http://127.0.0.1:8000/rpc }, capabilities: [ {name: data_query, desc: 执行数据查询任务} ], auth_schemes: [none], # 示例免认证生产用 OAuth2/mTLS interaction_modes: [sync, stream] } # 模拟业务数据 mock_data {sales: 98600, user_count: 12500, order_num: 3680} # 2. A2A 标准接口 # 1) 拉取 Agent Card 标准路由/.well-known/agent.json app.route(/.well-known/agent.json, methods[GET]) def get_agent_card(): return jsonify(AGENT_CARD) # 2) JSON-RPC 任务处理接口 (A2A 核心通信入口) app.route(/rpc, methods[POST]) def a2a_rpc(): req_data request.get_json() # 解析 JSON-RPC 标准字段 rpc_id req_data.get(id) method req_data.get(method) params req_data.get(params, {}) # A2A 任务状态枚举 task_status completed result {} # 能力分发匹配 Agent 提供的能力 if method data_query: query_key params.get(query_key, ) result[data] mock_data.get(query_key, 无对应数据) result[task_status] task_status result[context_id] params.get(context_id, ) else: return jsonify({ jsonrpc: 2.0, id: rpc_id, error: {code: -32601, message: 方法不存在} }) # 返回 JSON-RPC 标准响应 return jsonify({ jsonrpc: 2.0, id: rpc_id, result: result }) if __name__ __main__: app.run(host127.0.0.1, port8000, debugFalse)2. A2A 客户端 (a2a_client.py)python运行import requests import uuid # 服务端基础地址 SERVER_BASE http://127.0.0.1:8000 class A2AClientAgent: def __init__(self): self.agent_card None def fetch_agent_card(self): 步骤1拉取对方 Agent Card能力发现 url f{SERVER_BASE}/.well-known/agent.json resp requests.get(url, timeout5) self.agent_card resp.json() print( 成功获取对方 Agent Card ) print(self.agent_card, \n) def check_capability(self, cap_name: str) - bool: 步骤2校验对方是否具备目标能力 if not self.agent_card: return False caps [c[name] for c in self.agent_card[capabilities]] return cap_name in caps def call_a2a_task(self, method: str, params: dict): 步骤3发起 A2A 任务调用JSON-RPC rpc_url self.agent_card[endpoints][rpc] # 生成唯一任务ID、上下文IDA2A 标准字段 rpc_id str(uuid.uuid4()) context_id fctx_{uuid.uuid4()} params[context_id] context_id # 构造 JSON-RPC 2.0 请求体 rpc_body { jsonrpc: 2.0, id: rpc_id, method: method, params: params } print(f 发起A2A任务 | context_id: {context_id} ) resp requests.post(rpc_url, jsonrpc_body, timeout10) return resp.json() if __name__ __main__: # 初始化客户端智能体 client A2AClientAgent() # 1. 发现对方Agent client.fetch_agent_card() # 2. 校验能力 target_cap data_query if not client.check_capability(target_cap): print(f对方不具备 {target_cap} 能力) exit() # 3. 执行A2A跨智能体调用 rpc_result client.call_a2a_task( methoddata_query, params{query_key: sales} ) # 4. 打印最终结果 print( A2A 调用结果 ) print(rpc_result)三、运行步骤启动服务端bash运行python a2a_server.py服务会监听127.0.0.1:8000新开终端启动客户端bash运行python a2a_client.py预期输出plaintext 成功获取对方 Agent Card {...完整Agent Card信息...} 发起A2A任务 | context_id: ctx_xxxxxxx A2A 调用结果 { jsonrpc: 2.0, id: uuid字符串, result: { context_id: ctx_xxxxxxx, data: 98600, task_status: completed } }四、核心 A2A 规范要点代码对应Agent Card 标准路径固定路由/.well-known/agent.json所有 A2A Agent 必须实现用于能力自动发现。核心字段a2a_version协议版本用于兼容性校验capabilities对外暴露的能力列表客户端用来做能力匹配endpoints.rpcJSON-RPC 通信地址任务标识idJSON-RPC 请求 IDcontext_idA2A 上下文 ID多轮协作 / 关联任务靠它串联task_status任务状态pending/running/completed/failed五、生产级扩展方向增加认证对接 OAuth2 / APIKey /mTLS替换示例中auth_schemes: [none]流式交互基于 WebSocket/SSE 实现 A2A Stream 模式异步长任务增加push notificationWebhook回调注册中心接入公共 / 企业 Agent 注册中心实现大规模 Agent 发现状态持久化数据库存储task/context数据支持跨会话协作六、结合 MCP 联动补充当前示例是Agent ↔ Agent (A2A)如果想让服务端 Agent 再调用工具只需在服务端内部接入MCPClient Agent →(A2A)→ Server Agent →(MCP)→ 外部工具/数据库/接口形成完整的「智能体协作 工具调用」协议栈。
