企业官网建设流程全解析

A2A协议实战：手把手教你让两个Agent真正"对话"起来

写在前面

上一篇文章讲了A2A的设计动机——为什么需要Agent之间的协作协议。理论讲完了，今天该动手了。

说实话，我第一次写A2A代码的时候，踩了不少坑。A2A的规范跟MCP很不一样，概念也更多，什么Agent Card、Task、Artifact、Message……光看文档有点晕。后来我写了几个小demo，慢慢才理清楚。

这篇文章我会从最基础的开始，一步步带你写两个A2A Agent，让它们真正"对话"起来。中间会穿插我踩过的坑和一些理解，希望能帮你少走弯路。

一、先搞清楚A2A的几个核心概念

动手之前，先把A2A的几个核心概念搞明白。A2A的概念比MCP多，不搞清楚的话写代码会很懵。

1. Agent Card（智能体名片）

每个A2A Agent都有一张"名片"，叫Agent Card。这是一个JSON文件，描述了Agent的基本信息：名称、描述、能力、支持的协议版本、端点地址等。

其他Agent想要跟你通信，首先要获取你的Agent Card，知道你是谁、能做什么、怎么联系你。

{"name":"订单查询助手","description":"专门处理订单查询和退换货的智能体","url":"https://api.example.com/order-agent","version":"1.0.0","capabilities":["order_query","refund","exchange"]}

你可以把Agent Card理解成一个人的"简历"——别人通过简历了解你能做什么，然后决定要不要找你合作。

2. Message（消息）

A2A的基本通信单位是Message。一条消息包含发送者、接收者、消息类型、内容等信息。跟HTTP请求不同，A2A的消息是双向的、可以多轮的。

消息有几种类型：

• text：普通文本消息
• part：结构化内容（比如任务卡、结果）
• metadata：元数据（比如状态更新、进度通知）

3. Task（任务）

Task是A2A里一个很重要的概念。当一个Agent委托另一个Agent做一件事时，就创建一个Task。Task有状态（pending、working、completed、failed等），可以跟踪进度。

你可以把Task想象成公司里的"工单"——你给同事派一个工单，同事处理，你可以随时查看工单状态，处理完了会通知你。

4. Artifact（产物）

Task完成后，会产生Artifact，也就是任务的产出物。可以是文本、文件、链接、结构化数据等。

这四个概念是A2A的基石。后面的代码都是围绕它们展开的。

二、环境准备

A2A目前官方提供了Python和Node.js的SDK。我用Python来演示，因为大部分做Agent的同学都熟悉Python。

# 创建虚拟环境python-mvenv a2a-envsourcea2a-env/bin/activate# 安装A2A SDKpipinstalla2a

这里我踩了第一个坑：a2a这个包名跟另一个不相关的包冲突了，安装的时候要注意看清楚。官方的包名是google-a2a或者a2a-sdk，具体以官方文档为准。我安装的时候用的是：

pipinstalla2a-sdk

如果安装失败，先升级pip和setuptools：

pipinstall--upgradepip setuptools

三、写第一个A2A Agent（服务端）

我们先写一个最简单的A2A Agent——一个"加法计算器Agent"，它能接收两个数字，返回它们的和。虽然简单，但包含了A2A服务端的完整结构。

# calculator_agent.pyfroma2aimportA2AApp,AgentCard,Task,Messageimportuvicorn# 定义Agent Cardagent_card=AgentCard(name="计算器助手",description="一个简单的计算器智能体，支持加法运算",version="1.0.0",capabilities=["addition"])# 创建A2A应用app=A2AApp(agent_card=agent_card)# 处理任务创建@app.taskasyncdefhandle_task(task:Task):"""当有新任务创建时调用"""print(f"收到新任务：{task.id}")print(f"任务输入：{task.input}")# 解析输入try:a=float(task.input.get("a",0))b=float(task.input.get("b",0))result=a+b# 任务完成，返回结果task.status="completed"task.output={"result":result,"expression":f"{a}+{b}={result}"}exceptExceptionase:task.status="failed"task.output={"error":str(e)}returntask# 处理消息（用于多轮对话）@app.messageasyncdefhandle_message(message:Message):"""当收到普通消息时调用"""print(f"收到消息：{message.content}")returnMessage(content="收到你的消息了！")if__name__=="__main__":uvicorn.run(app,host="0.0.0.0",port=8000)

