如果你在2025年之后开始认真弄Agent协作大概率会撞上A2AAgent2Agent协议。而当你打开这套协议的Python仓库第一眼看到的往往不是那些讲原理的文档而是a2a-types这个包。它很不起眼很多教程只在安装依赖时带一句但它恰恰是整个Python SDK里最值得先读透的一块。不夸张地说理解了a2a-types你再看A2A协议就事半功倍。简单说a2a-types负责把协议里的JSON结构翻译成你能直接操作的Python对象并在发送前把Python对象再“变回”协议要求的JSON格式。它不负责网络请求也不负责服务端路由只专注类型定义、参数校验和序列化。我认为所有要写Agent、要评估A2A协议、以及想搞懂“Agent之间到底传什么数据”的人都应该先从这一个包入手。下面的内容就是我从语法、参数到实际项目里跑通案例的完整记录。1. 为什么要单独搞一个“类型层”先搞清a2a-types在生态里的位置1.1 A2A协议解决的是“Agent互相交活”的通信问题A2A协议本质是一套面向Agent协作的开放规范。各个Agent由不同团队、不同语言、不同框架开发彼此要能互相发现、派发任务、查询进度、拿回结果中间必须有一个大家都认的“数据结构”。A2A就把这些数据结构的格式固定下来例如任务长什么样、消息里能带什么内容、状态枚举有哪些。你可以把它类比成两个公司之间对一套商务单据格式的约定单据上必须有哪些字段、日期用什么格式、金额用什么单位。只要双方都按约定填整个协作流程就能跑通。A2A协议底层的通信宿体是JSON-RPC 2.0走的通道一般是HTTP。但协议文档里那些Task、Message、Part之类的概念最终要在Python代码里落地就需要一个“类型层”。a2a-types就是这个类型层。1.2 类型层、客户端、服务端三块各管什么我把A2A的Python官方生态粗略拆成三块组件职责什么时候用a2a-types协议数据模型、枚举、校验、JSON序列化所有需要构造或解析A2A消息的地方a2a-sdk高层客户端、FastAPI服务端适配、任务管理要实际发HTTP请求、要对外提供服务时协议文档规范本身、字段语义、版本演进需要查“为什么这么设计”时我第一次接触时犯过一个方向性错误以为要跑通Agent协作就得先把客户端和服务端代码写好。实际上最应该先动手的是把类型层摸熟。原因很简单——你在本地就可以用一段JSON mock出远端Agent的返回然后用a2a-types解析整个过程不依赖网络。只要类型层解析逻辑是对的后面接任何HTTP传输层心里都有底。2. 从第一个模型开始AgentCard、Task和Message的语法长什么样2.1 安装与导入路径安装没什么特别的建议直接装pip install a2a-sdk官方SDK会顺手把a2a-types作为依赖带进来。如果你只想关心类型层也可以单独安装类型包。这里提醒一句目前不同分发版本在导入路径上略有差异有的用from a2a.types import ...有的用from a2a_types import ...。我下面统一按官方SDK仓库里常见的a2a.types命名空间写如果你的环境是独立包把导入名换成a2a_types即可其余代码逻辑完全一样。先看第一个例子用AgentCard把一个Agent“挂出去”from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, AgentProvider, ) card AgentCard( nameweekly-report-agent, description把原始业务数据变成周报摘要的助手, urlhttp://agent.internal:10080, version0.1.0, capabilitiesAgentCapabilities(streamingFalse), skills[ AgentSkill( idreport-gen, name生成周报, description输入原始业务数据输出周报摘要, ) ], providerAgentProvider( organizationExample Inc., urlhttps://example.com, ), ) print(card.model_dump(exclude_noneTrue))这个对象描述的是“我是一个什么样的Agent、我有什么技能、你到哪里找我”。A2A的发现机制就是靠这类卡片信息完成的。别小看这段代码它决定了对方Agent看到你的名片后是否愿意调用你。2.2 用Task和Message描述一次任务协作AgentCard是静态描述而真正干活的流程要用Task和Message来承载。一次最简单的任务协作可以长这样from datetime import datetime, timezone from a2a.types import Task, TaskState, TaskStatus, Message, Role, TextPart def now_ts() - str: return datetime.now(timezone.utc).isoformat().replace(00:00, Z) task Task( idtask_20250601_001, statusTaskStatus( stateTaskState.working, timestampnow_ts(), messageMessage( roleRole.agent, parts[TextPart(text已收到任务开始处理)], ), ), ) print(task.status.state)从语法上讲这就是Pydantic模型的常规用法但有几个关键点值得注意Task.status不是普通字符串而是完整的TaskStatus对象里面包含state、timestamp、message。Message.parts是一个列表列表里可以放不同类型的零件Part文本只是其中一种。构造时间戳时我习惯直接生成UTC的ISO格式字符串避免时区问题。2.3 序列化和反序列化a2a-types的日常操作类型层最大的价值在于和外部JSON的互转。Pydantic v2 提供的四个方法在a2a-types里就是日常主力# Python对象 - 普通dict data task.model_dump(exclude_noneTrue) # Python对象 - JSON字符串 json_str task.model_dump_json() # JSON字符串 - Python对象 task_from_json Task.model_validate_json(json_str) # dict - Python对象 task_from_dict Task.model_validate(data)我建议在构造线上请求时加上exclude_noneTrue或by_aliasTrue。前者把没设置的可选字段过滤掉减小消息体积后者会在协议字段和Python字段名不一致时输出协议要求的风格。后面我会专门讲这个坑。3. 核心参数逐个拆真正需要关心的字段与默认行为3.1 AgentCard参数让别人快速认识你AgentCard是所有Agent发现机制的起点。参数不多但每一个都影响“别人愿不愿意理你”。参数类型是否必填作用namestr是Agent名称建议短而明确descriptionstr是能力自我介绍描述能替对方解决什么问题urlstr是接收任务调用的HTTP入口versionstr是协议/功能版本号便于兼容判断providerAgentProvider否运营方主体信息capabilitiesAgentCapabilities否声明是否支持流式输出、推送等能力skillslist[AgentSkill]推荐核心技能清单让对方快速知道你擅长什么我在实际项目里发现description和skills是最影响匹配效果的。如果你写得太泛对方Agent做路由时很难判断是否该调用你写得具体一些反而能减少无意义的调用。比如“把原始业务数据变成周报摘要的助手”就比“AI助手”强得多。3.2 Task与TaskStatus参数任务从提交到结束的完整状态Task是A2A里的核心实体字段如下参数类型说明idstr全局唯一任务ID由服务端生成statusTaskStatus当前状态包含state、message、timestampmaybeContextIdstr可选如果任务派生自某个会话上下文带上上下文IDmetadatadict可选自定义附加信息不适合放大业务数据artifactslist可选任务产出物historylist可选历史消息记录TaskStatus.state的枚举值需要背下来因为几乎所有分支判断都围着它转枚举值含义submitted已提交等待Agent受理working正在处理中input-required需要更多输入等调用方补充completed处理完成failed处理失败canceled被取消rejected被拒绝许多人会把failed和rejected混在一起处理其实它们语义不同rejected表示这个请求Agent不想接failed表示接了但没做成。后面接告警和重试策略时这个区别很关键。3.3 Message与Part参数消息体由哪几种零件组成Message是Agent之间传递信息的载体核心参数是role和parts。role只有两个值user和agent。别以为“user”就是人类用户在A2A里调用方的Agent同样会以user身份发消息真正执行任务的Agent则以agent身份回消息。parts是一个零件列表目前常见的有TextPart纯文本适合给人类看的说明、过程日志。FilePart文件引用适合传文档、图片、表格。DataPart结构化数据适合让另一个程序直接消费的JSON。AudioPart、VideoPart音视频内容用得相对少。我自己做项目时一个重要的取舍是给人看的放TextPart给程序用的放DataPart。很多人为了省事把所有内容都塞进文本里结果下一步Agent要解析时还得写正则或LLM抽取白白增加复杂度。3.4 构造Payload时最容易被忽略的一点role别搞反刚才说了初始请求的message.role通常是user即使发送方也是一个Agent。这个设计的本意是站在接收方视角所有外部请求都是“用户侧”来的。我见过好几个同事第一次写时都习惯性地填Role.agent结果对面解析完发现消息来源不对直接按协议规范拒收或产生奇怪行为。每次构造消息前先问自己一句这句话是从哪一侧发出的发送侧 -user接收侧 -agent。4. 实际应用案例一个订单质检Agent的“任务派发-处理-回传”演示4.1 场景设定这个案例来自我实际做过的内部工具业务背景不复杂有一个派单器需要调用“订单质检Agent”去检查一批订单质检Agent把每条订单的合格情况算出来最终返回汇总结果。派单器只关心三件事总条数、通过数、失败数以及失败订单ID列表。这个场景非常适合演示A2A因为它有明确的任务派发也有明确的结构化返回结果。整个流程是派单器构造一个Task相关的任务消息。质检Agent收到请求先回一个working状态。质检Agent计算完成后把结果放进DataPart状态改为completed。派单器先看状态再从DataPart里拿结构化数据。4.2 构造请求把业务数据装进DataPartfrom datetime import datetime, timezone from a2a.types import Message, Role, DataPart, TextPart, TaskState, TaskStatus, Task def now_ts() - str: return datetime.now(timezone.utc).isoformat().replace(00:00, Z) order_ids [ORD-001, ORD-002, ORD-003] # 把订单列表包装成结构化零件 payload Message( roleRole.user, parts[ TextPart(text请质检以下3条订单), DataPart(data{order_ids: order_ids}), ], ) # 在实际的SDK里这个message会进一步封装成TaskSendParams print(payload.model_dump_json(by_aliasTrue))重点看DataPart的用法我们把order_ids作为一个结构化数据放在零件里而不是拼进文本。这样质检Agent收到后可以直接从data[order_ids]取值不需要再解析文本。4.3 模拟Agent处理流程并返回Task接下来模拟质检Agent这一侧它收到请求后先创建一个working状态的任务然后计算结果最后把状态更新为completeddef run_quality_check(task_id: str) - Task: initial_message Message( roleRole.agent, parts[TextPart(text已收到质检请求开始处理)], ) task Task( idtask_id, statusTaskStatus( stateTaskState.working, timestampnow_ts(), messageinitial_message, ), ) # 模拟计算 total 3 passed 2 failed 1 failed_ids [ORD-002] result_message Message( roleRole.agent, parts[ TextPart(text质检完成共3条记录1条失败), DataPart( data{ total: total, passed: passed, failed: failed, failed_ids: failed_ids, } ), ], ) task.status TaskStatus( stateTaskState.completed, timestampnow_ts(), messageresult_message, ) return task task run_quality_check(task_order_20250601_001)注意我更新状态时是给task.status整个赋了一个新的TaskStatus对象而不是在原对象上修改字段。这样写更清晰也避免少传字段导致校验报错。4.4 客户端解析结果先看状态再取结构化数据派单器这一侧拿到远端Agent返回的JSON后先用Task.model_validate_json解析然后根据状态进入不同分支from a2a.types import DataPart, TextPart def parse_quality_result(payload_json: str) - dict: task Task.model_validate_json(payload_json) if task.status.state TaskState.working: raise ValueError(任务还在处理中请稍后轮询) if task.status.state TaskState.failed: error_text for part in task.status.message.parts: if isinstance(part, TextPart): error_text part.text raise RuntimeError(f质检失败: {error_text}) if task.status.state TaskState.completed: for part in task.status.message.parts: if isinstance(part, DataPart): return part.data raise RuntimeError(任务完成但没有结构化结果) raise RuntimeError(f未处理的状态: {task.status.state}) result parse_quality_result(task.model_dump_json()) print(result)这里有几个很实际的经验分支判断要放在最前面而不是每个分支里都打印结果。先判断working还是input-required可以有效避免“结果没拿到却硬解析”的问题。用isinstance(part, DataPart)做类型收窄代码可读性和可靠性都更高。不要假设completed状态一定带DataPart有些Agent可能只回文本所以最后一层兜底要写好。4.5 完整回环验证把Task转成JSON再解析回来为了让这套逻辑在本地先跑通我建议你直接做一次“自问自答”验证task run_quality_check(task_order_20250601_001) payload_json task.model_dump_json(by_aliasTrue) result parse_quality_result(payload_json) assert result[total] 3 assert result[failed_ids] [ORD-002] print(全链路验证通过)这一步非常值。只要a2a-types的序列化与反序列化逻辑没问题后面接真实HTTP服务时你只需要把payload_json换成网络层返回的响应即可排查问题的范围会小很多。5. 踩坑记录序列化别名、时间戳和枚举大小写5.1 snake_case还是camelCase两套字段名交叉验证协议规范里很多字段是camelCase风格比如taskId、pushNotificationsPython 习惯是snake_case。a2a-types在Pydantic模型里通常会通过别名机制做转换。我在第一次联调时遇到过一个问题自己这边model_dump()出来的JSON是snake_case对端 Agent 用camelCase解析导致字段找不到。后来我把几个模型的输出都打印出来逐字段对比才发现问题。给你一个简单的排查习惯card AgentCard(...) print(card.model_dump()) # 看Python风格字段名 print(card.model_dump(by_aliasTrue)) # 看协议风格字段名两者一对比立刻就能知道当前环境的别名配置。给线上发送消息时我基本都会用by_aliasTrue因为A2A协议文档里字段风格以camelCase为准。5.2 timestamp别用time.time()给TaskStatus一个规范的UTC时间TaskStatus.timestamp在协议里必须是符合RFC3339的字符串类似2025-06-01T10:00:00.123Z。我见过有人直接传time.time()的浮点数结果类型校验直接报错还有人用本地时间字符串造成跨时区Agent判断“超时”时出现几小时的偏差。我的工具函数很简单from datetime import datetime, timezone def now_ts() - str: return datetime.now(timezone.utc).isoformat().replace(00:00, Z)用timezone.utc生成UTC时间再手动把00:00替换成Z和协议示例保持一致。如果你要处理毫秒Pydantic 会自动保留ISO格式里的微秒一般不用额外操心。5.3 枚举值写错导致校验失败TaskState的枚举取值是带连字符的小写字符串例如input-required。有人会写成input_required或InputRequired一解析就报错。这类错误在本地可能不明显一旦数据经过网络层对端严格校验时就会直接返回异常。如果你写了一个函数专门把远端字符串转成TaskState可以这样做from a2a.types import TaskState def safe_state(raw: str): for item in TaskState: if item.value raw: return item raise ValueError(funknown task state: {raw})这样至少报错信息是清晰的。实际上用Task.model_validate_json解析整个响应时Pydantic也会帮你做这一步不过单独做一次状态解析能让你更快定位是哪个字段坏了。5.4 版本差异与升级建议a2a-types跟随A2A协议演进版本变化时可能出现字段增删、枚举调整。我在项目里的做法是在requirements.txt里锁死主版本例如a2a-sdk0.x.y不要直接latest。升级前先跑一轮“构造-序列化-反序列化-断言”的契约测试可以用之前4.5那套自问自答脚本。用Pydantic模型自带的model_json_schema()导出模型结构在写跨语言对接的文档时直接当接口契约用。一旦跨语言方升级了协议版本你这边只要对比schema差异就能很快评估影响面。最后再说一个我实际养成的小习惯每次写Agent服务我都会先建一个protocol_contracts_test.py把AgentCard、Task、Message这三个模型的构造和JSON往返都测一遍。这个测试不依赖网络跑起来几秒钟但能挡住绝大多数因类型字段写错导致的低级问题。A2A协议的概念再多落到Python代码上终究是靠这些Pydantic模型撑起来的把类型层用习惯比背下整本协议文档更实用。