做AI应用开发这几年我越来越觉得一个矛盾很有意思单Agent能力越来越强多Agent协作却一直像在搞外交。不同框架、不同服务之间消息格式各写各的任务状态互相看不懂接口对接全靠临时写胶水代码调试起来更是浪费一整天。A2A协议想解决的就是这个痛点。在Python侧落地这套协议时a2a-types包是你绕不开的第一块拼图。这篇文章就围绕a2a-types的语法、参数体系和实际应用案例把我从项目里趟出来的经验完整分享出来。1. A2A协议背景与a2a-types的定位1.1 为什么Agent之间需要一份“共同语言”单个Agent解决单点问题多Agent协作才能解决复杂流程。比如一个客服Agent需要调用一个订单查询Agent再让一个售后Agent接手这中间如果每个Agent都用自己定义的消息格式双方就要先约定字段命名、嵌套结构、状态枚举。项目小的时候还能靠文档和面对面对接Agent一多这种约定立刻爆炸。A2AAgent2Agent协议做的就是这件事定义Agent之间如何描述自己、如何接收任务、如何回传结果和中间消息。它的目标不是取代特定框架而是成为跨框架的公共通信语言。你可以用LangChain托管自己的Agent也可以完全手写一套服务只要你的Agent遵循A2A的规范其他遵循A2A的Agent就能跟你正常对话。这一点跟HTTP在Web中的作用是类似的。用惯了A2A以后你再回头看没有协议的时期会明显感觉到那种“每天手写对接层”的疲惫是完全没有必要重复的。协议把两件事统一了消息长什么样、任务怎么流转。1.2 a2a-types在Python生态里扮演什么角色A2A协议在Python侧最常用的实现是a2a-sdk而a2a-types就是SDK里的核心类型包实际代码里一般以a2a.types模块的形式被引用。它涵盖了你实现一个A2A服务端或客户端所需的各种数据模型AgentCard、AgentCapabilities、AgentSkill、Task、TaskState、Message、Part、Artifact、Participant等。这些类型大多基于Pydantic构建。为什么用Pydantic而不是普通dataclass因为A2A协议是跨语言的Agent之间传输的永远是JSON。Pydantic能帮你做三层事情第一用声明式语法定义字段和类型第二在运行时强校验收到的JSON数据避免脏数据穿透到业务逻辑第三自动生成JSON Schema可以拿来做契约文档和测试。一句话总结a2a-types的定位它是你写A2A代码时所有数据结构的权威来源。与其各项目在接口文档里手写字段表不如直接引入这套类型让代码本身成为契约。2. a2a-types语法核心从模型定义到消息构造2.1 Pydantic模型与类型注解的基本语法先说最基本的语法规则。a2a-types里的每个核心对象都是一个Pydantic模型声明方式很直观用class继承BaseModel类属性写字段名和类型。from pydantic import BaseModel, Field from enum import Enum from typing import Optional class TaskState(str, Enum): SUBMITTED submitted WORKING working INPUT_REQUIRED input-required COMPLETED completed FAILED failed CANCELED canceled class Task(BaseModel): id: str kind: Optional[str] None status: TaskState input: Optional[Part] None # 任务输入通常是一条消息内容 artifacts: list[Artifact] Field(default_factorylist) history: list[Message] Field(default_factorylist)注意TaskState(str, Enum)这种写法。它让枚举值同时也是字符串序列化成JSON时直接输出submitted而不是TaskState.SUBMITTED这种Python对象表达。这一点在跨语言通信中非常重要因为对方服务大概率不是Python它不认识你的枚举对象只认字符串。字段默认值的定义也有讲究。history和artifacts我用Field(default_factorylist)而不是直接写 []。后者会让所有实例共享同一个list对象某个实例修改时其他实例也会被影响这是Pydantic新手最常见的隐性问题。嵌套类型是另一个关键语法点。Task.input的类型是PartPart本身是抽象基类下面还有TextContent、FileContent、DataContent等具体子类。Pydantic处理这种多态嵌套时会在JSON里留下一个特殊字段标记具体类型反序列化时自动还原成正确的子类实例。这是a2a-types能支持复杂消息结构的基础。2.2 消息内容类型TextContent、FileContent与DataContentA2A协议里消息内容被抽象为Part。一个Message对象它的content就是Part或Part列表。这样设计的意图非常清晰没有把消息内容限定为“纯文本”而是允许文本、文件、结构化数据混合存在。TextContent最常用构造方式很简单from a2a.types import Message, TextContent, Role msg Message( messageIdmsg-001, roleRole.AGENT, contentTextContent(text我已经处理完你的请求结果是...), )FileContent用起来稍复杂一点它需要声明文件的名字和内容来源from a2a.types import FileContent file_part FileContent( namereport.pdf, mimeTypeapplication/pdf, urihttps://storage.example.com/report.pdf, )这里的核心语法点是消息内容是多态的。很多初级项目会直接把Message.content写成string这在A2A里走不通因为协议要求content是一个Part对象而不是裸字符串。理解了这一点再去看a2a-types就顺了不是它非要绕弯子而是它为了兼容结构化场景必须设计成“可扩展的内容单元”。DataContent则用于传结构化数据比如一个JSON对象或量化指标数组。实际项目里我经常用它返回表格型数据让下游Agent直接拿到结构化结果省去一层文本解析。2.3 JSON序列化与反序列化前后端对接的语法基础A2A服务之间的通信走的是JSON所以序列化语法必须熟练。Pydantic v2里常用的有四个方法model_dump()、model_dump_json()、model_validate()、model_validate_json()。# 对象转JSON字符串 json_str task.model_dump_json() # JSON字符串转回对象 task_obj Task.model_validate_json(json_str) # 对象转字典 data task.model_dump()我在项目里发现一个容易踩坑的地方model_dump()默认返回普通Python对象比如datetime还是datetime枚举还是枚举而model_dump_json()会做完整序列化datetime变成ISO8601字符串。如果你们有自己的序列化层一定要想清楚用哪个别混着用。同理在接收外部请求时如果前端已经是一个字典对象用Task.model_validate(data)如果是从请求体拿到的原始字符串用Task.model_validate_json(body)。这两种场景用反了对轻则类型报错重则解出脏数据。3. 关键参数详解把每个字段的意义吃透3.1 AgentCard对外发布能力时的参数A2A体系里每个Agent都要通过AgentCard来描述自己。相当于名片加上服务目录。别人拿到你的AgentCard才知道该不该把任务交给你。核心参数可以列成一个表参数类型是否必填说明namestr是Agent名称最好保持稳定descriptionstr是一句话说清你的Agent能干什么urlstr是服务端点A2A请求入口versionstr是版本号协议升级时用来兼容判断capabilitiesAgentCapabilities否能力声明如是否支持流式响应skillslist[AgentSkill]否具体技能列表含名称和描述providerstr否组织或团队标识iconUrlstr否图标地址用于展示capabilities参数是我最关注的。它本身又是一个嵌套对象典型字段有streaming、pushNotifications。如果你希望客户端能通过SSE实时接收进度必须把streamingTrue声明出来否则客户端会走普通的同步轮询路径。skills字段容易被忽略但它恰恰是Agent能被“找到能力”的关键。A2A协议希望Agent能根据技能描述做任务路由。举个实际例子我有两个Agent一个是“订单查询”一个是“文档摘要”。如果它们的AgentCard都只写个泛泛的description调用方根本不知道哪个处理什么。3.2 Task与TaskState任务状态机的参数约束Task是A2A通信中最核心的工作单元。它涵盖一次请求从提交到完成的全过程。Task的参数除了基础id关键是status字段它严格走状态机。A2A协议里TaskState定义了我上面写过的六个状态submitted、working、input-required、completed、failed、canceled。状态合法流转非常严格我来整理一下当前状态允许转到的状态说明submittedworking、input-required、completed、failed、canceled任务刚创建服务端可以立即开始或拒绝workinginput-required、completed、failed、canceled执行中可能请求补充信息input-requiredsubmitted、working、completed、failed、canceled等待用户补充输入收到后可回到工作状态completed无终态failed无终态canceled无终态参数设计上要注意的是input-required这个状态不是错误状态它是一种正常交互逻辑。当Agent缺信息时服务端返回一个Task状态为input-required同时发送一条Message说明缺什么。客户端拿到这个状态后补充信息再通过重启任务或发送额外输入的方式继续。实践中我经常看到有人把failed当成唯一的错误出口。这种思路会让协作流程变得很呆板遇到缺信息的情况就直接失败下游Agent也没有机会再追问。建议判断好业务逻辑能走input-required就尽量不要直接failed。3.3 Message与Participant消息沿革的关键参数Task.history是一系列Message组成的。每条Message记录一次agent或user之间的交互。它的通用参数包括messageId、role、content、createdAt、contextId、metadata。role只有两个合法值agent和user。这里要特别注意在A2A协议中agent和user是参与方类型。一个Agent服务调用另一个Agent时调用方在消息里的role也是agent。很多开发者在设计时容易混淆这一点总以为调用方就应该是user实际上双方可能都是Agent。Participant类型用来描述参与方信息from a2a.types import Participant agent_participant Participant( namesummarizer-bot, identifierhttps://agent.example.com/summarizer, agentOrUseragent, ) user_participant Participant( nameoperator-01, identifierlocal-user-001, agentOrUseruser, )这些参数在消息审计和历史追溯时特别有用。多Agent协作时一个任务的history可能横跨好几个Agent每一步谁说了什么靠Participant字段能讲清楚。建议大家在写Message时不要偷懒该填createdAt和contextId都填上否则后面定位问题极其痛苦。4. 实际应用案例用a2a-types搭建一个跨Agent文档摘要服务4.1 环境准备与安装不管你是要建服务端还是客户端先安装SDK。以我的经验直接安装a2a-sdk最省事它会连带把a2a.types带进来。pip install a2a-sdk如果你是在已有项目里只想引用类型也可以单独确认一下包是否可用from a2a.types import Task, AgentCard print(Task.model_json_schema())顺利跑通这两行就说明环境没问题。Python版本建议3.10以上a2a-sdk对较新的类型语法依赖比较明显还在用3.8的话容易出现语法兼容问题。4.2 服务端定义AgentCard并处理Task我这次要搭一个“文档摘要Agent”。它的能力很单一接收一段长文本返回不超过三条核心要点的摘要。服务端我用FastAPI承载A2A端点单独挂在路由上。先定义AgentCardfrom a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, ) summarizer_card AgentCard( name文档摘要助手, description输入长文本返回结构化摘要适合会议纪要、文章浓缩、报告提炼, urlhttps://agent.internal.example/summarizer/a2a, version1.0.0, capabilitiesAgentCapabilities(streamingFalse, pushNotificationsFalse), skills[ AgentSkill( idtext-summary, name文本摘要, description对输入文本做三句摘要返回TextContent ) ], )然后实现核心的send task逻辑。这里的要点是接收一个Task从它的input里取出文本内容处理完以后返回一个新的Task对象状态置为completedartifacts里放摘要结果。from fastapi import FastAPI, Request from a2a.types import Task, TaskState, TextContent, Artifact, Message, Role app FastAPI() app.post(/a2a/task/send) async def handle_send(request: Request): payload await request.json() incoming_task Task.model_validate(payload) # 从input取文本a2a-types会把Part解析成具体子类 input_part incoming_task.input text input_part.text if hasattr(input_part, text) else # 这里是核心处理逻辑 summary await produce_summary(text) # 构造完成态Task completed_task Task( idincoming_task.id, kindincoming_task.kind, statusTaskState.COMPLETED, inputincoming_task.input, artifacts[ Artifact( namesummary-result, parts[TextContent(textsummary)], ) ], historyincoming_task.history [ Message( messageIdmsg-summary-001, roleRole.AGENT, contentTextContent(textsummary), ) ], ) return completed_task.model_dump()如果你用SDK提供的A2AServer类整体结构会更高级会把路由、存储、任务并发都处理掉。但即使手写这个POST接口已经能跑通协议的主流程。关键在于每个Task都必须从某个入口请求中通过model_validate构造再保证返回的Task带着同样的id让客户端能对应上。4.3 客户端发起任务、推送消息、获取结果客户端调用A2A服务时你不需要引整个SDK。最直接的办法就是requests发JSON。我习惯先构建Task对象再序列化POST出去。import requests from a2a.types import Task, TaskState, TextContent, Part request_payload Task( idreq-001, kindtext-summary, statusTaskState.SUBMITTED, inputTextContent(text这里放一篇长文...), ).model_dump() resp requests.post( https://agent.internal.example/summarizer/a2a/task/send, jsonrequest_payload, timeout30, ) result Task.model_validate_json(resp.text) if result.status TaskState.COMPLETED: artifact result.artifacts[0] for part in artifact.parts: if hasattr(part, text): print(摘要结果, part.text)这一段看似简单但在真实项目里我提醒三件事。第一别把input的类型写错提交时必须是Part子类对象。第二收到结果后一定要用Task.model_validate_json而不是直接字典索引因为部分字段如artifacts可能不存在Pydantic能帮你兜住边界。第三超时参数一定要显式设置跨Agent调用最常见的问题就是上游卡住下游干等。4.4 扩展流式传输与推送通知的参数配置如果摘要任务比较慢比如要处理几百页PDF同步等待就不合适了。A2A支持两种异步机制SSE流式传输和推送通知。先说SSE。要给AgentCard的capabilities设置streamingTrue然后在服务端增加一个SSE接口。客户端打开连接后服务端可以持续推送Task状态变化。from sse_starlette.sse import EventSourceResponse app.get(/a2a/task/sse) async def stream_task(request: Request): async def event_generator(): for progress in [working, still working, completed]: yield {data: progress} return EventSourceResponse(event_generator())流式参数的核心是streamingTrue这个声明。很多开发者在服务端实现了SSE却忘了在AgentCard里把它打开结果客户端永远走同步轮询等于白做。这个参数在跨系统协作时就是你的“能力广告位”声明了别人才敢按流式的语义去用你的接口。推送通知则是更高级的模式需要客户端提供接收回调的地址服务端处理完任务后主动POST结果到回调地址。我在团队里的建议是内部系统优先用推送外部协作优先用SSE。原因是推送通知需要双方都有公网可达端点安全策略上更麻烦SSE只需要客户端能访问服务端即可部署成本低得多。5. 常见问题与排查技巧实录5.1 Pydantic版本差异导致的兼容问题a2a-types早期依赖Pydantic现在基本都基于Pydantic v2。但很多老项目里还留着v1的代码习惯。最容易出问题的是这两个v1的.dict()在v2中改成了.model_dump()。v1的validator在v2中改成了field_validator。v1的parse_obj在v2中改成了model_validate。如果你在项目里看到AttributeError: Task object has no attribute dict多半就是装了SDK以后Pydantic v1/v2共存或者代码里残留了旧写法。排查时第一步先确认pydantic.__version__然后统一改成v2API。5.2 状态机非法流转任务状态机通信时必须严格按合法方向走。有一次我们服务端在处理一个已经completed的Task时又给它推了一条working状态的消息客户端直接报协议解析失败。排查这类问题建议在TaskStore层加一层状态校验。用什么技术栈无所谓核心逻辑是valid_transitions { TaskState.SUBMITTED: {TaskState.WORKING, TaskState.INPUT_REQUIRED, TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELED}, TaskState.WORKING: {TaskState.INPUT_REQUIRED, TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELED}, TaskState.INPUT_REQUIRED: {TaskState.SUBMITTED, TaskState.WORKING, TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELED}, }每次状态更新前先查表非法跳转直接拒绝。这个小逻辑看似简单在跨Agent协作里能省掉大量“为什么数据被覆盖”的返工。5.3 消息内容类型序列化的坑最常见的序列化报错是这样的ValueError: Error processing field content: expected Part, got str原因几乎都是直接把字符串传给了content字段。A2A里Message.content必须是Part类型或其列表。解决方法是包一层TextContent不要嫌麻烦。反过来反序列化时也要留意。Task.model_validate_json理论上能把嵌套的Part还原成具体子类从而在hasattr(part, text)判断中正确定位。如果你发现part是一个普通字典而不是Part子类说明你的SDK版本没有正确启用多态反序列化或者你在中间层做了不必要的json.loads后再手写dict传值。5.4 参数命名与版本升级A2A作为协议仍然在演进。我在项目里遇到过字段改名的问题比如早期版本里用agentName后来统一成participant.name。一旦SDK升级旧代码构造出的JSON跟新SDK解析逻辑不匹配就会出莫名其妙的校验错误。我的处理方式比较保守项目中固定一个SDK版本升级时做全量回归测试重点抓AgentCard和Task字段。另外我会用Task.model_json_schema()生成一份契约Schema放进项目的契约测试里。对方改了字段测试就会直接失败而不是等到联调时才发现。结语关于a2a-types我最后的几点体会把这个包用了一整轮以后我的感受是a2a-types真正解决的并不只是“少写几个类”的问题。它逼着你把Agent之间的通信边界想清楚——哪些消息是状态变更哪些是业务数据哪些是能力描述。这些边界在自由定义消息格式的项目里很难守住但在A2A这种规范化体系下从一开始就是结构化的。如果非要说一个最值得新手注意的经验那就是先花时间把TaskState状态机和Part消息多态这两块啃透再上手写业务逻辑。语法本身花一个小时就能看完真正影响项目成败的是你对状态流转和消息结构的理解是否到位。后续如果你的Agent规模继续变大建议把AgentCard能力描述也加到自动化巡检里每天跑一次确保线上Agent的行为跟它“名片”上写的保持一致。