[智能体-345]：A2A (Agent2Agent) 最小可运行示例

A2A (Agent2Agent) 最小可运行示例

基于JSON-RPC 2.0 + HTTP实现标准 A2A 通信，包含：Agent Card 定义、服务端 Agent、客户端 Agent、完整调用链路，纯 Python、无额外复杂依赖。

环境依赖

仅需内置库 +flask（简易 HTTP 服务）

bash

运行

pip install flask

一、整体架构说明

1. Server Agent（服务端）：对外暴露 A2A 接口，提供数据查询能力，自带标准Agent Card
2. Agent Card：遵循A2A 规范，描述身份、能力、接口地址、支持协议
3. Client Agent（客户端）：拉取对方 Agent Card → 能力校验 → 发起 A2A 任务调用
4. 通信格式：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(host="127.0.0.1", port=8000, debug=False)

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, timeout=5) 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、上下文ID（A2A 标准字段） rpc_id = str(uuid.uuid4()) context_id = f"ctx_{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, json=rpc_body, timeout=10) 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( method="data_query", params={"query_key": "sales"} ) # 4. 打印最终结果 print("=== A2A 调用结果 ===") print(rpc_result)

三、运行步骤

1. 启动服务端

bash

运行

python a2a_server.py

服务会监听127.0.0.1:8000

1. 新开终端，启动客户端

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 规范要点（代码对应）

1. Agent Card 标准路径固定路由：/.well-known/agent.json，所有 A2A Agent 必须实现，用于能力自动发现。
2. 核心字段
   • a2a_version：协议版本，用于兼容性校验
   • capabilities：对外暴露的能力列表（客户端用来做能力匹配）
   • endpoints.rpc：JSON-RPC 通信地址
3. 任务标识
   • id：JSON-RPC 请求 ID
   • context_id：A2A 上下文 ID，多轮协作 / 关联任务靠它串联
   • task_status：任务状态（pending/running/completed/failed）

五、生产级扩展方向

1. 增加认证：对接 OAuth2 / APIKey /mTLS，替换示例中auth_schemes: ["none"]
2. 流式交互：基于 WebSocket/SSE 实现 A2A Stream 模式
3. 异步长任务：增加push notification（Webhook）回调
4. 注册中心：接入公共 / 企业 Agent 注册中心，实现大规模 Agent 发现
5. 状态持久化：数据库存储task/context数据，支持跨会话协作

六、结合 MCP 联动（补充）

当前示例是Agent ↔ Agent (A2A)，如果想让服务端 Agent 再调用工具，只需在服务端内部接入MCP：Client Agent →(A2A)→ Server Agent →(MCP)→ 外部工具/数据库/接口形成完整的「智能体协作 + 工具调用」协议栈。