几个月前，我接手一个内部系统，需要让两个 AI Agent 互相传话。一个负责解析订单信息，另一个负责查库存，两个 Agent 用着完全不同的框架，连消息格式都各说各话。第一版我用 requests 直接调对方接口，每接一个 Agent 就要写一套适配代码，参数映射、错误处理、超时重试全都手搓，实在是烦。后来换成了 a2a-protocol 这个 Python 包，总算是把这件事从“治标”变成了“治本”。如果你也在做多 Agent 协作，或者准备把 Agent 服务暴露给第三方，那么这篇内容应该能帮你少走不少弯路。
a2a-protocol 是 Agent2Agent（A2A）协议的 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 安装与导入：留意版本差异
安装很简单，直接用 pip：
pip 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( name="log-analysis-agent", description="分析日志文本，提取错误级别和摘要", url="http://localhost:8000/a2a", version="1.0.0", )然后我构造一条给它看的消息：
message = Message( message_id="e1b8f2c4-2a77-4a1e-b7be-9b2e5a6d7f00", role=MessageRole.USER, content=[TextPart(text="2025-06-01 10:00:00 ERROR timeout connecting to db")], )最后把这个消息包成一个任务：
task = Task(id="task-0001", status=TaskStatus.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( name="alert-agent", description="接收日志分析结果，生成告警消息", url="http://alert-agent:8000/a2a", version="0.2.0", capabilities={"streaming": True, "push_notifications": False}, authentication=Authentication(schemes=["bearer"], credentials="token-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_id="a1b2...", role=MessageRole.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 或 artifacts |
FAILED | 失败 | 异常或逻辑错误 |
CANCELED | 已取消 | 客户端主动取消 |
Task里最重要的参数自然是messages和artifacts。messages保存所有对话消息，artifacts保存最终产物，比如生成的文件、JSON 数据。
一个字一个字地构造 Task 容易出错，所以我通常会用一个工厂函数：
def create_task(content: str, task_id: str) -> Task: msg = Message( message_id=str(uuid.uuid4()), role=MessageRole.USER, content=[TextPart(text=content)], ) return Task(id=task_id, status=TaskStatus.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( name="log-analysis-agent", description="分析日志文本，提取级别和摘要", url="http://localhost:8001/a2a", version="1.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_id=str(uuid.uuid4()), role=MessageRole.AGENT, content=[DataPart(data=result)], task_id=task.id, parent_message_id=task.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_id=str(uuid.uuid4()), role=MessageRole.AGENT, content=[TextPart(text=alert_text)], task_id=task.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_id=str(uuid.uuid4()), role=MessageRole.USER, content=[TextPart(text=text)], ) return Task(id=str(uuid.uuid4()), status=TaskStatus.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(mode="json")}, } with httpx.Client(timeout=30) as client: resp = client.post(url, json=payload) 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_id="x", role="user", content="error")序列化出来 content 是字符串，但接收方按列表解析，直接抛异常。正确做法永远是content=[TextPart(text="error")]。
第三个坑是任务状态枚举的兼容性。不同版本的 a2a-protocol 对“处理中”的叫法不一样，有的叫WORKING，有的叫WORK_IN_PROGRESS。如果你的系统里同时跑着多个客户端和服务端，最好在构造请求和解析响应时都做一层状态映射。
第四个坑是 metadata 参数。我在里面放了一个datetime对象，结果调用model_dump(mode="json")时直接报错，因为 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 协议最吸引我的地方。
