A2A协议实战：Agent Card与任务状态机全解析

最近圈子里聊得最密的话题除了Agent本身就是“让Agent和Agent怎么互相说话”。大模型能力越来越强单Agent能做的事已经很惊人但当你想把一个写代码的Agent、一个查数据的Agent、一个跑业务的Agent串成一条流水线时问题就来了它们之间的信息格式、任务状态、结果交付方式谁说了算A2AAgent-to-Agent协议就是冲着这个痛点来的。它由Google在2025年4月提出、6月发布1.0版本是一套用于不同Agent之间发现彼此、传递任务、返回结果的开放通信标准。这篇文章我不讲PPT主要结合我实际接入A2A协议、写Agent Card、调试跨Agent任务踩过的坑说说A2A到底怎么用、值得不值得接以及从0到1落地时最容易忽视的细节。适合正在做多Agent系统、Agent编排或者打算让自家Agent对外开放能力的开发者参考。1. A2A协议到底在解决什么问题1.1 这个时代Agent很多但没打通我从2024年开始接触Agent开发从最开始自嗨式的单Agent工具调用到后来尝试多Agent协作一个很深的感受是单Agent的能力上限很多时候不是模型决定的而是工具链和协作方式决定的。一个Agent内部搞几个工具做到“会调函数”很容易但真到企业级场景比如让一个客服Agent把工单转给一个售后Agent再由售后Agent调起财务Agent生成退款单这种跨系统、跨团队、跨技术栈的协作如果全靠自己写API对接每多一个Agent就多一套要维护的接口协议。A2A解决的正是这一层互通问题。它不是让你在进程内调用另一个对象的函数而是定义了一套“两个Agent面对面交谈”的标准流程怎么互相自我介绍、怎么下发任务、怎么回传结果、怎么处理长任务、怎么协商内容格式。它假设每个Agent都是独立运行的实体可能部署在不同机器、不同团队甚至不同公司彼此只通过HTTP通信。这一点和内部RPC、函数调用有本质区别。如果只做单机、单进程的Agent原型A2A确实用不上但只要你的Agent需要“对外开放能力”或者“消费别的Agent的能力”A2A就值得认真考虑。它是目前少有的、有完整规范、有SDK、有1.0正式版本托底的跨Agent通信方案。1.2 A2A和MCP的分工很多人会把A2A和MCP搞混。简单说MCP是把“模型连接工具”这件事标准化它解决的是Agent和工具/数据源之间的互通核心是让模型能通过统一接口读取文件、查数据库、调API。A2A解决的是Agent和Agent之间的互通核心是任务流转和结果交付。两者不在同一层MCP是Agent的“手”A2A是Agent和Agent之间的“对话链”。用一句话记住两者的分工MCP管“Agent怎么用工具”A2A管“Agent怎么找Agent、怎么把事情托付给别的Agent”。如果你的系统里已经有MCP server可以让一个A2A端点内部去调MCP client把工具能力包装成Agent能力对外暴露。我在实际项目里就是这么干的内部工具全部走MCP对外统一用A2A暴露两个协议各管各的互不冲突。1.3 A2A的三个设计目标A2A协议在规范里反复强调三个原则这三个原则直接决定了你写代码时的很多选择。第一是“以任务为中心”。A2A的所有核心交互都围绕Task任务展开Agent之间不是无目的地聊天而是有一个明确的目标比如“帮我查一下这周的销售数据”就是一个Task。整个协议的生命周期、状态机、结果返回都是围绕这个Task设计的。第二是“企业级可用”。规范里自带了企业集成相关的设计比如认证、安全、审计、长时运行任务的支持。它没有把Agent通信做成玩具级的“你发一句我回一句”而是考虑了真实系统里最容易被忽略的部分任务断了怎么办、跑太久了怎么办、权限怎么校验。第三是“开放与中立”。A2A不绑定任何大模型厂商、不绑定任何Agent框架。哪怕你的Agent是OpenAI写的、还是LangGraph搭的、还是自己手搓的Python脚本只要实现A2A规范就能互通。这一点对多厂商混合部署的团队特别重要我见过太多团队被单一框架锁死想换个框架都要重构接口。2. Agent Card智能体的信息名片2.1 Agent Card长什么样Agent Card是A2A体系里最容易被轻视、又最重要的一层。通俗点说它就是每个Agent对外暴露的“自我介绍”我叫什么、能干什么、接受什么输入、返回什么输出、在哪里可以调用、需要什么认证方式。所有想和这个Agent协作的对手方第一步都是先拿到并解析这个JSON文件。一个标准的Agent Card挂在HTTP服务器上的/.well-known/agent.json路径下路径可以自定义但A2A客户端默认会先看这个位置。下面是我实际用过的一个简化版{ name: knowledge-retrieval-agent, description: 企业知识库检索Agent支持基于自然语言查询内部文档、返回带来源引用的摘要, url: https://agent.example.com/a2a, version: 1.0.0, skills: [ { id: knowledge_search, name: 知识检索, description: 输入查询语句检索知识库并返回Top K结果, tags: [rag, search, knowledge-base], examples: [帮我查一下新的报销流程, 找一下服务器部署规范文档] } ], capabilities: { streaming: true, pushNotifications: false }, security: { authentication: { schemes: [ { type: bearer, description: 使用企业SSO签发的Bearer Token } ] } }, defaultInputModes: [text], defaultOutputModes: [text] }这段JSON里skills是最关键的部分。它相当于Agent的能力清单另一个Agent或Agent编排平台会基于skills来决定“这件事该不该托付给你”。我在给团队设计Agent Card时要求每个skill必须写清楚三件事能做什么、典型的输入示例、相关的关键词标签。这三个字段看着简单却直接决定了Agent被发现和匹配的成功率。description写得太虚比如“帮助用户解决各种问题”在生态里基本等于没有描述。除了字段本身一个容易被忽略的点是Agent Card不是写一次就不变的。当你给Agent新增了某个能力一定要同步更新skills列表。我在联调时遇到过好几次“对方Agent说能做这个事但实际调过去发现根本没有对应逻辑”的情况最后追查下来都是因为能力文档和代码实现脱节了。建议把Agent Card当成接口文档的一部分来维护变动时走评审流程而不是让某个开发者默默改一版就完事。2.2 从0.3到1.0Agent Card改了什么A2A在正式发布前经历了多个候选版本社区里讨论最多的就是0.3版本和1.0版本。0.3版本更像一个“能用但还在打磨细节”的候选版。当时Agent Card的字段结构还没有完全定型比如认证信息的表示方式、能力声明的组织方式都还在迭代。我记得0.3时代有团队把security相关字段写在skill下面在1.0里统一被收敛到了Agent整体级别的security字段这样权限语义更清晰一个Agent一套认证方案而不是每个能力一套认证方案。1.0版本把很多字段拧紧了capabilities字段明确区分streaming流式输出、pushNotifications推送通知、stateTransitionHistory状态转换历史记录三类能力每个都是独立布尔值方便客户端精确判断该用哪种交互方式。security字段从“参考建议”变成了“显式声明”。Agent必须明确告诉调用方它支持哪种认证方案比如bearer、oauth2、apiKey等。这让生产环境下的握手变得更可预期。skills增加了更一致的inputModes、outputModes声明方式并且1.0对超时、重试、错误码也给出了更明确的语义约定。实际上从0.3切到1.0代码层面的迁移成本并不大主要工作是调整Agent Card JSON里的字段路径。如果你现在才开始接入直接用1.0就行如果手头有基于0.3的存量实现重点检查security、capabilities、skills三块的字段映射即可。不要被版本号吓到A2A的兼容性设计比很多企业级中间件良心得多。2.3 编写Agent Card的实操建议写Agent Card有几点经验踩过坑才会在意。第一url字段一定要填Agent实际接收请求的端点而不是Agent Card所在路径。这个错误我在早期犯过把Agent Card的地址填成了A2A请求端点结果对方Agent拿到卡片后把任务消息POST到了一个只返回JSON介绍文件的路径上直接导致415。url必须指向实现A2A方法如tasks/send的HTTP端点。第二skill的examples不要随便写。A2A生态里的Agent发现服务会拿这些示例做语义匹配也会把它们展示在选择界面上。我自己写的时候会把真实用户问过的高频问题放进去而不是编造几个“标准问法”。真实高频问题往往口语化反而比精心设计的问法更能帮助匹配。第三安全字段尽早声明。即使你的Agent暂时只在内部网络跑也把security.authentication.schemes写了哪怕只是{type: none}。这样后面接入OAuth时不至于因为格式变更返工客户端也拿得到明确的鉴权预期。还有一个很值钱的习惯用jq快速校验Agent Card格式或者写一个简单的测试脚本定期请求自己Agent的/.well-known/agent.json确认URL可达、字段完整。因为Agent Card是别人接入你Agent时最先看到的东西它挂了你的Agent在生态里就相当于不存在。我在CI里加了一个简单的检查任务一旦agent.json字段不符合规范立刻报警省掉了很多下游同事的“为什么连不上”的抱怨。3. 核心机制拆解任务、消息与三种交互模式3.1 任务生命周期与状态机A2A协议里最核心的概念就是Task。一个Task代表一次完整的Agent间协作比如“请帮我生成一份项目周报”就是一个Task。这个Task会经历从创建、运行到完成或取消的完整生命周期。在1.0规范里Task的核心状态包括submitted任务已提交等待Agent处理。workingAgent正在执行中。input-requiredAgent需要调用方补充更多信息才能继续。completed任务成功结束结果放在artifacts产物里。canceled任务被取消。failed任务执行失败error字段里会携带失败原因。我在开发时最喜欢做的类比是Task状态机和现实中的工单系统几乎一一对应。你提交一个工单先受理submitted然后处理中working处理过程中客服问你补充资料input-required最后办结completed或退回failed。如果你做过工单系统A2A的任务状态机基本不需要重新学。值得注意的坑是input-required这个状态。很多第一次实现A2A的开发者会把“Agent需要补充信息”理解成“直接把信息放在一条新消息里发过去就行”。在A2A里不是这样你需要在同一个Task下继续发送消息而不是新建一个Task。一个Task内的多轮消息追加才是正确的追问方式。我在接入第一个跨Agent场景时就因为这里理解错了导致对方Agent连续创建了三个新Task业务上下文完全对不上。3.2 Message与Part的结构Task之下是MessageMessage之下是Part。这个三层结构是A2A消息模型的核心。一个Message代表一轮对话或一次状态更新比如“用户问了一个问题”是一条Message“Agent回答了一部分”也是一条Message。每个Message都带role字段用来区分是谁发的user、agent还是system。每个Message的parts里可以放多个PartPart是实际内容单元。1.0规范里常见的有三类TextPart纯文本最常用。FilePart文件内容比如PDF、图片、表格可以带URL或字节数据。DataPart结构化数据比如JSON对象。这对于Agent之间传递业务对象极其有用比如传一个订单对象、一个用户实体直接用DataPart不必为了文本化把JSON硬塞进字符串再靠正则解析出来。Part化设计的直接好处是Agent之间可以混着传图文和结构化数据调用方可以精确感知“哪一段内容是文本、哪一段是文件、哪一段是结构化业务数据”而不是像传统聊天流一样只能看到一串渲染后的字符串。我第一次用DataPart传业务对象时明显感觉跨Agent的数据解析工作量下降了至少一半。我放一个Message的简化例子便于理解{ messageId: msg-01, role: agent, parts: [ { type: text, text: 查询结果如下 }, { type: data, data: { total: 128, items: [ { id: P1001, name: 项目A } ] } } ] }这里的data字段是自由结构的JSON具体结构由双方Agent协商。规范不限定你传什么东西但强烈建议用一个双方都能理解的schema最好在Agent Card或双方文档里写清楚避免数据语义对不上。3.3 拉取、推送与流式三种交互方式A2A协议支持三种交互方式它们的差别主要体现在长任务处理上。拉取模式Polling客户端调用tasks/send创建任务后拿到的可能只是一个任务ID然后轮询tasks/get查询状态直到状态变成completed。这个模式实现最简单适合任务执行时间在几秒内的场景。推送模式Push NotificationAgent在执行中通过Webhook通知客户端“我更新了状态”客户端再主动拉取结果。这个模式适合任务执行时间较长、轮询浪费严重的场景。Agent Card里的capabilities.pushNotifications就是告诉调用方支不支持这种模式。流式模式Streaming客户端通过SSEServer-Sent Events从Agent端持续接收消息流不用反复请求。这个模式适合对话式、流式生成的场景比如让Agent边写代码边把输出推送过来体验最接近ChatGPT打字机式输出。我在实际项目里的选择标准很简单任务3秒内能完成的用拉取就够任务超过10秒的优先考虑推送如果是对话或码农场景直接走流式。你不用把三种都实现但必须在Agent Card里如实声明能力否则对方Agent按错误方式调用轻则体验差重则直接把提交任务当接口卡住超时。声明清楚之后由调用方来选择它支持的交互方式这个协商过程本身就是A2A智能的地方。4. 手把手接入A2A一个可复现的最小实现4.1 环境准备与目录结构这一章我直接拿Python为例。选Python不是因为A2A只支持Python而是因为A2A官方SDKa2a-sdk生态最成熟社区示例也多。用其他语言也一样核心是HTTPJSON。我建议的最小实现包含四个文件一个目录整体结构是这样的a2a-demo/ ├── agent_card.py # 返回Agent Card的端点 ├── task_manager.py # 任务状态管理 ├── agent_service.py # 具体业务逻辑这里是模拟生成摘要 ├── server.py # FastAPI入口注册所有A2A端点 └── client_demo.py # 另一个Agent发起调用的演示脚本依赖就用FastAPI加uvicorn再加a2a-sdk里的类型定义。安装命令pip install fastapi uvicorn a2a-sdk如果网络环境装不了官方SDK也没关系A2A本质就是HTTPJSON自己手写请求处理也不难。下面我给的是基于FastAPI、偏手写的实现好处是你可以清楚地看到每个端点在干什么不至于被SDK封装“包”住。我在调试协议细节时反而更喜欢看裸的请求和响应。4.2 实现Agent Card入口先写Agent Card。这个端点最无脑就是把一个JSON文件返回出去。但我建议单独起一个路由并且用CI校验它的正确性避免改业务代码时不小心把字段改坏了。from fastapi import FastAPI app FastAPI() AGENT_CARD { name: demo-summary-agent, description: 接收一段长文本返回不超过3句话的摘要, url: http://localhost:8000/a2a, version: 1.0.0, skills: [ { id: summarize, name: 文本摘要, description: 输入一段文本返回简洁摘要, examples: [帮我把这篇产品说明总结一下] } ], capabilities: { streaming: False, pushNotifications: False, stateTransitionHistory: True }, security: { authentication: { schemes: [{type: none}] } }, defaultInputModes: [text], defaultOutputModes: [text] } app.get(/.well-known/agent.json) async def agent_card(): return AGENT_CARD这里我把url指向了http://localhost:8000/a2a这就是后面实现A2A方法的地方。如果你想部署到线上记得把localhost换成正向代理地址并且让这个地址可以被其他Agent从公网访问到。Agent Card如果只在本地可见其他Agent无论如何都连不上你。4.3 实现Task执行端点接下来是重点。A2A协议在1.0版本里的核心方法包括tasks/send创建一个Task并开始执行。tasks/get按ID查询任务状态。tasks/cancel取消任务。message/send在已有Task里追加消息。我不打算把一堆SDK魔法堆上来我直接写一个最朴素的使用方视角客户端发来一个JSON我解析出里面的taskId、message.parts然后执行我的业务逻辑最后返回一个Task对象。这里为了控制篇幅我实现tasks/send和tasks/get两个端点就够演示了。import uuid from fastapi import FastAPI, Request app FastAPI() # 内存任务存储演示用生产环境请换成Redis或数据库 TASKS {} app.post(/a2a/tasks/send) async def tasks_send(request: Request): payload await request.json() task_id str(uuid.uuid4()) message payload.get(message, {}) parts message.get(parts, []) # 提取用户文本 user_text for part in parts: if part.get(type) text: user_text part.get(text, ) # 模拟业务处理这里可以接任意Agent逻辑 result_text summarize_text(user_text) # 构造任务对象 task { id: task_id, status: completed, messages: [ { messageId: str(uuid.uuid4()), role: agent, parts: [ {type: text, text: result_text} ] } ], artifacts: [ { name: summary, parts: [ {type: text, text: result_text} ] } ] } TASKS[task_id] task return task app.post(/a2a/tasks/get) async def tasks_get(request: Request): payload await request.json() task_id payload.get(id) if task_id not in TASKS: return {error: {code: -32602, message: task not found}} return TASKS[task_id] def summarize_text(text: str) - str: # 示例直接截断前50个字作为“摘要”真实场景接LLM或RAG return text[:50] (... if len(text) 50 else )这段代码没有用官方SDK但完全实现了A2A最基本的“同步任务”方式客户端发来一条消息服务端当场执行完并返回completed状态的Task。如果你想支持长任务把返回时的状态改成working再另起线程执行然后把tasks/get做成能查到最新状态的版本即可原理完全一样。代码里有个细节我把artifacts产物和messages都放进了Task。这是A2A的一个特点它把“对话消息”和“任务成果”分开存放。别的Agent要拿结果优先看artifacts而不是翻消息记录。这个设计对做数据处理流水线特别友好任务完成后的产物可以直接被下一个Agent消费没必要再解析一轮聊天记录。4.4 在另一个Agent里调用服务端写好了客户端怎么调简单起见我直接用httpx发起请求。下面这段是一个“客户端Agent”想调用“摘要Agent”的示例import httpx # 1. 获取Agent Card card httpx.get(http://localhost:8000/.well-known/agent.json).json() print(发现Agent:, card[name]) print(能力:, [s[name] for s in card[skills]]) # 2. 发起任务 send_payload { jsonrpc: 2.0, id: 1, method: tasks/send, params: { message: { role: user, parts: [{type: text, text: 这是一段很长的产品文档内容请帮我总结一下。}] } } } resp httpx.post(http://localhost:8000/a2a, jsonsend_payload).json() task resp.get(result, resp) print(任务状态:, task[status]) # 3. 读取产物 for artifact in task.get(artifacts, []): for part in artifact[parts]: print(摘要结果:, part[text])注意官方SDK封装了JSON-RPC请求的细节这里手写的method、params是为了让你看清协议到底传了什么。真实项目里直接用官方SDK更省心但理解裸协议能帮你更快排查问题。另外A2A的路由定义我建议统一/a2a接收所有方法POST/.well-known/agent.json给Agent Card。message/send也走同一个URL对应method: message/send。不用为每个方法单独建路由JSON-RPC本身已经能区分方法名。这个设计一开始会有点不习惯但用熟之后反而觉得干净所有入口都在一个地方。5. 生产落地中的常见问题与避坑速查5.1 我踩过的几个坑第一个坑Agent Card的url填错了。这个前面提过我这里说具体一次经历。当时我把Agent Card里的url填成了https://xxx.com/.well-known/agent.json结果客户端拿到Card后把任务消息POST到了这个路径那个路径的处理器只返回JSON文件内容不处理POST请求直接报405。排查了半天直到我人工敲了一次HTTP请求才发现是url指向的错误。这个错特别低级但特别容易发生尤其当Agent Card文件是通过配置中心动态生成、而url字段是从环境变量拼接来的时候。第二个坑任务状态更新不及时。最开始实现长任务时我直接在同步函数里跑了一个耗时的模型推理导致tasks/send请求卡住十几秒客户端早就超时了。正确做法是让tasks/send立即返回working状态把耗时逻辑丢到后台队列或线程池去跑再通过更新任务存储让tasks/get返回最新状态。如果追求实时体验再叠加SSE推送。这一条是A2A落地和生产化的核心区别很多demo项目能跑通但一上生产就超时基本都栽在这里。第三个坑没有处理input-required状态。一个真正可用的Agent在处理复杂任务时经常需要向调用方要更多信息。比如“你要我查销售数据但没说是哪个区域”。如果实现里遇到这种情况直接报failed调用方体验会很差。规范里专门设计了input-required状态就是让你把“追问”也做成一种正常的任务状态。这块需要和业务结合设计好不是简单返回一个状态而是要约定追问的问题怎么结构化地传递给调用方比如放在该Task下的message里由调用方后续再提交一条带答案的message。5.2 常见错误排查速查表我把一些A2A接入过程中最常见的错误整理成一张速查表按“症状-原因-处理”三条线查调试时会快很多。症状可能原因排查方法找不到Agent Card路径不是/.well-known/agent.json服务器未暴露公网先用curl请求该路径检查反向代理是否过滤了带.well-known的路径请求返回404Agent Card中url指向的A2A端点路径不存在核对Card中的url与server中注册的POST路径是否一致返回401/403security鉴权配置和客户端对不上检查Card中security.schemes确认Bearer Token有效tasks/send超时长时间任务在同步函数里执行改为后台执行先返回working再用tasks/get或SSE获取结果任务总停在working没有更新任务状态存储检查后台任务是否真的跑完确认任务状态写入是否成功JSON解析报错客户端和服务端的Part结构不一致检查Part的type字段和对应payload字段是否匹配这张表我是在真实联调中一点点攒下来的。如果你刚开始接触A2A建议直接把这张表贴到团队协作文档里能帮队友省下不少“为何连不上”的排查时间。另外补充一个调试技巧在本地联调时可用ngrok之类的隧道工具把本地Agent端点临时暴露成公网地址让远程Agent能够访问到。但要注意这类隧道工具只适合开发调试别直接用于生产环境稳定性无法保证。5.3 上线前建议做好的几件事第一给Agent加认证和鉴权。A2A协议本身不强制认证方式但生产环境绝对不能用security: {type: none}。哪怕是内部系统也建议用OAuth2或API Key做一层身份校验。如果Agent对外开放接口给第三方Agent调用更要严格校验调用方身份否则你的Agent能力会被别人免费“借用”还有数据泄露风险。第二给每个Task加超时和重试策略。Agent之间的调用链路往往很长A2A协议不会自动帮你做超时重试。你要在调用方设置合理的超时阈值比如同步任务5秒、异步任务至少30秒对失败任务做有限次数的重试建议最多3次并做好幂等处理防止同一个任务被重复提交。Task id的设计就是天然的幂等键重试时复用同一个task id即可。第三做好审计日志。Agent之间协作一旦出问题没有日志几乎没法查。我在生产环境里要求每个A2A请求必须记录来源Agent标识、初始任务ID、状态变化时间线、最终产物摘要。这样即使跨团队协作出了事故也能快速定位是哪个环节掉了链子。第四容量和限流。你暴露的Agent端点如果不做限流被多个Agent同时调用很快会打挂。常用的方案是令牌桶限流按调用方身份区分配额。我在网关层给每个外部Agent分配了独立的QPS配额最高优先级的协作Agent可以跑到50 QPS普通对外能力只给10 QPS。配额按Card里声明的name来标识方便对账。我这段时间高强度用过A2A之后最大的感受是它不是一个让你“很兴奋”的协议而是那种“越用越觉得稳”的基建型协议。它没有许诺什么AI乌托邦只是老老实实把Agent之间“怎么介绍自己、怎么交任务、怎么还结果”这件事定清楚了。如果你正打算做多Agent编排或者想把自己的Agent能力开放出去给同事的Agent调用A2A值得从今天开始试一下。动手时先别急着上框架把Agent Card写明白、把Task状态机理顺后面的一切都会顺很多。这是我在反复联调中体会最深的一句话。