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 之间的分工协作。这就是A2AAgent2Agent协议登场的背景2025 年 4 月由 Google 提出随后捐给 Linux 基金会2026 年 3 月发布 1.0 正式版Google、Anthropic、Salesforce、SAP 等几十家厂商集体签约。如果说 MCP 是AI 时代的 USB-CA2A 就是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.0stdio / HTTP / SSE | JSON-RPC 2.0HTTP / SSE 流式 || 核心产物 | tools / resources / prompts | Agent Card / Task / Message / Artifact || 状态管理 | 无状态一次调用 | Task 状态机 长任务生命周期 |A2A 的四个核心概念1.Agent CardAgent 的名片JSON 格式声明能力、接口地址、协议版本发布在 /.well-known/agent-card.json供其他 Agent 发现2.Task任务委托给远程 Agent 的工作单元拥有唯一 ID 和状态机submitted → working → input-required → completed / failed / canceled3.Message / Part消息体由多个 Part 组成支持文本、文件、结构化 JSON 等富媒体4.Artifact长任务运行中产生的持久化产物生成的文件、结构化结果客户端可随时拉取。整个协作流程可以概括为四步发现——客户端读取远程 Agent 的 Agent Card确认对方的能力、接口与协议版本委托——通过 SendMessage 提交任务并声明自己接受的输出格式跟踪——凭 task id 轮询或订阅任务状态长任务期间可接收中间消息与产物结算——任务进入终态completed / failed / canceled后取回全部 Artifact。这套流程与人类团队协作的分工—执行—汇报高度同构只是把人与人之间的沟通成本压缩成了两个 JSON-RPC 调用。三、代码实战零依赖实现 A2A 1.0 服务端官方有 a2a-sdkPython但为了看清协议本质我们用 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(silentTrue) 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(port5000)启动服务后先用 curl 发现 Agent Card再提交任务# 1. 发现 AgentA2A 客户端的第一步永远是读名片 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, jsonpayload, 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 Card2. 本月把团队内一个高频 Agent如报表、质检改造成 A2A Server内部先互操作3. 本季度评估 LangGraph / CrewAI 的 A2A 集成设计跨部门的 Agent 目录与任务治理规范。参考资料A2A Protocol 官方规范Linux FoundationWaggle《A2A for Beginners》Part 2麦肯锡《2026 企业级 AI 代理经济报告》a2a-sdk PyPI 文档。时效性提示A2A 1.0 于 2026 年 3 月发布本文基于 1.0 绑定建议 3 个月内跟进官方 SDK 更新。
