上篇我们聊了大方向上的Agent协作形态为什么需要Agent之间直接对话以及A2A协议解决的核心问题。这篇进入“真刀真枪”的部分消息到底长什么样、连接怎么建立、任务状态怎么流转、鉴权怎么做最后给你一个能在本地跑起来的最小Demo再加上我联调过程中踩过的几个坑。如果你是那种“看完协议文档总觉得差点意思”的人这篇应该能补上那段距离。1. 从Agent Card开始连接建立前的那张“名片”A2A协议里有个很朴素的前提——一个Agent要能主动发起任务它得先知道另一个Agent在哪儿、能干什么、怎么联系。这跟两个人合作一样你可以不认识对方但总得先拿到对方的名片。Agent Card就是这个名片。1.1 一张Agent Card里到底写了什么按照A2A协议规范的思路Agent Card是一个JSON文档描述Agent的公开信息。我在自己的模拟项目里维护的卡片大概长这样{ name: Zenith Translation Agent, description: 负责英文到中文的文本翻译支持技术文档、日常文本风格, version: 1.2.0, url: https://agent.example.com/translation, endpoints: [ { path: /a2a, protocol: a2a, transport: http } ], capabilities: { task_state: true, streaming: true, push_notifications: false }, skills: [ { id: translate_text, name: 翻译文本, description: 输入英文文本输出中文翻译, input_modalities: [text, file], output_modalities: [text, file] }, { id: format_doc, name: 格式化文档, description: 对原始文档进行结构化整理输出Markdown格式, input_modalities: [text], output_modalities: [text] } ], provider: { organization: example.dev, version: 1.0.0, reference: agent-card-2024 } }别被这么多字段唬住。最关键的就三个endpoints——Agent暴露的调用入口既包括A2A专用路径也可以包含HTTP的握手路径。这是别人“敲门”的门牌号。capabilities——Agent支持哪些交互方式比如是否支持长任务状态查询、是否支持流式返回、是否支持主动推送。skills——Agent能干的活儿每一项都定义清楚输入输出用文本还是文件。这相当于工具清单客户端会根据这个清单决定把什么任务派给谁。我见过不少新手一上来就急着写业务代码把Agent Card当成“事后补的配置文件”。实际上这个卡片是协作的源头。客户端通常都是先读卡、再决定调不调你卡片写得不清楚Agent做得再好也可能没人调。1.2 卡片的发现与缓存刷新Agent Card的获取方式一般是直接通过URL拉取。业界常见的做法是Agent把卡片放在一个稳定路径下比如https://agent.example.com/.well-known/agent.json或者由注册中心返回。你在本地调试时也可以把它当成一个静态文件让HTTP服务直接serve出来。这里有一个非常容易踩的坑客户端会缓存Agent Card而且缓存刷新策略各不相同。我在某个内部模拟项目中遇到过Agent上线了新skill我更新了卡片但另一边的编排程序调用了半天还是提示“unknown skill”。排查到最后才发现对方把卡片缓存存在本地Redis里TTL设置成了24小时。我改完卡片对方要隔一天才拿到最新能力。如果你在写客户端建议对Agent Card做两级处理会话外缓存比如5-10分钟降低频繁拉卡的带宽开销一旦收到agent_card_updated这类变更通知立即强制刷新。如果你在写Agent服务端就要保证卡片接口不要被网关缓存误伤。许多团队在Nginx层给静态JSON做了强制缓存忘了排除Agent Card路径结果所有协作方都被旧卡片误导。2. 消息是JSON-RPCA2A通信的“包裹”长什么样A2A没有发明一套新的消息语言而是直接采用了JSON-RPC 2.0作为消息层基础。这个决策在实践上非常省事不必自己定义序列化规则和错误格式所有语言生态里几乎都有现成的JSON-RPC库或者至少你能很快手写一个解析器。2.1 为什么要用JSON-RPC而不是REST有很多人问过我这个问题。我个人的理解是Agent协作是一种“远程能力调用”它天然对应RPC模型。你对另一个Agent说“帮我翻译这段文字”“帮我查一下某份文档”本质上是一个带参数的过程调用。REST风格更适合资源管理但不太适合这种“请求某个Agent干活”的语义。JSON-RPC 2.0有几个优势请求/响应结构非常清楚字段固定容易做中间层debug标准里定义了错误对象error code协作双方可以快速对齐失败原因支持“通知”类消息即不需要对方返回响应的单向消息适合做事件通知与语言和传输层解耦HTTP、WebSocket、SSE都能承载同一套消息体。A2A对JSON-RPC做了一些扩展比如在params里携带任务ID、消息对象、会话上下文等。但核心骨架还是标准的那一套。2.2 一条消息从客户端到Agent的完整链路我以一个最简单的“发送任务”请求为例展示一条消息从调用方到Agent的完整链路。假设编排方要让翻译Agent执行一次翻译{ jsonrpc: 2.0, id: req-8f3c8d0e-b2f7-4b1f-9a52-4a2c9c7f8b30, method: tasks/send, params: { taskId: task-1701000000001, message: { role: agent, sender: agent-orchestrator-01, time: 2025-05-12T10:00:00Z, parts: [ { type: text, text: 请把下面这段英文翻译成中文\nThe quick brown fox jumps over the lazy dog. } ] } } }逐字段拆解jsonrpc和id是JSON-RPC的固定内容id由调用方生成用来关联请求和响应。很多人会忽略id的唯一性这个问题我后面在踩坑部分专门说。method是A2A定义的方法名。常见的包括tasks/send向Agent发起一个新的任务并携带第一条消息message/send在已有任务上继续发消息tasks/get查询任务状态tasks/cancel请求取消任务tasks/pushNotificationConfig配置push通知。params.taskId标识一个任务实例。注意任务ID由调用方主动生成而不是Agent返回——这是很多刚从REST习惯转过来的人需要适应的点。params.message消息本身由role、sender、time和parts组成。parts是消息内容的数组每个part都有一个type字段可以是text、file或data。Agent端收到这条消息后正常会返回一个tasks/send响应。如果任务是异步的响应里会带上当前任务状态一般是working。客户端再通过tasks/get轮询或通过streaming拿后续结果。2.3 内容对象与artifacts的边界A2A协议里有两个容易混淆的概念message.parts和artifacts。简单来说parts是“Agent之间的对话内容”有点像一个Agent对另一个Agent说的一句话这句话可能是文本、文件引用或者结构化数据。而artifacts是“Agent完成工作后产出的成果物”是这个Agent“做出来”的东西。我举个例子。你让翻译Agent“翻译并整理一份技术文档”你的请求文本是parts翻译完成后Agent返回的Markdown格式文档属于artifacts。这两个概念在协议实现里对应不同的数据结构千万别混在一起。如果你想让下游Agent直接消费结果就应该把结果放到artifacts里让下游用file或data类型引用。如果只是对话中临时传一下再放parts也不迟。这个区分直接影响协作链路里下游Agent的解析逻辑。3. 任务生命周期多Agent协作真正在“跑”的东西两个Agent之间的一次合作在A2A协议眼里就是一个“任务”task。任务从创建到结束经历一系列状态。理解这套状态机基本上就理解了A2A的协作骨架。3.1 任务状态机从submitted到completed在A2A规范草案里一个任务通常包含以下几种状态状态含义典型触发条件submitted任务已提交等待Agent接收处理调用方发送tasks/send后Agent刚收到任务workingAgent正在处理中Agent开始执行任务input-requiredAgent需要更多输入才能继续下游Agent发现缺少必要参数或信息completed任务成功完成产出artifacts所有工作处理完毕结果已生成failed任务执行失败发生不可恢复的错误canceled任务被取消调用方发送tasks/cancelAgent确认取消正常流程下状态按照submitted - working - completed流转。但input-required这个状态特别值得多说两句因为它在Agent协作场景里非常常见。3.2 为什么不建议所有链路都做成“一次到位”很多Agent协作框架喜欢把任务做得“全自动、无人工”。但实际业务场景里下游Agent经常遇到信息不足的情况。我遇到过的一个真实例子一个数据分析Agent要调用一个报表Agent生成月度报告。数据分析Agent只传了“生成5月份报告”这个指令但没有说明报告模板和面向的受众。报表Agent返回了input-required状态列出了自己缺失的两项信息。数据分析Agent收到后补充了“面向管理层、使用简化模板”任务才继续往下走。A2A把input-required设计成合法状态就是在承认一个事实Agent之间的协作不是每次都能靠一个prompt打通。与其强行让Agent猜不如让下游Agent明确表达“我需要什么”。在实际工程中我建议在Agent端显式处理input-required而不是把它当作异常返回失败。当Agent发现自己拿到的信息不足以完成用户需求时应该生成一个结构化反馈列出缺什么、建议怎么补。这样协作链路虽然多了一轮往返但成功率会明显提高。3.3 多轮交互轮询、流式与推送任务状态有两种获取方式。最简单的是轮询——客户端不停地调tasks/get直到任务进入终态。代码简单但在长任务场景下效率不高。更推荐的是流式。A2A里的streaming通常基于SSEServer-Sent EventsAgent在处理任务的过程中持续向客户端推送任务状态更新。这种机制非常适合“生成一段吐一段”的场景比如翻译长文档、逐段总结会议纪要。第三种是push通知。Agent在任务状态变更时主动给客户端配置好的webhook地址发通知。相比SSEpush适合那些处理时间特别长、客户端不想一直挂着连接的任务。我给你的选型建议很简单本地调试、任务秒级返回轮询就行任务在十几秒到几分钟且希望体验流畅上SSE流式分钟级以上或跨网调用用push通知让Agent在完成时主动告诉你。尽量不要一上来就把三种方式全实现。先做轮询跑通了再按需加streaming最后再考虑push。能力太多会增加联调成本。4. 鉴权与信任边界Agent网络里最容易翻车的地方A2A协议的早期版本设计得很酷但有一点很容易被忽略——怎么保证只有该调用你的人能调用你以及Agent之间怎么证明“自己是自己”。4.1 三种鉴权等级怎么选A2A协议规范里提到了几种鉴权思路大致可以分成三类方式适合场景说明无鉴权开放调用本地开发、内网可信环境、测试环境Agent服务不校验调用方身份任何人拿到URL都能调用Bearer Token / 共享密钥半受信环境服务端间调用调用方在请求头携带预共享的TokenAgent校验Token是否合法OAuth2 授权流程跨组织、面向用户授权的场景走标准的授权服务器流程获取scope受限的访问令牌在本地跑Demo时用第一种最省事。但只要你把Agent网络暴露到非完全可信环境就至少要上Bearer Token。原因很简单Agent是有执行力的不是只读接口。别人一旦能调用你的Agent就是免费利用你的算力、数据、甚至对外发消息的能力。4.2 用户授权传递与权限最小化多Agent协作里有一个容易被忽略的细节如果用户的请求要经过多个Agent那下游Agent应该以什么身份执行是用户的身份还是调用方Agent的身份按照我在项目里的经验这里最好遵循“权限最小化”原则。每个Agent只拥有完成自身任务所需的最小权限不要把它上游的完整用户context一路透传下去。比如翻译Agent只需要“读取原文、返回译文”的权限不该拥有用户数据删除权限。A2A协议允许在消息中携带上下文和用户相关信息但在实践中我会非常克制。除非确有必要比如要结合用户偏好做个性化处理否则一律不传用户敏感信息。Agent链越长数据暴露面越大每多一次转发就多一份泄露风险。我在做模拟项目X时一开始图省事让编排Agent把用户原始token直接传给所有下游Agent。后来发现翻译Agent的日志系统会把完整token打出来当场吓出一身冷汗。从那以后凡是跨Agent调用一律只传最小化的用户身份引用具体敏感凭证由专门服务管理。4.3 本地联调时怎么处理鉴权如果你的Agent是本地起服务的先别急着上OAuth2。我推荐的分步策略是这样的本地首版关闭鉴权专注把协议流程跑通内网测试上Bearer Token用环境变量管理密钥别写死在代码里跨网络发布再考虑完整OAuth2流程并且认真设计scope。另外请记住Token的传输一定要走HTTPS否则和明文裸奔没区别。本地可以HTTP远程必须TLS。5. 一个最小可用的多Agent协作Demo光讲概念不给代码总觉得没落地。我这边给你一个可以本地直接跑通的最小Demo两个Agent服务A负责接收“翻译整理”任务B负责做翻译的核心处理。其实严格来说这个Demo是“编排方-执行Agent-翻译能力”的最小链路但你完全可以按这个模式扩展成更多Agent的协作网络。5.1 场景设计我设计了一个“技术文档翻译整理”场景一个“编排Agent”或者叫客户端Agent它收到用户提交的英文文本需要把任务转交给翻译Agent翻译Agent内部是个独立服务它实现A2A协议接口对外暴露一个translate_textskill翻译Agent完成任务后把结果以artifact的形式返回。我用Python写Web框架选择了FastAPI再用一个极其简单的离线词库映射实现“翻译”。这样你不需要申请任何第三方API key也能跑通。项目结构长这样a2a-demo/ ├── agent_a.py # 翻译Agent服务端 ├── client_b.py # 调用方编排方 └── requirements.txt5.2 翻译Agent服务端实现from fastapi import FastAPI, Request, Response import json import uuid import time app FastAPI() # 一个简化版“翻译引擎”实际项目中换成翻译API即可 TRANSLATION_MAP { hello: 你好, world: 世界, agent: 智能体, protocol: 协议, collaboration: 协作 } def simple_translate(text: str) - str: # 对英文文本做简单的单词映射翻译 words text.lower().replace(,, ).replace(., ).split() translated [TRANSLATION_MAP.get(w, w) for w in words] return .join(translated) def build_result(status: str, task_id: str, artifactNone): result { jsonrpc: 2.0, id: str(uuid.uuid4()), result: { task: { id: task_id, status: status, history: [ { role: agent, time: time.strftime(%Y-%m-%dT%H:%M:%SZ, time.gmtime()), parts: [] } ] } } } if artifact: task result[result][task] task[history][0][parts].append({ type: file, name: translated.txt, url: http://localhost:8000/files/translated.txt, content: artifact }) task[artifacts] [ { name: translated.txt, type: text/plain, content: artifact } ] return result app.post(/a2a) async def a2a_endpoint(request: Request): body await request.json() method body.get(method) if method tasks/send: params body.get(params, {}) task_id params.get(taskId, ) message params.get(message, {}) # 提取消息文本 parts message.get(parts, []) full_text for part in parts: if part.get(type) text: full_text part.get(text, ) # 模拟内部Agent执行处理 translated simple_translate(full_text.strip()) # 返回完成状态和结果 return build_result(completed, task_id, artifacttranslated) if method tasks/get: params body.get(params, {}) task_id params.get(taskId, ) return build_result(completed, task_id) return Response( contentjson.dumps({jsonrpc: 2.0, error: {code: -32601, message: Method not found}, id: body.get(id, None)}), media_typeapplication/json )这里我故意没写Cancellation、Streaming和Push先把最核心的tasks/send和tasks/get跑通。这个服务本质上是把协议层接口暴露在/a2a路径上内部逻辑就是普通的函数处理。5.3 调用方编排方实现import requests import json import uuid import time AGENT_URL http://localhost:8000/a2a def send_task(text: str): task_id ftask-{int(time.time() * 1000)} payload { jsonrpc: 2.0, id: freq-{uuid.uuid4()}, method: tasks/send, params: { taskId: task_id, message: { role: agent, sender: agent-orchestrator-01, time: time.strftime(%Y-%m-%dT%H:%M:%SZ, time.gmtime()), parts: [ { type: text, text: text } ] } } } resp requests.post(AGENT_URL, jsonpayload) print(原始响应:, json.dumps(resp.json(), indent2, ensure_asciiFalse)) data resp.json() if result in data: task data[result][task] print(任务状态:, task[status]) if task.get(artifacts): print(翻译结果:, task[artifacts][0][content]) if __name__ __main__: send_task(hello agent, welcome to protocol collaboration)保存后在两个终端分别执行python -m uvicorn agent_a:app --port 8000 python client_b.py如果能看到类似这样的输出说明最小链路已经通了任务状态: completed 翻译结果: 你好 世界 欢迎 到 协议 协作5.4 跑通之后的验证与扩展跑通基础版的顺序是这样的确认服务端日志里有收到tasks/send请求确认客户端能正确解析任务状态和artifacts在服务端加一个tasks/get接口让客户端改用“先send再轮询get”的方式模拟异步长任务扩展Transport层加一个SSE端点把working状态实时推出去。我自己做第一个A2A Demo时从写代码到跑通大概花了一个多小时。大部分时间其实不是花在业务代码上而是在确认“JSON-RPC字段有没有对齐”。所以真心建议你动手时先把两边的请求/响应报文打出来一对一发消息先确认协议层没问题再写业务逻辑。6. 联调实录几个让我反复排查的坑协议的好处是规范统一但联调的时候坑往往藏在“规范没说死”的细节里。这里分享几个我实际遇到过的鬼问题每一个都让我查了不少于半小时。6.1 Agent Card里的URL写成了127.0.0.1最经典的坑。本地起服务时我在Agent Card的url字段写了http://127.0.0.1:8000。单机调试没问题但一旦把客户端挪到另一台机器或者跑在容器里对方拿到这张卡片后怎么都连不上我的Agent。这不是协议问题是基础设施常识问题。但因为它太常见了我还是要提。Agent Card里的地址应该是其他Agent“能访问到”的地址。在本机联调可以写localhost跨机器联调务必写局域网IP或域名容器环境就要用服务名或映射后的公网入口。建议做法把Agent Card里的URL做成环境变量不同环境注入不同值。不要手改太容易漏。6.2 消息ID不唯一导致任务状态错乱这是我在一个模拟项目X里最深刻的教训。当时写客户端代码图省事把JSON-RPC的id写成了固定字符串比如message-001。服务端倒也没报错但任务状态查询返回的结果经常对不上。查到最后才发现服务端的框架用id作为请求关联的键固定ID导致后到的响应覆盖了先到响应的结果。明明两个请求都对返回的数据却张冠李戴。解决很简单客户端每个请求的id都生成唯一值比如UUID。我用的是request_id str(uuid.uuid4())同时taskId也要保证唯一。如果你同时向同一个Agent发多个任务taskId重复会导致Agent端的任务状态互相覆盖。这个基本要靠调用方自己保证。6.3 上下文过大与部分失败A2A消息里的parts是可以携带大量文本的。我在一次做长文档翻译时把整份文档塞进一条消息结果请求体到了好几MB服务端处理超时整个任务失败。后来我调整了策略把大文档拆成多段Agent之间按段传递任务或者先用文件上传把文档放出来消息里只带文件URL让Agent自己去拉取。经验是文本尽量只放“指令和关键信息”大块内容用文件引用代替。协议虽然允许你放很大的text但不代表设计里期待你这么用。用file类型 临时URL的方案既减少了消息体积又天然支持了Agent从远端拉取文件的异步流程。6.4 结果用text带过下游Agent无法结构化消费早期写Demo时我的翻译Agent把结果放在一个text part里“翻译完成结果是……”。下游Agent如果只想拿最终译文还得自己从那句自然语言里抠文本非常蠢。后来改成把结果放在artifacts里用data类型或file类型统一表达下游Agent直接解析task.artifacts[0].content就能拿到干净结果。这就是我在2.3节强调的边界问题实践里极其关键。我的建议是在Agent开发规范里直接约定凡是需要下游继续消费的产出一律进artifacts凡是人看的自然语言反馈才进message parts。这样你每次联调都不用猜对方返回的东西该往哪取。结尾到目前为止A2A协议的实践派思路已经比较完整了从Agent Card建立身份和技能认知到JSON-RPC消息让Agent之间能说话再到任务状态机驱动多轮协作最后加上鉴权边界保护整个网络。这套链路我前后折腾了近两周才从“看得懂协议文档”变成“能跟Agent真的对话”。我个人的体会是A2A的最大价值不在于消息格式多么精致而在于它给Agent协作提供了一个“可测试的沟通基线”。只要每个Agent都遵循同一套消息和状态约定独立开发和部署的Agent就能组成临时的协作网络而不用提前约定彼此的内部API。如果你打算在真实项目里落地我建议不要一上来就照搬整套规范。先只实现tasks/send和tasks/get用轮询跑通一条最小业务链路再逐步补上streaming、push通知和OAuth2。协议是工具不是门面真正让协作变可靠的是你在状态管理和消息设计里的每一次取舍。之后我会再写一篇关于Agent协作中数据流治理和错误恢复策略的内容继续把这块做实。