A2A 协议实战：从 Agent Card 到跨系统多智能体协作（MCP 之外的下一站）
2026 年被称为"智能体爆发年"。麦肯锡《2026 企业级 AI 代理经济报告》显示，采用多智能体协作架构的系统，任务完成率较单体 Agent 提升 4.2 倍。但当你想让"自己团队的 Agent"调用"隔壁团队部署的远程 Agent"时，问题来了：两边根本说不上话。
一、引言：MCP 解决了"手"，没解决"脑与脑"
过去一年，开发者都听过 MCP（模型上下文协议）——它让 Agent 能调用数据库、搜索、文件等外部工具，被称为"AI 时代的 USB-C"。但请注意：MCP 连接的是 Agent 与工具，而 2026 年真正卡住企业的是另一个问题——Agent 与 Agent 之间怎么协作。
你的订单 Agent 需要把发票任务交给财务团队的 Agent；你的客服 Agent 想调用供应商的售后 Agent。跨团队、跨平台、跨编程语言，双方都用 MCP 也白搭——MCP 不负责 Agent 之间的分工协作。
这就是A2A（Agent2Agent）协议登场的背景：2025 年 4 月由 Google 提出，随后捐给 Linux 基金会，2026 年 3 月发布 1.0 正式版，Google、Anthropic、Salesforce、SAP 等几十家厂商集体签约。如果说 MCP 是"AI 时代的 USB-C"，A2A 就是"Agent 时代的 HTTP"——定义了 Agent 之间如何发现彼此、如何委托任务、如何跟踪进度。
本文用零依赖的 Python 代码，从 Agent Card 到手写 JSON-RPC 端点，带你完整实现一个 A2A 1.0 服务端和客户端。本质就一句话：把"人调用人"的协作流程，标准化成"Agent 调用 Agent"的协议。
二、核心原理：MCP 与 A2A 的分工
先建立一个稳定判断：LangGraph 之类框架解决"应用内编排"，A2A 解决"跨系统协作"。两者不互斥，一个生产级系统往往同时使用。
| 维度 | MCP | A2A |
| --- | --- | --- |
| 连接对象 | Agent ↔ 工具 / 资源 | Agent ↔ Agent |
| 关注重点 | 工具怎么暴露、资源怎么读取 | 任务怎么提交、状态怎么跟踪 |
| 典型问题 | 模型如何调用数据库、搜索、文件 | 一个 Agent 如何把任务委托给远程 Agent |
| 传输层 | JSON-RPC 2.0（stdio / HTTP / SSE） | JSON-RPC 2.0（HTTP / SSE 流式） |
| 核心产物 | tools / resources / prompts | Agent Card / Task / Message / Artifact |
| 状态管理 | 无状态，一次调用 | Task 状态机 + 长任务生命周期 |
A2A 的四个核心概念：
1.Agent Card：Agent 的"名片"，JSON 格式，声明能力、接口地址、协议版本，发布在 `/.well-known/agent-card.json`，供其他 Agent 发现；
2.Task（任务）：委托给远程 Agent 的工作单元，拥有唯一 ID 和状态机（`submitted → working → input-required → completed / failed / canceled`）；
3.Message / Part：消息体，由多个 Part 组成，支持文本、文件、结构化 JSON 等富媒体；
4.Artifact：长任务运行中产生的持久化产物（生成的文件、结构化结果），客户端可随时拉取。
整个协作流程可以概括为四步：发现——客户端读取远程 Agent 的 Agent Card，确认对方的能力、接口与协议版本；委托——通过 `SendMessage` 提交任务，并声明自己接受的输出格式；跟踪——凭 task id 轮询或订阅任务状态，长任务期间可接收中间消息与产物；结算——任务进入终态（completed / failed / canceled）后取回全部 Artifact。这套流程与人类团队协作的"分工—执行—汇报"高度同构，只是把人与人之间的沟通成本，压缩成了两个 JSON-RPC 调用。
三、代码实战：零依赖实现 A2A 1.0 服务端
官方有 `a2a-sdk`（Python），但为了看清协议本质，我们用 Flask 手写一个单位换算 Agent，只依赖 Flask + Requests。先看Agent Card：
# server.py —— 零依赖 A2A 1.0 服务器（仅 Flask） from flask import Flask, request, jsonify from uuid import uuid4 from datetime import datetime, timezone app = Flask(__name__) TASKS = {} # 内存任务存储，生产环境请换 Redis / Postgres AGENT_CARD = { "name": "Unit Converter", "description": "转换温度、距离、重量等常用单位", "version": "1.0.0", "supportedInterfaces": [ # 1.0 起传输元数据统一放在这里 {"url": "http://localhost:5000", "protocolBinding": "JSONRPC", "protocolVersion": "1.0"} ], "capabilities": {"streaming": False, "pushNotifications": False}, "defaultInputModes": ["text/plain"], "defaultOutputModes": ["text/plain"], "skills": [ {"id": "temperature", "name": "温度换算", "description": "华氏度/摄氏度互转", "examples": ["Convert 100 Fahrenheit to Celsius"]} ], "provider": {"organization": "A2A Tutorial", "url": "http://localhost:5000"}, }然后是路由与 JSON-RPC 分发。注意 A2A 1.0 要求请求头 `A2A-Version: 1.0`，且方法名从 0.3 时代的 `message/send`、`tasks/get` 改成了 `SendMessage`、`GetTask`——这是判断网上示例是否过时的最快方法：
def utc_now(): return datetime.now(timezone.utc).isoformat() def convert(text: str) -> str: # 简化换算：100 Fahrenheit -> 37.78 Celsius if "Fahrenheit" in text: f = float(text.split()[1]) return f"{round((f - 32) * 5 / 9, 2)} Celsius" return "暂不支持该换算" @app.route("/.well-known/agent-card.json") def agent_card(): return jsonify(AGENT_CARD) # 第一步：让其他 Agent 能发现我 @app.route("/", methods=["POST"]) def jsonrpc(): # 第二步：A2A 1.0 强制校验协议版本头 if request.headers.get("A2A-Version") != "1.0": return jsonify({"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "A2A-Version must be '1.0'"}}), 400 body = request.get_json(silent=True) or {} method, params, req_id = body.get("method"), body.get("params", {}), body.get("id") if method == "SendMessage": return handle_send(req_id, params) if method == "GetTask": return handle_get(req_id, params) if method == "CancelTask": return handle_cancel(req_id, params) return jsonify({"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": "Method not found"}}), 404 def handle_send(req_id, params): # 第三步：收到消息 -> 执行 -> 以 Task 形式返回（可异步轮询） msg = params["message"] text = msg["parts"][0]["text"] task_id, ctx_id = str(uuid4()), msg.get("contextId", str(uuid4())) result = convert(text) task = { "id": task_id, "contextId": ctx_id, "status": {"state": "TASK_STATE_COMPLETED", "timestamp": utc_now()}, "history": [msg, {"messageId": str(uuid4()), "contextId": ctx_id, "role": "ROLE_AGENT", "parts": [{"text": f"换算完成：{result}"}]}], "artifacts": [{"artifactId": str(uuid4()), "name": "conversion-result", "parts": [{"text": result}]}], } TASKS[task_id] = task return jsonify({"jsonrpc": "2.0", "id": req_id, "result": {"task": task}}) def handle_get(req_id, params): # 第四步：客户端凭 task id 拉取任务状态与产物 task = TASKS.get(params.get("id")) if not task: return jsonify({"jsonrpc": "2.0", "id": req_id, "error": {"code": -32004, "message": "Task not found"}}), 404 return jsonify({"jsonrpc": "2.0", "id": req_id, "result": task}) def handle_cancel(req_id, params): # 终态任务不可取消，返回 A2A 规范错误码 -32002 return jsonify({"jsonrpc": "2.0", "id": req_id, "error": {"code": -32002, "message": "TaskNotCancelableError"}}), 400 if __name__ == "__main__": app.run(port=5000)启动服务后，先用 curl 发现 Agent Card，再提交任务：
# 1. 发现 Agent（A2A 客户端的第一步永远是读名片） curl -s http://localhost:5000/.well-known/agent-card.json | python -m json.tool # 2. 委托任务：发送换算消息 curl -s -X POST http://localhost:5000 \ -H "Content-Type: application/json" -H "A2A-Version: 1.0" \ -d '{"jsonrpc":"2.0","id":"req-1","method":"SendMessage", "params":{"message":{"messageId":"msg-1","contextId":"ctx-1", "role":"ROLE_USER","parts":[{"text":"Convert 100 Fahrenheit to Celsius"}]}}}' \ | python -m json.tool # 3. 用返回的 task id 轮询状态 curl -s -X POST http://localhost:5000 -H "Content-Type: application/json" -H "A2A-Version: 1.0" \ -d '{"jsonrpc":"2.0","id":"req-2","method":"GetTask","params":{"id":"<TASK_ID>"}}' \ | python -m json.tool客户端同样遵循"先发现、后调用"的流程——这正是跨系统协作的核心：调用方不硬编码接口，而是动态读取 Agent Card：
# client.py —— 另一个 Agent 的视角：发现 -> 委托 -> 取结果 import uuid import requests AGENT_URL = "http://localhost:5000" # 第一步：发现能力，选择 JSONRPC 接口 card = requests.get(f"{AGENT_URL}/.well-known/agent-card.json").json() iface = next(i for i in card["supportedInterfaces"] if i["protocolBinding"] == "JSONRPC") def send_message(text: str) -> dict: payload = {"jsonrpc": "2.0", "id": str(uuid.uuid4()), "method": "SendMessage", "params": {"message": { "messageId": str(uuid.uuid4()), "contextId": str(uuid.uuid4()), "role": "ROLE_USER", "parts": [{"text": text}]}, "configuration": {"acceptedOutputModes": ["text/plain"]}}} resp = requests.post(AGENT_URL, json=payload, headers={"A2A-Version": "1.0"}) return resp.json()["result"]["task"] task = send_message("Convert 100 Fahrenheit to Celsius") print(task["status"]) # {'state': 'TASK_STATE_COMPLETED', ...} print(task["artifacts"][0]["parts"][0]["text"]) # 37.78 Celsius运行 `python server.py` 后再跑 `python client.py`，即可看到完整闭环：发现 → 委托 → 状态查询 → 取回产物。
四、2026 最新演进：A2A 生态的三个方向
1.SDK 与框架集成：官方 `a2a-sdk` 提供 `A2ACardResolver`、`InMemoryTaskStore` 等组件，几行代码即可发布合规服务端；LangGraph、CrewAI、Google ADK 均已支持 A2A 包装——你的编排图可以直接作为 A2A Server 对外暴露。
2.长任务治理：1.0 支持 `tasks/sendSubscribe` 走 SSE 流式返回进度、`tasks/pushNotification/set` 配置 webhook 推送、Artifact 流式产出——这是"任务提交后 20 分钟才有结果"的企业场景的刚需。
3.标准化与合规：A2A 已被 Linux 基金会托管，与 MCP 一同构成"工具层 + 协作层"的标准栈；配合 AgentOps（智能体运营）与 IEEE 的自主智能体互操作标准，企业跨部门、跨供应商的 Agent 生态正从"演示"走向"生产"。
五、总结与行动建议
5 个关键结论：
• MCP 管"Agent 调用工具"，A2A 管"Agent 调用 Agent"，两者互补而非互斥；
• Agent Card 是协作的起点：没有名片，就没有发现，就没有互操作；
• Task 状态机 + Artifact 是长任务协作的地基，别再让 Agent 同步干等；
• 1.0 的 `A2A-Version` 头与 `SendMessage/GetTask` 方法名是识别新旧示例的标尺；
• 协议标准化正在重演 REST 的历史：先乱后统，早接入者占生态位。
3 个行动建议：
1. 本周：跑通本文代码，用 curl 亲自"发现"一次 Agent Card；
2. 本月：把团队内一个高频 Agent（如报表、质检）改造成 A2A Server，内部先互操作；
3. 本季度：评估 LangGraph / CrewAI 的 A2A 集成，设计跨部门的 Agent 目录与任务治理规范。
参考资料：A2A Protocol 官方规范（Linux Foundation）；Waggle《A2A for Beginners》Part 2；麦肯锡《2026 企业级 AI 代理经济报告》；a2a-sdk PyPI 文档。
时效性提示：A2A 1.0 于 2026 年 3 月发布，本文基于 1.0 绑定，建议 3 个月内跟进官方 SDK 更新。