这段代码做了几件事：

1. 定义了Agent Card，描述这个Agent是谁、能做什么
2. 创建了A2A应用，它会自动处理A2A协议的握手、路由等
3. 注册了任务处理器——当有新任务创建时，自动调用这个函数
4. 注册了消息处理器——当收到普通消息时，自动回复

运行这个Agent：

python calculator_agent.py

它会在8000端口启动一个HTTP服务，其他Agent可以通过http://localhost:8000访问它。

这里我踩了第二个坑：A2A的服务端默认用的是HTTP传输，但规范里也支持其他传输方式。如果你用的是旧版SDK，可能需要手动配置路由。建议直接用最新版的SDK，很多东西都封装好了。

四、写A2A客户端（调用方）

服务端写好了，接下来写客户端去调用它。客户端的代码更简单：

# client.pyfroma2aimportA2AClient,Taskimportasyncioasyncdefmain():# 创建客户端，连接到计算器Agentclient=A2AClient(base_url="http://localhost:8000")# 第一步：获取Agent Card，确认对方身份和能力card=awaitclient.get_agent_card()print(f"连接到Agent：{card.name}")print(f"描述：{card.description}")print(f"能力：{card.capabilities}")# 第二步：创建一个任务task=Task(input={"a":15,"b":27},description="计算15加27的结果")created_task=awaitclient.create_task(task)print(f"任务已创建：{created_task.id}")print(f"任务状态：{created_task.status}")# 第三步：等待任务完成（轮询状态）whilecreated_task.statusin["pending","working"]:awaitasyncio.sleep(0.5)created_task=awaitclient.get_task(created_task.id)print(f"任务状态：{created_task.status}")# 第四步：获取结果ifcreated_task.status=="completed":print(f"计算结果：{created_task.output['expression']}")print(f"结果值：{created_task.output['result']}")else:print(f"任务失败：{created_task.output.get('error')}")awaitclient.close()if__name__=="__main__":asyncio.run(main())

运行客户端（确保服务端已经启动）：

python client.py

你应该能看到类似这样的输出：

连接到Agent：计算器助手 描述：一个简单的计算器智能体，支持加法运算 能力：['addition'] 任务已创建：task_abc123 任务状态：working 任务状态：completed 计算结果：15.0 + 27.0 = 42.0 结果值：42.0

看到这个输出，就说明你的第一个A2A通信跑通了！两个Agent通过A2A协议完成了一次任务协作。

五、多轮对话：让Agent之间"聊起来"

上面的例子是一次性的任务委托——客户端发一个任务，服务端处理完返回结果。但A2A更强大的地方在于支持多轮对话。

我们来写一个更复杂的例子：一个"旅行规划Agent"和一个"机票预订Agent"之间的多轮协作。

# travel_agent.py（客户端，主动发起对话）froma2aimportA2AClient,Messageimportasyncioasyncdefmain():client=A2AClient(base_url="http://localhost:8001")# 机票Agent的地址# 发送第一条消息，咨询机票msg1=Message(content="你好，我想订一张明天从北京到上海的机票")reply1=awaitclient.send_message(msg1)print(f"机票Agent：{reply1.content}")# 根据回复，提供更多信息msg2=Message(content="经济舱，上午的航班优先")reply2=awaitclient.send_message(msg2)print(f"机票Agent：{reply2.content}")# 确认预订msg3=Message(content="好的，就订这个航班，乘客张三，身份证号110xxxx")reply3=awaitclient.send_message(msg3)print(f"机票Agent：{reply3.content}")awaitclient.close()if__name__=="__main__":asyncio.run(main())

服务端（机票Agent）需要维护对话状态：

