几个月前我接手一个内部系统需要让两个 AI Agent 互相传话。一个负责解析订单信息另一个负责查库存两个 Agent 用着完全不同的框架连消息格式都各说各话。第一版我用 requests 直接调对方接口每接一个 Agent 就要写一套适配代码参数映射、错误处理、超时重试全都手搓实在是烦。后来换成了 a2a-protocol 这个 Python 包总算是把这件事从“治标”变成了“治本”。如果你也在做多 Agent 协作或者准备把 Agent 服务暴露给第三方那么这篇内容应该能帮你少走不少弯路。a2a-protocol 是 Agent2AgentA2A协议的 Python 实现。它定义了一套标准的“名片、消息、任务”模型以及一套基于 JSON-RPC 的传输约定让不同 Agent 之间可以互相发现、互相调用不需要关心对方底层到底用的什么框架。这篇文章我会从协议背景讲起然后拆解包里最常用的几个类和参数最后用一个日志分析 Agent 加告警 Agent 的实战案例告诉你它到底怎么落地。1. A2A协议出现的背景Agent之间为什么需要统一的“普通话”1.1 从REST接口到Agent协议的距离早几年我们做 AI 集成思路很朴素Agent 就是个带业务逻辑的 HTTP 接口。你要调用它就给它发一段 JSON等它回一段 JSON。问题在于这种接口是“为特定场景定制”的订单 Agent 的/create_order和库存 Agent 的/check_stock它们的参数、返回值、错误语义完全是两套体系。我把这两个 Agent 接起来之后光是字段映射就写了 100 多行代码。后来再加第三个 Agent又要再来一遍。更麻烦的是对方 Agent 升级了接口我会直接挂掉。这种 Ray 式集成显然不应该是多 Agent 时代的主流。A2A 协议想做的事情很简单定义一个统一的“客户端-服务端”契约。 无论 Agent 内部逻辑有多复杂对外都表现为一个标准端点。有人给这个端点发一个标准格式的任务消息端点返回标准格式的任务状态和结果。这样一来不同团队的 Agent 只要都实现同一个协议就能直接互相调用。1.2 a2a-protocol在设计上偷学了Web的哪些招第一次看 A2A 协议的文档你会发现它非常眼熟。它做三件事能力发现、任务投递、结果获取。这几乎就是 Web 那套思路搬到了 Agent 之间。能力发现对应的是 AgentCard。每个 Agent 对外发布一张“名片”名片里写清楚自己叫什么、能干什么、端点在哪个 URL。其它 Agent 拿到名片就能判断“这事该不该找它”。这有点像 Web 里的 robots.txt 加上 OpenAPI 描述只不过描述对象变成了 Agent 能力。任务投递和结果获取对应的是 Task 和 Message。客户端先创建一个 Task把消息放进去发给服务端服务端更新 Task 状态从 submitted 变成 working最后变成 completed同时把结果作为新的 Message 或 artifact 写进去。整个流程本质上是异步任务队列只不过传输层用了 HTTP JSON-RPC。这种设计非常务实Agent 的处理经常要几秒甚至几分钟同步等待不现实所以协议把任务状态机作为一等公民客户端可以轮询也可以等服务端回调。我自己的体会是A2A 并不适合所有场景。如果你要做低延迟高频的流式交互比如实时语音 Agent 每一帧都要通信那直接用 WebSocket 更合适。A2A 的目标是“Agent 之间异步协作”是为后端服务设计的。弄清楚了这一点再看后面的代码就不会觉得别扭。2. 装包和第一印象a2a-protocol 的核心对象与基础语法2.1 安装与导入留意版本差异安装很简单直接用 pippip install a2a-protocol我目前用的版本顶层包名可以这样导入from a2a_protocol import AgentCard, Message, Task, TaskStatus from a2a_protocol.types import TextPart, MessageRole需要提醒一句A2A 协议还比较年轻版本更新时偶尔会用TaskStatus.WORKING偶尔会用TaskStatus.PROCESSING甚至不同小版本里Message的字段名也会有变化。建议你装完之后用python -c import a2a_protocol; print(a2a_protocol.__version__)确认一下版本再对照官方文档看细节。安装包里默认依赖 pydantic所以这些核心类基本都是 pydantic 模型可以直接用model_dump()序列化成 JSON。这一点很关键因为后续不管是写服务端还是客户端我都靠它把对象转成协议数据。2.2 AgentCard、Message、Task 三位一体刚接触这个包的时候最容易混淆的就是这三个类到底分别管什么。我用一个本地生活例子来解释AgentCard是“简历”。它描述 Agent 的身份和能力让别的 Agent 知道你是谁、能接什么活。Message是“社交软件里的会话消息”。它包含谁说的、说了什么、属于哪个会话。Task是“工单系统里的任务单”。它把一条或多条消息包装在一起记录这个任务从创建到完成的状态。比如我定义一个“日志分析 Agent”它的简历长这样card AgentCard( namelog-analysis-agent, description分析日志文本提取错误级别和摘要, urlhttp://localhost:8000/a2a, version1.0.0, )然后我构造一条给它看的消息message Message( message_ide1b8f2c4-2a77-4a1e-b7be-9b2e5a6d7f00, roleMessageRole.USER, content[TextPart(text2025-06-01 10:00:00 ERROR timeout connecting to db)], )最后把这个消息包成一个任务task Task(idtask-0001, statusTaskStatus.SUBMITTED, messages[message])这段代码基本就是 a2a-protocol 的“最小骨架”。你会发现它没有很多黑魔法只是把 Agent 通信里的关键信息拆成了三个可序列化的数据类。2.3 一条消息从发送到完成的完整链路理解了这个链路的生命周期后面写代码才能不失控。一个典型任务长这样客户端构造一个Task里面放一个 role 为 user 的Message。客户端把这个 Task 序列化成 JSON-RPC 请求POST 到服务端的/a2a端点。服务端创建同名 Task但状态改成WORKING同时返回给客户端。Agent 开始处理处理过程中可以追加消息当它想把中间结果发出来时就加一条 role 为 agent 的消息。处理完成后Task 状态变成COMPLETED结果放在新增的 Message 或artifacts数组里。客户端通过轮询拿到最终 Task 对象从中取最新消息或产出物。链路里最关键的点是Task 是唯一的状态载体Message 是附属于 Task 的通信内容。你不需要单独维护一个“会话状态”因为状态就在 Task 上。3. 参数拆解我要配置哪些字段每个字段有什么坑3.1 AgentCard 参数让其他 Agent 能找到你AgentCard是别人了解你 Agent 的唯一入口字段不多但每个都很重要。我整理了一份常用参数表参数是否必填含义我踩过的坑name是Agent 唯一标识一定要全局唯一多个 Agent 用同一个名字会让调用方混乱description是能干什么、边界是什么不能写“万能助手”最好写清输入输出例如“输入错误日志输出 JSON 摘要”url是A2A 服务端点填localhost会导致跨容器、跨机器调用失败要用可解析的地址version否版本号升级接口时建议带上版本capabilities否是否支持流式、推送通知等如果声明了push_notifications但没有实现回调对方会傻等authentication否认证方式共享密钥或 Bearer Token别把密钥写死在简历里default_input_modalities否默认输入类型比如 text 或 file声明了就得真的支持解析文件在代码里我通常这样构造卡片from a2a_protocol import AgentCard, Authentication card AgentCard( namealert-agent, description接收日志分析结果生成告警消息, urlhttp://alert-agent:8000/a2a, version0.2.0, capabilities{streaming: True, push_notifications: False}, authenticationAuthentication(schemes[bearer], credentialstoken-xxx), )注意capabilities是一个字典布尔值千万不要乱写。我第一次就把push_notifications写成了 True结果客户端一直没收到回调任务卡在 working 状态直到超时。后来改成 False客户端才知道要走轮询。3.2 Message 和 Part 参数内容怎么装才不会丢Message本身不是一个字符串它的content是Part对象的列表。这一点新手最容易翻车。常见的Part有TextPart普通文本字段是text。FilePart文件引用字段是file和mime_type。DataPart结构化 JSON 数据字段是data。构造消息时尽量用显式类型from a2a_protocol.types import DataPart msg Message( message_ida1b2..., roleMessageRole.USER, content[ TextPart(text分析以下日志), DataPart(data{lines: 100, source: application.log}), ], )role字段也很关键。标准里大概有user和agent两种个别版本还可能出现system。它可以理解为“这句话是任务发起人说的还是 Agent 回复的”。服务端判断任务是否处理完一般会看最新一条消息是不是 role 为 agent 的消息所以要保证角色写对。另外还有两个可选参数值得注意parent_message_id如果要回复前一条消息这里填前一条的 message_id。多轮对话全靠它串成一条链。metadata一个自由字典。我经常在里面放agent_id、trace_id方便链路追踪。3.3 Task 参数与状态流转异步任务的精髓Task是最容易被忽略参数坑的对象。它的状态不是随便填的协议规定了几种我在实际项目里经常被这几种子状态卡住状态含义何时出现SUBMITTED任务已创建客户端刚发出来WORKING处理中Agent 开始干活INPUT_REQUIRED需要更多输入Agent 发现信息不足COMPLETED已完成结果写入 messages 或 artifactsFAILED失败异常或逻辑错误CANCELED已取消客户端主动取消Task里最重要的参数自然是messages和artifacts。messages保存所有对话消息artifacts保存最终产物比如生成的文件、JSON 数据。一个字一个字地构造 Task 容易出错所以我通常会用一个工厂函数def create_task(content: str, task_id: str) - Task: msg Message( message_idstr(uuid.uuid4()), roleMessageRole.USER, content[TextPart(textcontent)], ) return Task(idtask_id, statusTaskStatus.SUBMITTED, messages[msg])注意一个细节Task的messages即使是初始消息也得放进列表里不能直接传单个对象。协议要求它是数组。4. 实战让日志分析 Agent 和告警 Agent 自动协作4.1 场景设计为什么选这个案例理论讲多了容易飘还是看一个能跑起来的例子。我这次选的是运维领域最常见的场景一个日志分析 Agent专门从原始日志里提取错误级别和摘要一个告警 Agent接收分析结果生成一条上游系统能识别的告警消息。为什么拆成两个 Agent 而不是写成一个因为日志分析和告警策略分别由两个团队维护他们的发布节奏不一样。用 A2A 拆开之后日志分析团队升级模型不影响告警 Agent告警 Agent 修改规则也不影响日志分析的输出。这就是 Agent 协作的价值。拓扑上我们假设客户端只连接日志分析 Agent日志分析 Agent 内部调用告警 Agent。整个过程走完客户端会拿到一个最终的告警结果。4.2 服务端 A日志分析 Agent我用 FastAPI 搭了一个最小服务端。a2a-protocol 的类可以很方便地转成 JSON-RPC 响应。import json import uuid from datetime import datetime from fastapi import FastAPI, Request from a2a_protocol import AgentCard, Task, TaskStatus from a2a_protocol.types import TextPart, DataPart, MessageRole, Message app FastAPI() CARD AgentCard( namelog-analysis-agent, description分析日志文本提取级别和摘要, urlhttp://localhost:8001/a2a, version1.0.0, ) def analyze_log(text: str) - dict: if ERROR in text: level high summary 数据库连接超时 elif WARN in text: level medium summary 连接池使用率偏高 else: level low summary 无异常 return {level: level, summary: summary, raw_length: len(text)} app.get(/.well-known/agent-card.json) async def agent_card(): return CARD.model_dump() app.post(/a2a) async def handle(request: Request): payload await request.json() method payload.get(method) message_id payload.get(id, 1) if method tasks/send: task_data payload.get(params, {}).get(task, {}) task Task.model_validate(task_data) task.status TaskStatus.WORKING latest_content task.messages[-1].content[0].text result analyze_log(latest_content) reply Message( message_idstr(uuid.uuid4()), roleMessageRole.AGENT, content[DataPart(dataresult)], task_idtask.id, parent_message_idtask.messages[-1].message_id, ) task.messages.append(reply) task.status TaskStatus.COMPLETED task.artifacts.append(result) return {jsonrpc: 2.0, id: message_id, result: task.model_dump()} return {jsonrpc: 2.0, id: message_id, error: {code: -32601, message: method not found}}这段代码里最有用的设计是我始终把业务逻辑和协议逻辑分开。analyze_log只是普通函数真正的 A2A 交互发生在handle里。这样后续你想从 FastAPI 换成别的 ASGI 框架业务代码不用动。注意Task.model_validate(task_data)是 pydantic v2 的标准用法如果你用的是旧版可能要改成parse_obj。4.3 服务端 B告警 Agent告警 Agent 的代码结构完全一样只是业务逻辑不同。它不要求返回复杂日志只需要从DataPart里读取 JSON然后生成告警文案。app.post(/a2a) async def alert_handle(request: Request): payload await request.json() method payload.get(method) message_id payload.get(id, 1) if method tasks/send: task Task.model_validate(payload[params][task]) latest task.messages[-1].content[0] # 如果是 DataPart 才正常解析 if hasattr(latest, data): analysis latest.data else: analysis {level: unknown, summary: latest.text} alert_text f[{analysis[level]}] {analysis[summary]} reply Message( message_idstr(uuid.uuid4()), roleMessageRole.AGENT, content[TextPart(textalert_text)], task_idtask.id, ) task.messages.append(reply) task.status TaskStatus.COMPLETED task.artifacts.append({alert: alert_text}) return {jsonrpc: 2.0, id: message_id, result: task.model_dump()} return {jsonrpc: 2.0, id: message_id, error: {code: -32601, message: method not found}}两个服务端除了 AgentCard 里的名字、地址不同协议层的代码几乎一模一样。这其实就是 A2A 最大的价值你写一次协议层所有 Agent 都能复用。4.4 客户端一次性拉通客户端这边我用httpx来发请求没有用底层 against 一些复杂的 SDK因为这样能看到整个协议的样子。import httpx import uuid from a2a_protocol import Task, TaskStatus from a2a_protocol.types import Message, MessageRole, TextPart def create_task(text: str) - Task: msg Message( message_idstr(uuid.uuid4()), roleMessageRole.USER, content[TextPart(texttext)], ) return Task(idstr(uuid.uuid4()), statusTaskStatus.SUBMITTED, messages[msg]) def send_task(url: str, task: Task) - Task: payload { jsonrpc: 2.0, id: 1, method: tasks/send, params: {task: task.model_dump(modejson)}, } with httpx.Client(timeout30) as client: resp client.post(url, jsonpayload) resp.raise_for_status() result resp.json()[result] return Task.model_validate(result) if __name__ __main__: client_task create_task(2025-06-01 10:00:00 ERROR timeout connecting to db) final_task send_task(http://localhost:8001/a2a, client_task) for msg in final_task.messages: print(msg.role, msg.content[0]) print(artifacts:, final_task.artifacts)客户端整个流程没有碰任何 HTTP 细节只和Task打交道。服务端的地址也可以从 AgentCard 的url字段动态获取甚至可以设计一个 Agent 注册中心客户端先查卡片再调任务这就是 A2A 的完整服务发现语义。5. 接入过程中我不吐不快的坑以及几条优化建议5.1 四个我实际遇到的坑这个包我用了一个多月整体很顺手但有几个坑确实让我熬夜调过。第一个坑是 AgentCard 里的url地址。我在本地开发时填了http://localhost:8000/a2a结果一部署到 Docker 里另一个容器根本访问不到这个 localhost。后来我在配置里用服务名或者环境变量动态生成。第二个坑是 Message content 类型。我一开始图省事直接给content传了一个字符串Message(message_idx, roleuser, contenterror)序列化出来 content 是字符串但接收方按列表解析直接抛异常。正确做法永远是content[TextPart(texterror)]。第三个坑是任务状态枚举的兼容性。不同版本的 a2a-protocol 对“处理中”的叫法不一样有的叫WORKING有的叫WORK_IN_PROGRESS。如果你的系统里同时跑着多个客户端和服务端最好在构造请求和解析响应时都做一层状态映射。第四个坑是 metadata 参数。我在里面放了一个datetime对象结果调用model_dump(modejson)时直接报错因为 datetime 不是 JSON 原生类型。规范的做法是提前转成 ISO 字符串或者统一用时间戳。5.2 让 a2a-protocol 用得更顺的进阶配置如果你确定要用这个包做正式项目我建议你做三件事。第一把 AgentCard 放在一个公共配置模块里所有服务端启动时都从这里加载。这样不会出现几个 Agent 的卡片描述与真实能力不一致。第二为每个任务生成稳定的 UUID 作为任务 ID这样客户端重试时可以幂等服务端如果发现同一个 task_id 已经存在可以直接返回已有任务不重复执行。第三为任务增加metadata.trace_id把 A2A 交互日志和业务日志串起来排查问题时能省大量时间。我在生产环境里还加了一个保险客户端轮询 Task 时如果一段时间内状态没有变化就主动取消任务并告警。这不是 a2a-protocol 自带的功能但配合协议的任务状态机做起来非常容易。最后分享一个小技巧。如果你想让日志分析 Agent 调用告警 Agent在第一个 Agent 的 handler 里直接使用httpx调用第二个 Agent 的/a2a即可不需要引入额外的编排框架。协议本身就是为这种嵌套调用设计的一个 Agent 既可以当客户端也可以当服务端。我实际项目里就是让订单 Agent 动态调用了库存 Agent整个过程和上面案例里的客户端调用方式一模一样。这种组合方式非常灵活也是 A2A 协议最吸引我的地方。 
长河君 · 2026/10/8 9:26:17 · 阅读 1.2k 用 a2a-protocol 实现多Agent协作：从接口适配到协议标准
几个月前我接手一个内部系统需要让两个 AI Agent 互相传话。一个负责解析订单信息另一个负责查库存两个 Agent 用着完全不同的框架连消息格式都各说各话。第一版我用 requests 直接调对方接口每接一个 Agent 就要写一套适配代码参数映射、错误处理、超时重试全都手搓实在是烦。后来换成了 a2a-protocol 这个 Python 包总算是把这件事从“治标”变成了“治本”。如果你也在做多 Agent 协作或者准备把 Agent 服务暴露给第三方那么这篇内容应该能帮你少走不少弯路。a2a-protocol 是 Agent2AgentA2A协议的 Python 实现。它定义了一套标准的“名片、消息、任务”模型以及一套基于 JSON-RPC 的传输约定让不同 Agent 之间可以互相发现、互相调用不需要关心对方底层到底用的什么框架。这篇文章我会从协议背景讲起然后拆解包里最常用的几个类和参数最后用一个日志分析 Agent 加告警 Agent 的实战案例告诉你它到底怎么落地。1. A2A协议出现的背景Agent之间为什么需要统一的“普通话”1.1 从REST接口到Agent协议的距离早几年我们做 AI 集成思路很朴素Agent 就是个带业务逻辑的 HTTP 接口。你要调用它就给它发一段 JSON等它回一段 JSON。问题在于这种接口是“为特定场景定制”的订单 Agent 的/create_order和库存 Agent 的/check_stock它们的参数、返回值、错误语义完全是两套体系。我把这两个 Agent 接起来之后光是字段映射就写了 100 多行代码。后来再加第三个 Agent又要再来一遍。更麻烦的是对方 Agent 升级了接口我会直接挂掉。这种 Ray 式集成显然不应该是多 Agent 时代的主流。A2A 协议想做的事情很简单定义一个统一的“客户端-服务端”契约。 无论 Agent 内部逻辑有多复杂对外都表现为一个标准端点。有人给这个端点发一个标准格式的任务消息端点返回标准格式的任务状态和结果。这样一来不同团队的 Agent 只要都实现同一个协议就能直接互相调用。1.2 a2a-protocol在设计上偷学了Web的哪些招第一次看 A2A 协议的文档你会发现它非常眼熟。它做三件事能力发现、任务投递、结果获取。这几乎就是 Web 那套思路搬到了 Agent 之间。能力发现对应的是 AgentCard。每个 Agent 对外发布一张“名片”名片里写清楚自己叫什么、能干什么、端点在哪个 URL。其它 Agent 拿到名片就能判断“这事该不该找它”。这有点像 Web 里的 robots.txt 加上 OpenAPI 描述只不过描述对象变成了 Agent 能力。任务投递和结果获取对应的是 Task 和 Message。客户端先创建一个 Task把消息放进去发给服务端服务端更新 Task 状态从 submitted 变成 working最后变成 completed同时把结果作为新的 Message 或 artifact 写进去。整个流程本质上是异步任务队列只不过传输层用了 HTTP JSON-RPC。这种设计非常务实Agent 的处理经常要几秒甚至几分钟同步等待不现实所以协议把任务状态机作为一等公民客户端可以轮询也可以等服务端回调。我自己的体会是A2A 并不适合所有场景。如果你要做低延迟高频的流式交互比如实时语音 Agent 每一帧都要通信那直接用 WebSocket 更合适。A2A 的目标是“Agent 之间异步协作”是为后端服务设计的。弄清楚了这一点再看后面的代码就不会觉得别扭。2. 装包和第一印象a2a-protocol 的核心对象与基础语法2.1 安装与导入留意版本差异安装很简单直接用 pippip install a2a-protocol我目前用的版本顶层包名可以这样导入from a2a_protocol import AgentCard, Message, Task, TaskStatus from a2a_protocol.types import TextPart, MessageRole需要提醒一句A2A 协议还比较年轻版本更新时偶尔会用TaskStatus.WORKING偶尔会用TaskStatus.PROCESSING甚至不同小版本里Message的字段名也会有变化。建议你装完之后用python -c import a2a_protocol; print(a2a_protocol.__version__)确认一下版本再对照官方文档看细节。安装包里默认依赖 pydantic所以这些核心类基本都是 pydantic 模型可以直接用model_dump()序列化成 JSON。这一点很关键因为后续不管是写服务端还是客户端我都靠它把对象转成协议数据。2.2 AgentCard、Message、Task 三位一体刚接触这个包的时候最容易混淆的就是这三个类到底分别管什么。我用一个本地生活例子来解释AgentCard是“简历”。它描述 Agent 的身份和能力让别的 Agent 知道你是谁、能接什么活。Message是“社交软件里的会话消息”。它包含谁说的、说了什么、属于哪个会话。Task是“工单系统里的任务单”。它把一条或多条消息包装在一起记录这个任务从创建到完成的状态。比如我定义一个“日志分析 Agent”它的简历长这样card AgentCard( namelog-analysis-agent, description分析日志文本提取错误级别和摘要, urlhttp://localhost:8000/a2a, version1.0.0, )然后我构造一条给它看的消息message Message( message_ide1b8f2c4-2a77-4a1e-b7be-9b2e5a6d7f00, roleMessageRole.USER, content[TextPart(text2025-06-01 10:00:00 ERROR timeout connecting to db)], )最后把这个消息包成一个任务task Task(idtask-0001, statusTaskStatus.SUBMITTED, messages[message])这段代码基本就是 a2a-protocol 的“最小骨架”。你会发现它没有很多黑魔法只是把 Agent 通信里的关键信息拆成了三个可序列化的数据类。2.3 一条消息从发送到完成的完整链路理解了这个链路的生命周期后面写代码才能不失控。一个典型任务长这样客户端构造一个Task里面放一个 role 为 user 的Message。客户端把这个 Task 序列化成 JSON-RPC 请求POST 到服务端的/a2a端点。服务端创建同名 Task但状态改成WORKING同时返回给客户端。Agent 开始处理处理过程中可以追加消息当它想把中间结果发出来时就加一条 role 为 agent 的消息。处理完成后Task 状态变成COMPLETED结果放在新增的 Message 或artifacts数组里。客户端通过轮询拿到最终 Task 对象从中取最新消息或产出物。链路里最关键的点是Task 是唯一的状态载体Message 是附属于 Task 的通信内容。你不需要单独维护一个“会话状态”因为状态就在 Task 上。3. 参数拆解我要配置哪些字段每个字段有什么坑3.1 AgentCard 参数让其他 Agent 能找到你AgentCard是别人了解你 Agent 的唯一入口字段不多但每个都很重要。我整理了一份常用参数表参数是否必填含义我踩过的坑name是Agent 唯一标识一定要全局唯一多个 Agent 用同一个名字会让调用方混乱description是能干什么、边界是什么不能写“万能助手”最好写清输入输出例如“输入错误日志输出 JSON 摘要”url是A2A 服务端点填localhost会导致跨容器、跨机器调用失败要用可解析的地址version否版本号升级接口时建议带上版本capabilities否是否支持流式、推送通知等如果声明了push_notifications但没有实现回调对方会傻等authentication否认证方式共享密钥或 Bearer Token别把密钥写死在简历里default_input_modalities否默认输入类型比如 text 或 file声明了就得真的支持解析文件在代码里我通常这样构造卡片from a2a_protocol import AgentCard, Authentication card AgentCard( namealert-agent, description接收日志分析结果生成告警消息, urlhttp://alert-agent:8000/a2a, version0.2.0, capabilities{streaming: True, push_notifications: False}, authenticationAuthentication(schemes[bearer], credentialstoken-xxx), )注意capabilities是一个字典布尔值千万不要乱写。我第一次就把push_notifications写成了 True结果客户端一直没收到回调任务卡在 working 状态直到超时。后来改成 False客户端才知道要走轮询。3.2 Message 和 Part 参数内容怎么装才不会丢Message本身不是一个字符串它的content是Part对象的列表。这一点新手最容易翻车。常见的Part有TextPart普通文本字段是text。FilePart文件引用字段是file和mime_type。DataPart结构化 JSON 数据字段是data。构造消息时尽量用显式类型from a2a_protocol.types import DataPart msg Message( message_ida1b2..., roleMessageRole.USER, content[ TextPart(text分析以下日志), DataPart(data{lines: 100, source: application.log}), ], )role字段也很关键。标准里大概有user和agent两种个别版本还可能出现system。它可以理解为“这句话是任务发起人说的还是 Agent 回复的”。服务端判断任务是否处理完一般会看最新一条消息是不是 role 为 agent 的消息所以要保证角色写对。另外还有两个可选参数值得注意parent_message_id如果要回复前一条消息这里填前一条的 message_id。多轮对话全靠它串成一条链。metadata一个自由字典。我经常在里面放agent_id、trace_id方便链路追踪。3.3 Task 参数与状态流转异步任务的精髓Task是最容易被忽略参数坑的对象。它的状态不是随便填的协议规定了几种我在实际项目里经常被这几种子状态卡住状态含义何时出现SUBMITTED任务已创建客户端刚发出来WORKING处理中Agent 开始干活INPUT_REQUIRED需要更多输入Agent 发现信息不足COMPLETED已完成结果写入 messages 或 artifactsFAILED失败异常或逻辑错误CANCELED已取消客户端主动取消Task里最重要的参数自然是messages和artifacts。messages保存所有对话消息artifacts保存最终产物比如生成的文件、JSON 数据。一个字一个字地构造 Task 容易出错所以我通常会用一个工厂函数def create_task(content: str, task_id: str) - Task: msg Message( message_idstr(uuid.uuid4()), roleMessageRole.USER, content[TextPart(textcontent)], ) return Task(idtask_id, statusTaskStatus.SUBMITTED, messages[msg])注意一个细节Task的messages即使是初始消息也得放进列表里不能直接传单个对象。协议要求它是数组。4. 实战让日志分析 Agent 和告警 Agent 自动协作4.1 场景设计为什么选这个案例理论讲多了容易飘还是看一个能跑起来的例子。我这次选的是运维领域最常见的场景一个日志分析 Agent专门从原始日志里提取错误级别和摘要一个告警 Agent接收分析结果生成一条上游系统能识别的告警消息。为什么拆成两个 Agent 而不是写成一个因为日志分析和告警策略分别由两个团队维护他们的发布节奏不一样。用 A2A 拆开之后日志分析团队升级模型不影响告警 Agent告警 Agent 修改规则也不影响日志分析的输出。这就是 Agent 协作的价值。拓扑上我们假设客户端只连接日志分析 Agent日志分析 Agent 内部调用告警 Agent。整个过程走完客户端会拿到一个最终的告警结果。4.2 服务端 A日志分析 Agent我用 FastAPI 搭了一个最小服务端。a2a-protocol 的类可以很方便地转成 JSON-RPC 响应。import json import uuid from datetime import datetime from fastapi import FastAPI, Request from a2a_protocol import AgentCard, Task, TaskStatus from a2a_protocol.types import TextPart, DataPart, MessageRole, Message app FastAPI() CARD AgentCard( namelog-analysis-agent, description分析日志文本提取级别和摘要, urlhttp://localhost:8001/a2a, version1.0.0, ) def analyze_log(text: str) - dict: if ERROR in text: level high summary 数据库连接超时 elif WARN in text: level medium summary 连接池使用率偏高 else: level low summary 无异常 return {level: level, summary: summary, raw_length: len(text)} app.get(/.well-known/agent-card.json) async def agent_card(): return CARD.model_dump() app.post(/a2a) async def handle(request: Request): payload await request.json() method payload.get(method) message_id payload.get(id, 1) if method tasks/send: task_data payload.get(params, {}).get(task, {}) task Task.model_validate(task_data) task.status TaskStatus.WORKING latest_content task.messages[-1].content[0].text result analyze_log(latest_content) reply Message( message_idstr(uuid.uuid4()), roleMessageRole.AGENT, content[DataPart(dataresult)], task_idtask.id, parent_message_idtask.messages[-1].message_id, ) task.messages.append(reply) task.status TaskStatus.COMPLETED task.artifacts.append(result) return {jsonrpc: 2.0, id: message_id, result: task.model_dump()} return {jsonrpc: 2.0, id: message_id, error: {code: -32601, message: method not found}}这段代码里最有用的设计是我始终把业务逻辑和协议逻辑分开。analyze_log只是普通函数真正的 A2A 交互发生在handle里。这样后续你想从 FastAPI 换成别的 ASGI 框架业务代码不用动。注意Task.model_validate(task_data)是 pydantic v2 的标准用法如果你用的是旧版可能要改成parse_obj。4.3 服务端 B告警 Agent告警 Agent 的代码结构完全一样只是业务逻辑不同。它不要求返回复杂日志只需要从DataPart里读取 JSON然后生成告警文案。app.post(/a2a) async def alert_handle(request: Request): payload await request.json() method payload.get(method) message_id payload.get(id, 1) if method tasks/send: task Task.model_validate(payload[params][task]) latest task.messages[-1].content[0] # 如果是 DataPart 才正常解析 if hasattr(latest, data): analysis latest.data else: analysis {level: unknown, summary: latest.text} alert_text f[{analysis[level]}] {analysis[summary]} reply Message( message_idstr(uuid.uuid4()), roleMessageRole.AGENT, content[TextPart(textalert_text)], task_idtask.id, ) task.messages.append(reply) task.status TaskStatus.COMPLETED task.artifacts.append({alert: alert_text}) return {jsonrpc: 2.0, id: message_id, result: task.model_dump()} return {jsonrpc: 2.0, id: message_id, error: {code: -32601, message: method not found}}两个服务端除了 AgentCard 里的名字、地址不同协议层的代码几乎一模一样。这其实就是 A2A 最大的价值你写一次协议层所有 Agent 都能复用。4.4 客户端一次性拉通客户端这边我用httpx来发请求没有用底层 against 一些复杂的 SDK因为这样能看到整个协议的样子。import httpx import uuid from a2a_protocol import Task, TaskStatus from a2a_protocol.types import Message, MessageRole, TextPart def create_task(text: str) - Task: msg Message( message_idstr(uuid.uuid4()), roleMessageRole.USER, content[TextPart(texttext)], ) return Task(idstr(uuid.uuid4()), statusTaskStatus.SUBMITTED, messages[msg]) def send_task(url: str, task: Task) - Task: payload { jsonrpc: 2.0, id: 1, method: tasks/send, params: {task: task.model_dump(modejson)}, } with httpx.Client(timeout30) as client: resp client.post(url, jsonpayload) resp.raise_for_status() result resp.json()[result] return Task.model_validate(result) if __name__ __main__: client_task create_task(2025-06-01 10:00:00 ERROR timeout connecting to db) final_task send_task(http://localhost:8001/a2a, client_task) for msg in final_task.messages: print(msg.role, msg.content[0]) print(artifacts:, final_task.artifacts)客户端整个流程没有碰任何 HTTP 细节只和Task打交道。服务端的地址也可以从 AgentCard 的url字段动态获取甚至可以设计一个 Agent 注册中心客户端先查卡片再调任务这就是 A2A 的完整服务发现语义。5. 接入过程中我不吐不快的坑以及几条优化建议5.1 四个我实际遇到的坑这个包我用了一个多月整体很顺手但有几个坑确实让我熬夜调过。第一个坑是 AgentCard 里的url地址。我在本地开发时填了http://localhost:8000/a2a结果一部署到 Docker 里另一个容器根本访问不到这个 localhost。后来我在配置里用服务名或者环境变量动态生成。第二个坑是 Message content 类型。我一开始图省事直接给content传了一个字符串Message(message_idx, roleuser, contenterror)序列化出来 content 是字符串但接收方按列表解析直接抛异常。正确做法永远是content[TextPart(texterror)]。第三个坑是任务状态枚举的兼容性。不同版本的 a2a-protocol 对“处理中”的叫法不一样有的叫WORKING有的叫WORK_IN_PROGRESS。如果你的系统里同时跑着多个客户端和服务端最好在构造请求和解析响应时都做一层状态映射。第四个坑是 metadata 参数。我在里面放了一个datetime对象结果调用model_dump(modejson)时直接报错因为 datetime 不是 JSON 原生类型。规范的做法是提前转成 ISO 字符串或者统一用时间戳。5.2 让 a2a-protocol 用得更顺的进阶配置如果你确定要用这个包做正式项目我建议你做三件事。第一把 AgentCard 放在一个公共配置模块里所有服务端启动时都从这里加载。这样不会出现几个 Agent 的卡片描述与真实能力不一致。第二为每个任务生成稳定的 UUID 作为任务 ID这样客户端重试时可以幂等服务端如果发现同一个 task_id 已经存在可以直接返回已有任务不重复执行。第三为任务增加metadata.trace_id把 A2A 交互日志和业务日志串起来排查问题时能省大量时间。我在生产环境里还加了一个保险客户端轮询 Task 时如果一段时间内状态没有变化就主动取消任务并告警。这不是 a2a-protocol 自带的功能但配合协议的任务状态机做起来非常容易。最后分享一个小技巧。如果你想让日志分析 Agent 调用告警 Agent在第一个 Agent 的 handler 里直接使用httpx调用第二个 Agent 的/a2a即可不需要引入额外的编排框架。协议本身就是为这种嵌套调用设计的一个 Agent 既可以当客户端也可以当服务端。我实际项目里就是让订单 Agent 动态调用了库存 Agent整个过程和上面案例里的客户端调用方式一模一样。这种组合方式非常灵活也是 A2A 协议最吸引我的地方。