a2a-types：Python实现A2A协议的类型层，规范Agent通信

做 Agent 工程的朋友最近应该没少听到 A2A 这三个字母。Agent2Agent官方定位是解决 AI Agent 之间互相调用、互相通信的互操作问题而a2a-types这个 Python 包就是 A2A 协议在 Python 生态里的类型定义层。它不像框架那样帮你把 HTTP、JSON-RPC、回调全部实现好而是把协议里所有消息、任务、Agent 卡片这些概念映射成了可导入的 Python 类。你拿到它之后能做的第一件事不是跑一个 demo而是立刻获得一套完整的类型提示和参数校验避免 Agent 之间发出去的消息格式千奇百怪。这篇文章我会从语法、参数到实际案例把a2a-types的用法完整拆一遍适合正在做 Agent 互操作、想自己实现 A2A 服务端或客户端的开发者参考。1. 先搞懂 A2A 协议与 a2a-types 的定位1.1 为什么 Agent 之间需要统一通信规范早期的 AI Agent 大多是单机脚本自己接工具、自己处理上下文。后来 Agent 越来越多大家发现一个问题让 Agent 互相调用的时候没有一个双方都认的“接口格式”。A 用 task_id 标识任务B 用 request_idC 干脆用 UUID 字符串联调起来全靠文档文档一过时就是灾难。A2A 做的事情就是把 Agent 之间交互的“表单”统一掉。你可以把它理解成两个人约见面A 说“明天下午”B 说“14 点整”鸡同鸭讲如果都按北京时间、24 小时制来表述就不会有歧义。A2A 就是那个 GMT8、24 小时制。而a2a-types把这份“表述规范”翻译成了 Python 里的类、枚举和校验规则这样你在 IDE 里写代码时不用翻协议文档也能知道一个消息对象该长什么样。这个价值在多人协作时尤其明显。你只需要让前后端都依赖同一份类型包协议更新了类型也跟着更新编译期或导入期就能发现接口不匹配比运行时才报错舒服太多了。1.2 a2a-types 在整个 SDK 里扮演什么角色大多数 A2A 生态里会同时出现几个概念a2a-types是纯类型层a2a是完整 SDK可能还有功能更全的客户端和服务端封装。a2a-types被单独拆出来是为了让只关心数据结构的人不用背上 HTTP 和异步框架的依赖。实际开发中你可以只用a2a-types定义消息然后自己用 FastAPI、Flask、甚至 Django 写接口。我试过这种组合灵活性比直接用大而全的 SDK 更高因为很多团队已经有自己的 Web 框架和鉴权体系再强行引入一套 SDK 反而是负担。类型层不受网络层影响也方便你跑单元测试。从依赖关系看a2a-types通常基于 Pydantic 的 BaseModel。这样设计有三个好处序列化直接model_dump()或.json()反序列化直接Model(**data)校验错误能精确定位到某个字段。你不需要自己写一堆if not isinstance(...)的防御代码。1.3 为什么类型层用 Pydantic 而不是 dataclass有人会问Python 自带 dataclass为什么还要 Pydantic核心区别是校验。A2A 协议里很多字段有严格的取值范围和嵌套结构比如TaskState只能是submitted、working、completed、failed等几个枚举值Part需要区分TextPart、FilePart、DataPart这些用 dataclass 做会很啰嗦而且运行时不校验错误数据会一路流传到业务深处。Pydantic 还有一个杀手级能力面对多态类型时可以自动判别。比如一个Part字段你传入{kind: text, text: hello}Pydantic 能根据kind字段自动反序列化成TextPart省去手写 if-else 分支。a2a-types正是利用了这一点让消息对象既能保持类型安全JSON 表达又足够简洁。2. 核心语法与参数逐个拆解2.1 a2a-types 的安装和版本选择安装没什么特殊的pip install a2a-types如果你的项目里已经装了a2a完整 SDK那a2a-types大概率已经被带了进来不需要重复安装。需要注意版本匹配协议本身还在演进a2a-types的版本号会跟随协议更新导入老版本的类名或者字段名在model_dump()时可能少字段。建议固定大版本比如a2a-types0.1,0.3避免小版本升级后协议字段变名导致线上事故。我在本地经常用一个干净环境测试python -m venv .venv source .venv/bin/activate pip install a2a-types pydantic装完之后做一次自检在 Python 里执行导入命令确保包可用from a2a.types import AgentCard, Message, Task print(AgentCard.__name__)2.2 消息与任务相关的类型参数消息和任务是你日常打交道最多的对象。我把常用到的类型用表格列一下类型核心参数用途Messagerole、parts、metadata通用消息role通常取user或agentAgentMessagemessageId、parts、metadataAgent 发出的消息有独立 IDTaskid、status、artifacts、metadata一个任务单元包含状态和产出TaskState枚举值任务状态submitted、working、completed、failed、canceled、input-requiredTaskStatusstate、message、timestamp任务摘要信息Artifactid、name、parts、metadata任务完成后交付的产出物Partkind消息内容的抽象基类TextParttext文本内容FilePartfile文件引用DataPartdata结构化数据构造一条 Agent 消息的典型写法from a2a.types import AgentMessage, TextPart msg AgentMessage( messageIdmsg-001, parts[ TextPart(text今天杭州天气怎么样), ], metadata{source: customer-service} ) print(msg.model_dump())这里最有用的参数是metadata。A2A 官方没限制它的结构你可以塞任意 JSON比如来源系统、链路 ID、会话 ID。后续排查问题时这些信息能救命。另一个值得注意的点是parts是列表类型这意味一条消息可以同时携带文本和文件不要把多段内容拆成多条消息虽然也能工作但语义上不完整。Task的构造稍微复杂一点它需要挂一个TaskStatus。你可以这么理解Task是任务实体TaskStatus是它的健康状态牌。每当任务推进你就创建一个新的状态对象而不是直接改Task里的字符串这样外部调用方能看到任务演进的轨迹。from a2a.types import Task, TaskStatus, TaskState task Task( idtask-001, statusTaskStatus( stateTaskState.submitted, message任务已进入队列, timestampdatetime.now(timezone.utc) ) )timestamp参数建议显式传 UTC 时间不要省略。因为后续序列化成 JSON 给其他 Agent 时带时区的时间不容易产生歧义否则你们俩都在各自的本地时间哪天跨时区部署就是事故。2.3 AgentCard 与 Capabilities 的参数细节Agent 不像 REST API 那样只有一个健康检查端点它需要一个“自我描述卡片”告诉别的 Agent 自己能干什么、支持什么传输方式。这个卡片就是AgentCard。构造时重点参数如下from a2a.types import AgentCard, Capabilities, Skill, Provider card AgentCard( nameweather-agent, description提供城市天气查询和预报服务, urlhttps://agent.example.com/a2a, version1.0.0, providerProvider( organizationexample-com, nameWeather Team ), capabilitiesCapabilities( streamingFalse, pushNotificationsFalse ), skills[ Skill( idweather_query, name天气查询, description根据城市名返回实时天气, tags[weather, query] ) ] )Capabilities的两个布尔参数决定了这个 Agent 是否支持流式输出、是否支持主动推送。如果你打算支持流式那么后续通信时要处理 SSEServer-Sent Events事件格式类型更复杂如果只做同步请求这两个参数都设False就行。很多初学者会忽略skills字段。它本质上是给其他 Agent 看的“菜单”其他 Agent 发现你的卡片后不会逐个尝试你的接口而会先搜索skills里的描述找到匹配的 skill 再发起调用。所以Skill的description不能乱写要像写 API 文档一样写清楚“什么情况下用”。我见过有团队把描述写成“提供了很多功能”实际调度效果几乎等于随机。2.4 类型语法细节枚举、Optional、Uniona2a-types的代码里大量使用了 Python 类型系统的高级语法如果你自己写协议模型这些语法也能直接用。三个方面最常用。第一枚举。A2A 的状态字段直接用TaskState枚举比裸字符串安全得多。你可以在业务代码里做分支if task.status.state TaskState.completed: ... elif task.status.state TaskState.failed: ...写完这块逻辑后如果协议新增了状态值类型提示会立刻让你意识到还有分支没处理。第二Optional。很多字段不是必填比如metadata、Artifact.name。定义模型时使用Optional[...]明确语义。调用端如果依赖某个可选字段要先做空值判断。第三Union与Part的多态。parts列表里的元素是TextPart、FilePart、DataPart的联合类型。处理时推荐isinstance判断from a2a.types import TextPart, FilePart, DataPart for part in msg.parts: if isinstance(part, TextPart): print(f文本: {part.text}) elif isinstance(part, FilePart): print(f文件: {part.file.name}) elif isinstance(part, DataPart): print(f数据: {part.data})这里有个小坑isinstance判断不能反过来写即不能用type(part) is TextPart因为 Pydantic 可能有子类继承场景type()精确判断会漏掉子类实例。2.5 参数校验的坑Pydantic 的校验很严格但也会带来一些“惯性坑”。最典型的就是给TaskState.completed传大写字符串Completed。Pydantic 默认枚举大小写敏感会直接抛ValidationError。我不止一次被这个卡过后来养成了一个习惯所有枚举状态统一转小写。另一个坑是metadata参数。如果你传入一个普通dict但其中某个 value 是自定义对象Pydantic 会尝试序列化或报错。建议只传 JSON 能表达的数据类型字符串、数字、布尔值、嵌套字典和列表都行。传 datetime 对象虽然能序列化但对方如果拿不到类型定义还得猜格式不如统一传 ISO 字符串。3. 实操案例两个 Agent 通过 A2A 通信3.1 案例背景与整体结构我们做一个最小可跑的案例一个客服 Agent 收到用户问题“杭州天气怎么样”它构造一个 A2A Task 发给天气 Agent天气 Agent 处理后返回结果。整体结构如下天气 Agent 服务FastAPI 实现监听/a2a端点。客服 Agent 客户端纯 Python 脚本发起 HTTP 请求。消息格式用a2a-types定义。这里我不会启用完整 SDK 的自动路由因为那样会掩盖很多细节。自己写三十行路由代码你能看到每个字段是怎么流转的之后切换完整 SDK 也心里有数。3.2 定义 Agent 能力并启动服务端先建一个文件weather_agent.py定义 AgentCard 和 FastAPI 路由。from datetime import datetime, timezone from typing import Optional from fastapi import FastAPI from pydantic import BaseModel from a2a.types import ( AgentCard, Capabilities, Skill, Provider, Task, TaskStatus, TaskState, AgentMessage, TextPart, Artifact ) app FastAPI() # AgentCard 用于自描述 agent_card AgentCard( nameweather-agent, description查询城市实时天气, urlhttp://localhost:8000/a2a, version1.0.0, providerProvider(organizationexample-com, nameWeather Team), capabilitiesCapabilities(streamingFalse, pushNotificationsFalse), skills[ Skill( idweather_query, name天气查询, description输入城市名返回天气描述和温度, tags[weather, query] ) ] ) # 模拟天气查询逻辑 def fake_weather(city: str) - str: return f{city} 今天多云气温 22 摄氏度东北风 3 级。然后实现一个简单的 JSON-RPC 风格接口。A2A 的传输规范通常会用method字段区分card和send等调用这里我按最简方式处理class A2ARequest(BaseModel): method: str params: dict app.post(/a2a) async def a2a_endpoint(req: A2ARequest): if req.method card: return agent_card.model_dump() if req.method send: # params 里通常包含 task 对象 task Task(**req.params[task]) return handle_task(task) return {error: method not supported}注意接口层我只负责拿到dict后转成Task。这正是类型包的价值点请求到业务之间隔了一个安全的转换层字段不对马上抛ValidationError不会在业务逻辑里现出诡异空指针。handle_task里推进任务状态并返回结果def handle_task(task: Task) - Task: # 标记任务 processing task.status TaskStatus( stateTaskState.working, message正在查询天气, timestampdatetime.now(timezone.utc) ) # 从任务消息里取城市名 city None for part in task.input.parts: if isinstance(part, TextPart): city part.text.strip() # 模拟天气结果 result_text fake_weather(city or 未知城市) # 生成产出 artifact artifact Artifact( idart-001, nameweather-result, parts[TextPart(textresult_text)] ) # 最终状态 completed task.status TaskStatus( stateTaskState.completed, message查询完成, timestampdatetime.now(timezone.utc) ) task.artifacts [artifact] return task这里有个容易被忽略的点task.input是什么依据 A2A 定义Task会携带输入内容可能是一个Message对象。在构造请求端时我们就要把用户问题塞进Task.input这样服务端才能从task.input.parts里取到城市名。启动服务uvicorn weather_agent:app --port 80003.3 构建 Task 消息并调用另一个 Agent客户端脚本client.py模拟客服 Agent。它先查询天气 Agent 的卡片再发送任务。import requests from datetime import datetime, timezone from a2a.types import AgentMessage, TextPart, Task, TaskStatus, TaskState BASE_URL http://localhost:8000/a2a # 1. 获取 AgentCard card_resp requests.post(BASE_URL, json{method: card, params: {}}) card_data card_resp.json() print(Agent:, card_data[name], card_data[skills]) # 2. 构建 Task msg AgentMessage( messageIdmsg-002, parts[TextPart(text杭州)] ) task Task( idtask-002, statusTaskStatus( stateTaskState.submitted, message客服转发用户问题, timestampdatetime.now(timezone.utc) ), inputmsg ) # 3. 发送给天气 Agent send_resp requests.post( BASE_URL, json{ method: send, params: {task: task.model_dump()} } ) result_task Task(**send_resp.json()) print(状态:, result_task.status.state) for artifact in result_task.artifacts: if isinstance(artifact, Artifact): for part in artifact.parts: if isinstance(part, TextPart): print(天气结果:, part.text)这一步可能是很多人第一次感受到a2a-types优势的地方。发送前task.model_dump()把 Task 转成普通 JSON收到响应后Task(**data)把 JSON 重新变成强类型对象。如果你手写字典处理校验和类型推断全得自己写代码量至少翻倍。3.4 接收结果与状态流转上面的流程看起来简单但你在真实业务里一定会遇到状态多次流转的情况。比如天气 Agent 收到任务后先返回working状态然后处理完成后再返回completed。我们的同步 demo 省略了中间态但真实做法是客户端先收到submitted/working。任务完成后服务端可能通过task/status推送或 WebSocket 通知也可能要求客户端轮询。客户端每次拿到新的Task对象去看status.state。a2a-types的TaskState枚举能帮你避免魔法字符串。我见过有人在业务里写if res[state] success后来服务端改成了completed客户端直接静默失败。用枚举后这个问题被完全消灭编译器会提示你用的值不存在。4. 常见问题与排查技巧实录4.1 字段名变迁与兼容问题A2A 协议迭代中字段名发生过调整。比如有的版本用agentCard有的版本用cardTask里的input字段类型也从纯文本变成了结构化消息。最稳妥的排查方式是直接打印模型的model_dump()把 JSON 贴到协议文档里对比。我做升级兼容时会在代码里做一个适配层def parse_task(data: dict) - Task: # 兼容旧的 input 字符串 if isinstance(data.get(input), str): data[input] {role: user, parts: [{kind: text, text: data[input]}]} return Task(**data)不要小看这个兼容函数。Agent 之间部署节奏不同你升级了类型层对方还没升级旧消息就会打进新字段。想避免这种问题要么消息模型里加版本号要么用适配层兜底。4.2 时间序列化导致的消息校验失败最经典的问题是日期时间从 JSON 里回来时变成了字符串而 Pydantic 对datetime字段默认要求 ISO8601 格式。如果对方传的是2025-06-01 12:30:00带空格解析会挂。排查技巧报错信息里会写出具体字段和解析失败的原始值不要只看最下面那行。常见的格式差别包括原始格式是否能被 Pydantic 解析2025-06-01T12:30:00Z能2025-06-01T12:30:0000:00能2025-06-01 12:30:00不能1717230600时间戳不能统一在发送端就生成 UTC ISO 字符串问题会少很多。如果收到别人的消息发现时间字段不过校验先 altair 这种规矩把对方的时间字符串规整成标准 ISO。4.3 Part 子类判断和自定义内容很多人拿到parts后想用match语句Python 3.10 以上确实能这么写for part in msg.parts: match part: case TextPart(textbody): print(body) case FilePart(filefile_ref): print(file_ref) case DataPart(databody): print(body)不过实际协议里还会出现没有kind的裸Part或者你用自定义的 Part 子类时没有正确登记类型。如果遇到“反序列化出来不是我的子类”的情况优先检查两处第一你的子类有没有在__init__里正确调用父类第二kind字段是否和其他类冲突。一个习惯是给自定义 Part 设置独特的kind值比如image_url不要和内置的text、file、data重名。4.4 调试 A2A 消息的实用套路调试时不要直接打印完整对象那个输出又长又难读。先调用model_dump_json(indent2)看序列化结果再逐层递归检查parts、artifacts、status字段。我一般会写一个迷你函数from a2a.types import Message, Task def summarize(obj): if isinstance(obj, Task): print(Task id:, obj.id) print(State:, obj.status.state) print(Artifacts:, [a.name for a in obj.artifacts or []]) elif isinstance(obj, Message): print(Role:, obj.role) print(Parts:, [(p.kind, getattr(p, text, )) for p in obj.parts])这类工具函数看起来简单但在多 Agent 联调时特别省时间。遇到通信问题先汇总两边收到的Task.status.state你对齐状态流转问题一般就定位了一半。5. 写在后面的建议与扩展方向5.1 从类型层过渡到完整 SDK如果a2a-types你用得很顺下一步可以考虑官方完整 SDK。完整 SDK 会在类型层之上帮你实现 JSON-RPC 端点、鉴权扩展、事件流等机制。我自己做内部系统时依然偏爱只依赖a2a-types因为团队已经有网关和服务治理重复造网络层的需求反而更少。如果你是项目从零开始就直接用完整 SDK省出来的时间足够把业务逻辑打磨得更好。5.2 和 MCP 的对比什么时候选 A2A很多人会把 A2A 和 MCP 弄混。MCP 解决的是 Agent 怎么调用工具A2A 解决的是 Agent 怎么调用另一个 Agent。两者不是替代关系而是互补关系。a2a-types不依赖任何 MCP 库你可以在一个 Agent 内部同时使用 MCP 拉数据、通过 A2A 和另一个 Agent 协作。选型时记住一点如果对方是纯工具用 MCP如果对方是完整的、需要理解和完成复杂任务的 Agent用 A2A。5.3 几个提高开发效率的习惯最后分享几个我实际用着很舒服的习惯。第一所有 Agent 交互的入参出参都定义成a2a-types的类型不要裸传字典。第二在日志里统一打印task.id和messageId多 Agent 调用链排查时这是唯一关联线索。第三给自己的 AgentCard 增加一个tags列表方便后续做 Agent 发现服务别让它空着。第四单元测试里直接用Task(**json_data)构造对象验证状态流转分支不需要依赖真实网络。我个人在实际操作中的体会是a2a-types这类类型层工具看似简单却能把 Agent 系统里最容易失控的“消息格式”问题前置处理掉。你只要把几个核心类型玩熟后面的业务扩展基本就是往parts和skills里加内容改动很克制心智负担也很小。希望能对正在做 Agent 互操作项目的你有帮助。