用HelloAgents写A2A多Agent系统代码量直接少了一半写在前面上一篇文章我们用原生A2A SDK写了两个Agent的协作代码量不小——要自己管理Agent Card、处理任务状态、维护会话、写HTTP服务……跑通是跑通了但写起来确实有点繁琐。这篇文章我们来看看HelloAgents框架是怎么封装A2A的。说实话我第一次用HelloAgents的A2A工具的时候最大的感受就是省事——很多原生SDK里需要自己写的样板代码框架都帮你封装好了你只需要关注业务逻辑。今天我就带大家看看HelloAgents的A2A工具都提供了什么以及怎么用它快速搭建多Agent系统。中间会对比原生SDK的写法让你直观感受到框架的价值。一、HelloAgents对A2A做了哪些封装在动手之前先看看HelloAgents在A2A方面提供了哪些工具。我整理了一下主要有这几块1. A2AAgent类这是最核心的类。你只需要继承它实现几个回调方法一个完整的A2A服务端Agent就写好了。不需要自己管HTTP服务、路由、Agent Card生成这些东西。2. A2AClient类封装了A2A客户端的所有操作——连接、发消息、创建任务、订阅状态。比原生客户端用起来更顺手错误处理也更完善。3. AgentRegistryAgent注册中心这是HelloAgents的一个亮点。它提供了一个内置的Agent注册中心Agent启动时自动注册其他Agent可以通过能力描述发现并连接它。原生A2A SDK里没有这个你得自己搞服务发现。4. MultiAgentOrchestrator多Agent编排器如果你需要协调多个Agent完成一个复杂任务这个编排器能帮你管理任务分解、分发、汇总。不用自己写调度逻辑。5. 内置的会话和状态管理HelloAgents自动帮你管理会话状态、任务状态、上下文传递。原生SDK里这些都得自己维护很容易出bug。这几块加起来基本上把多Agent系统里的脏活累活都包了。你只需要写每个Agent的业务逻辑剩下的交给框架。二、用HelloAgents写一个A2A服务端Agent我们先从最简单的开始——用HelloAgents写一个A2A服务端Agent。跟上一篇的计算器Agent做同样的事你可以对比一下代码量。# calculator_agent.pyfromhelloagentsimportA2AAgent,TaskResultclassCalculatorAgent(A2AAgent):name计算器助手description一个简单的计算器智能体支持加减乘除capabilities[calculation]asyncdefhandle_task(self,task):处理任务只需要关注业务逻辑try:afloat(task.input.get(a,0))bfloat(task.input.get(b,0))optask.input.get(op,)ifop:resultabelifop-:resulta-belifop*:resulta*belifop/:resulta/bifb!0elsefloat(inf)else:returnTaskResult.error(f不支持的运算符{op})returnTaskResult.success({result:result,expression:f{a}{op}{b}{result}})exceptExceptionase:returnTaskResult.error(str(e))# 启动Agent一行代码搞定if__name____main__:CalculatorAgent().run(host0.0.0.0,port8000)对比一下上一篇原生SDK的写法你会发现不需要手动定义Agent Card类属性自动生成不需要手动创建HTTP服务run()方法搞定不需要手动管理Task状态返回TaskResult就行不需要自己处理异常框架有默认错误处理代码量少了差不多一半而且更清晰——你一眼就能看出这个Agent是干什么的业务逻辑跟协议细节完全分离了。三、用HelloAgents写A2A客户端客户端的封装也很到位。我们来看看怎么用HelloAgents的A2AClient调用上面的计算器Agent# client.pyfromhelloagentsimportA2AClientimportasyncioasyncdefmain():# 创建客户端并连接clientawaitA2AClient.connect(http://localhost:8000)# 自动获取Agent Card打印对方信息print(f已连接{client.agent_card.name})print(f能力{client.agent_card.capabilities})# 创建任务一行代码resultawaitclient.create_task_and_wait(input{a:15,b:27,op:},timeout30)ifresult.success:print(f计算结果{result.output[expression]})else:print(f任务失败{result.error})awaitclient.close()if__name____main__:asyncio.run(main())注意这个create_task_and_wait方法——它把创建任务 → 轮询状态 → 获取结果这三步合成了一步。原生SDK里你得自己写循环轮询这里直接帮你封装好了。如果你不想等任务完成只想创建然后异步处理也可以用create_task# 创建任务不等待taskawaitclient.create_task(input{a:1,b:2})# 订阅状态变化asyncforeventinclient.subscribe_task(task.id):print(f状态{event.status})ifevent.statusin[completed,failed]:break发消息也很简单# 发送普通消息replyawaitclient.send_message(你好帮我算个东西)print(f回复{reply.content})四、多Agent协作的完整例子上面的例子都是单Agent调用我们来看看HelloAgents怎么处理多Agent协作。假设我们要做一个旅行规划系统有三个Agent旅行规划Agent接收用户需求分解任务协调其他Agent机票Agent查询和预订机票酒店Agent查询和预订酒店用HelloAgents的MultiAgentOrchestrator来实现# travel_system.pyfromhelloagentsimportA2AAgent,A2AClient,MultiAgentOrchestrator,TaskResult# 机票AgentclassFlightAgent(A2AAgent):name机票助手description查询和预订机票capabilities[flight_search,flight_booking]asyncdefhandle_task(self,task):iftask.typesearch:returnTaskResult.success({flights:[{flight:CA1501,time:08:30,price:580},{flight:MU5101,time:10:15,price:620}]})eliftask.typebook:returnTaskResult.success({order_id:FL12345,status:confirmed})returnTaskResult.error(未知任务类型)# 酒店AgentclassHotelAgent(A2AAgent):name酒店助手description查询和预订酒店capabilities[hotel_search,hotel_booking]asyncdefhandle_task(self,task):iftask.typesearch:returnTaskResult.success({hotels:[{name:如家酒店,price:299,rating:4.5},{name:汉庭酒店,price:259,rating:4.3}]})eliftask.typebook:returnTaskResult.success({order_id:HT67890,status:confirmed})returnTaskResult.error(未知任务类型)# 旅行规划Agent协调者classTravelPlannerAgent(A2AAgent):name旅行规划助手description帮助用户规划完整旅行方案capabilities[travel_planning]def__init__(self):super().__init__()# 在初始化时连接其他Agentself.flight_clientNoneself.hotel_clientNoneasyncdefon_start(self):Agent启动时的回调用来建立连接self.flight_clientawaitA2AClient.connect(http://localhost:8001)self.hotel_clientawaitA2AClient.connect(http://localhost:8002)asyncdefhandle_task(self,task):destinationtask.input.get(destination)daystask.input.get(days,1)# 并行调用机票和酒店Agentflight_taskself.flight_client.create_task_and_wait(input{type:search,destination:destination})hotel_taskself.hotel_client.create_task_and_wait(input{type:search,destination:destination,days:days})flight_result,hotel_resultawaitasyncio.gather(flight_task,hotel_task)# 汇总结果returnTaskResult.success({destination:destination,flights:flight_result.output[flights],hotels:hotel_result.output[hotels],summary:f为您找到{len(flight_result.output[flights])}个航班和{len(hotel_result.output[hotels])}家酒店})# 启动三个Agentif__name____main__:importasyncioasyncdefstart_all():awaitFlightAgent().start(port8001)awaitHotelAgent().start(port8002)awaitTravelPlannerAgent().start(port8000)asyncio.run(start_all())这个例子展示了HelloAgents的几个强大之处on_start回调Agent启动时自动执行可以用来建立连接、加载资源不用自己管生命周期。并行调用用asyncio.gather可以同时调用多个Agent框架层面完全支持异步。任务结果统一所有Agent的返回都是TaskResult处理起来很统一。启动简单每个Agent调用start()就能跑端口配置一下就行。五、Agent注册中心自动发现的魔力上面的例子里旅行规划Agent是硬编码了机票和酒店Agent的地址localhost:8001和localhost:8002。但在真实系统中Agent的地址可能会变数量也可能动态增减。HelloAgents提供了AgentRegistry来解决这个问题# 启动注册中心fromhelloagentsimportAgentRegistry registryAgentRegistry()awaitregistry.start(port9000)# Agent启动时自动注册classFlightAgent(A2AAgent):name机票助手registry_urlhttp://localhost:9000# 指定注册中心地址# ... 其他代码# 客户端通过注册中心发现AgentclientawaitA2AClient.discover(registry_urlhttp://localhost:9000,capabilityflight_search# 按能力查找)有了注册中心之后Agent启动时自动注册关闭时自动注销客户端不需要知道具体地址按能力查找就行支持多个同能力Agent的负载均衡Agent上下线对客户端透明这个功能在多Agent系统里太重要了。原生A2A SDK完全没有服务发现你得自己集成Consul、etcd之类的东西HelloAgents直接内置了。六、跟原生SDK的对比用了一段时间HelloAgents的A2A工具之后我做了一个跟原生SDK的对比维度原生A2A SDKHelloAgents A2A工具创建服务端Agent需要手动写HTTP服务、路由、Agent Card继承A2AAgent实现回调即可任务状态管理自己维护状态机框架自动管理返回TaskResult会话管理自己用session_id关联框架自动维护客户端调用创建任务→轮询→取结果三步create_task_and_wait一步到位服务发现无需自己集成内置AgentRegistry多Agent编排自己写调度逻辑内置MultiAgentOrchestrator错误处理自己try-catch统一异常处理TaskResult.error代码量相同功能约100行约40-50行学习曲线较陡需要理解协议细节平缓关注业务逻辑即可当然原生SDK也有它的优势——更灵活、更轻量、不依赖框架。如果你只需要一个非常简单的A2A通信原生SDK也够用。但如果要做稍微复杂一点的多Agent系统HelloAgents的封装能省很多事。七、我踩的几个坑用HelloAgents的A2A工具也不是完全一帆风顺说几个我遇到的问题坑1Agent的name不能重复。注册中心是按name来标识Agent的如果两个Agent叫同一个名字注册会冲突。我一开始图省事两个测试Agent都叫测试助手结果第二个注册不上。后来改成不同名字就好了。坑2capabilities要写准确。客户端按capability查找Agent的时候是精确匹配的。如果你注册的是flight_search客户端查flight是查不到的。建议统一命名规范比如用领域.动作的格式flight.search、flight.book。坑3长任务要设置合理的timeout。create_task_and_wait默认超时是30秒。如果你的任务处理时间比较长比如调用外部API、跑数据分析一定要手动设大一点的timeout否则会超时报错。坑4Agent之间的循环调用要小心。A调用BB又调用A这种循环调用如果没有终止条件会死循环。框架目前没有自动检测循环调用的机制需要自己在业务逻辑里注意。写在最后HelloAgents的A2A工具给我最大的感受是恰到好处的封装。它没有把A2A协议藏得太深——你依然能理解底层在发生什么但它把那些重复的、容易出错的样板代码都包了让你专注于业务逻辑。我觉得这就是好框架应该有的样子不是替你做所有决定而是把那些没技术含量但又必须做的事情做掉让你把精力花在真正有价值的地方。当然框架也不是银弹。特别复杂的场景、需要深度定制协议行为的场景可能还是得用原生SDK。但对于90%的多Agent应用场景HelloAgents的封装已经足够好了。下一篇文章我们会聊聊怎么在智能体中真正使用A2A工具——包括大模型怎么决定调用哪个Agent、怎么处理Agent返回的结果、多Agent协作中的错误恢复等。你们在用HelloAgents或者其他Agent框架的时候有没有觉得哪些封装特别好用或者哪些地方不够灵活欢迎在评论区聊聊。下一篇预告在智能体中使用A2A工具。我们会把A2A跟大模型真正结合起来看看Agent是怎么自主决定委托任务、跟其他Agent协商、处理协作结果的。