# flight_agent.py（服务端，处理多轮对话）froma2aimportA2AApp,AgentCard,Messageimportuvicorn agent_card=AgentCard(name="机票预订助手",description="帮助用户查询和预订机票",version="1.0.0",capabilities=["flight_search","flight_booking"])app=A2AApp(agent_card=agent_card)# 用字典保存每个会话的状态sessions={}@app.messageasyncdefhandle_message(message:Message):session_id=message.session_id# 初始化会话ifsession_idnotinsessions:sessions[session_id]={"step":0,"data":{}}session=sessions[session_id]# 根据对话轮次回复ifsession["step"]==0:session["step"]=1returnMessage(content="好的，请问您对舱位和时间有什么偏好吗？")elifsession["step"]==1:session["step"]=2session["data"]["preference"]=message.content# 模拟查询航班returnMessage(content="为您找到：国航CA1501，明天08:30起飞，经济舱票价580元。需要预订吗？")elifsession["step"]==2:session["step"]=3# 模拟预订returnMessage(content="预订成功！订单号FL20261008001，已发送确认邮件。祝您旅途愉快！")returnMessage(content="请问还有什么可以帮您？")if__name__=="__main__":uvicorn.run(app,host="0.0.0.0",port=8001)

这个例子展示了A2A的多轮对话能力。两个Agent之间可以像人一样来回沟通，逐步澄清需求，最终完成任务。这在MCP的工具调用模式下是很难实现的。

六、任务委托与状态跟踪

多轮对话是一种协作方式，任务委托是另一种。任务委托更正式，有明确的状态跟踪，适合比较重的任务。

A2A的Task有这些状态：

• pending：任务已创建，等待处理
• working：任务正在处理中
• completed：任务已完成
• failed：任务失败
• canceled：任务被取消

客户端可以随时查询任务状态，也可以取消任务。服务端可以在处理过程中更新状态、发送进度通知。

一个完整的任务委托流程：

# 客户端asyncdefdelegate_task():client=A2AClient(base_url="http://localhost:8000")# 创建任务task=awaitclient.create_task(Task(input={"url":"https://example.com/article"},description="总结这篇文章的主要内容"))print(f"任务创建：{task.id}, 状态：{task.status}")# 订阅任务状态变化（而不是轮询）asyncforeventinclient.subscribe_task(task.id):print(f"状态更新：{event.status}")ifevent.status=="completed":print(f"结果：{event.output['summary']}")breakelifevent.status=="failed":print(f"失败：{event.output.get('error')}")breakawaitclient.close()

用事件订阅代替轮询，效率更高，也更优雅。

七、我踩过的坑

最后说几个我在写A2A代码时踩的坑，帮大家避雷：

坑1：Agent Card的URL要写对。

Agent Card里的url字段是其他Agent连接你的地址。如果你写的是localhost，那其他机器上的Agent就连不上你。部署的时候一定要改成可访问的域名或IP。

坑2：Task的input/output必须是JSON可序列化的。

不能传Python对象、函数、datetime等，只能传字典、列表、字符串、数字这些基本类型。需要传复杂对象的话，先转成字典。

坑3：会话状态要自己管理。

A2A协议本身不维护会话状态，你需要自己用session_id来关联同一个对话的消息。我一开始以为SDK会自动管理，结果发现每个消息都是独立的，状态得自己存。

坑4：错误处理要做全。

网络超时、对方Agent挂了、任务失败……各种异常都要考虑到。特别是任务委托，一定要处理failed状态，不能假设任务一定会成功。

坑5：SDK版本变化快。

A2A还在快速演进中，不同版本的SDK API差异很大。我写的代码基于某个版本，过了一个月升级SDK之后发现好几个函数改名了。建议固定版本号，不要盲目升级。

写在最后

A2A的实战入门比MCP稍微复杂一些，因为概念更多、协作模式更丰富。但一旦理解了Agent Card、Message、Task这几个核心概念，写起来就很顺了。

今天的例子都比较简单，主要是帮大家建立感性认识。实际项目中，你可能需要处理更复杂的场景——多个Agent并行协作、任务分解和汇总、跨组织的Agent通信等等。这些我们后面会慢慢讲。

我的建议是：先把今天的代码跑通，然后试着改一改——比如把计算器改成支持加减乘除，或者加一个第三个Agent进来协作。动手改代码是理解协议最好的方式。

下一篇文章我们会看看HelloAgents框架是怎么封装A2A的，以及怎么用HelloAgents的A2A工具快速构建多Agent系统。

你们在写A2A代码的时候遇到了什么问题？欢迎在评论区留言，我会尽量解答。

下一篇预告：使用HelloAgents A2A工具。我们会看看HelloAgents框架对A2A做了哪些封装，怎么用它的工具类快速搭建多Agent协作系统。