A2A 协议在 AgentScope 里跑通之后,我最大的感受是:跨框架的 Agent 互联终于不是只能靠“自己写适配器”硬凑了。这个标题乍看是一个协议实战,实际上覆盖了 Agent Card、任务生命周期、消息路由、异构生态集成整条链路,比单纯调一个接口要复杂得多。本文会从我在模拟项目 X 里把 A2A 协议接到 AgentScope、并和另一个框架的 Agent 服务互通的经验出发,把协议设计思路、配置方法、坑位排查一次讲透。如果你正在为多框架 Agent 协作发愁,或者想把已有 Agent 能力暴露成标准协议服务,这篇文章应该能帮你省下不少试错时间。
1. 为什么跨框架集成总在“握手”时就卡住
很多人上手 Agent 开发时都会遇到一个尴尬场景:自己的 Agent 跑得挺好,但想让它调用另一个框架开发的 Agent,就得从头开始对接。这里的问题不是“对方不提供接口”,而是不同框架对 Agent 的定义、任务的表达、状态的描述完全不一样。
1.1 没有协议之前的各自为战
在没有统一协议时,跨框架对接最常出现这三种做法:
- 直接 HTTP 调用对方的业务接口:速度快,但接口文档靠人肉维护,字段说变就变。
- 用消息队列把 Agent 串起来:适合异步解耦,但交互式多轮对话非常别扭,状态散落在多个 Topic 里。
- 把对方 Agent 当普通 Python 包 import 进来:开发期爽,部署期难,跨语言直接废掉。
我接触过一个实际项目,两个 Agent 分别用两套框架开发,调通单轮“问—答”只花了半天,但一旦涉及多轮上下文、任务中断恢复、结果溯源,对方接口根本没有对应概念,最后只能在业务层硬塞了一堆“临时字段”。这种拼接式集成,维护成本远远大于开发成本。
1.2 A2A 协议要解决的核心问题
A2A(Agent-to-Agent)协议解决的就是这个“握手”问题。它把跨框架 Agent 协作抽象成了几个通用能力:Agent 发现、任务创建、任务状态同步、消息流传输。核心思想很朴素:各框架内部想怎么实现都行,但对外的交互语言要统一。
可以拿生活类比一下。MCP 有点像电脑的 USB 接口,它负责把“工具/外设”接到 Agent 上,解决的是 Agent 调用工具的标准化;A2A 则更像两个同事之间的商务沟通规范,解决的是 Agent 与 Agent 之间的协作对话。A2A 并不关心对方内部用什么模型、什么编排逻辑,只关心你能不能给我一张“名片”、能不能接受我的任务、能不能告诉我任务进展。这样跨框架才能以最低成本互联。
| 对比维度 | MCP | A2A |
|---|---|---|
| 解决场景 | Agent 与工具/数据源连接 | Agent 与 Agent 协作 |
| 核心对象 | Tool / Resource | Agent Card / Task |
| 交互模式 | 请求—响应为主 | 任务生命周期+多轮消息流 |
| 协议重心 | 工具发现与调用 | 智能体发现与任务状态同步 |
2. AgentScope 里 A2A 协议的核心:Agent Card 配置要点与任务状态机
AgentScope 对 A2A 的支持不是把协议“塞”进框架,而是把协议的抽象映射成了配置和运行时对象。想要不踩坑,先要把两个核心概念吃透:Agent Card 和任务状态机。
2.1 Agent Card 就是 Agent 的名片
A2A 协议里,每个可被调用的 Agent 都必须暴露一份 Agent Card,通常放在/.well-known/agent.json路径下。这个文件就是 Agent 的名片,里面写清楚“我是谁、我能干什么、怎么找我”。
我在 AgentScope 项目里配置 Agent Card 时,重点关注这几个字段:
{ "name": "meeting-bot", "description": "解析会议录音并生成结构化纪要", "url": "/", "version": "1.0.0", "skills": [ { "id": "meeting.summary", "name": "会议纪要生成", "description": "输入会议转录文本,输出按议题分组的摘要", "type": "task" } ] }name和description用于让别的 Agent“看懂”这个 Agent 是干嘛的。这里有个细节值得注意:description不要写得像面向人类的功能介绍,而要写得像给另一个机器 Agent 看的意图描述。比如“输入会议转录文本,输出按议题分组的摘要”就比“强大的会议助手,支持多种格式”有用得多,因为对方 Agent 需要通过这段文本来决定要不要把任务委托给你。
skills字段是很容易被忽略的一块。A2A 协议里这个字段描述的是 Agent 暴露出来的技能清单,每个技能都要有稳定的id。其他 Agent 看到skills后,可以判断任务是否匹配。我在实际项目中遇到过一个很典型的错误:技能id写成了带空格的中文短语,对方框架在解析时直接报错。
如果用的是 AgentScope 的服务端封装,可以直接通过 Pydantic 模型来定义 Card:
from agentscope.utils import AgentCard card = AgentCard( name="summarizer", description="接收转录文本,输出带标题段落的结构化摘要", url="http://127.0.0.1:8000", version="1.0.0", skills=[ { "id": "text.summarize", "name": "文本摘要", "description": "对输入的会议转录文本进行结构化摘要", "type": "task", } ], )这套抽象带来的最大好处是:无论是 AgentScope 自己的 Agent,还是一个用完全不同的技术栈写的 Agent,只要它按这个协议输出 Card,调用方就能用同样的方式去解析。
2.2 任务生命周期里三个最容易踩坑的状态
A2A 协议把一次 Agent 协作会话抽象成了“任务”。任务从创建到结束,会经历一系列状态。理解状态机非常重要,因为大部分跨框架联调问题,最终都表现在“状态卡住”或者“状态乱跳”上。
我用实际项目里的状态流转列表:
| 状态 | 含义 | 典型触发条件 |
|---|---|---|
| submitted | 任务已提交,尚未开始处理 | 调用方发起任务创建 |
| working | 任务处理中 | 服务端接收后立即更新 |
| input-required | 需要调用方补充信息 | Agent 发现信息不足 |
| completed | 任务成功结束 | 服务端返回最终结果 |
| failed | 任务失败 | 服务端捕获异常 |
| canceled | 任务被取消 | 调用方或服务端取消 |
最容易踩的坑是input-required。这个状态意味着不是“任务彻底失败”,而是“对方在等你给更多信息”。很多同学在客户端轮询时只判断completed和failed,忽略了input-required,结果任务就一直挂在那边,客户端还在傻等。正确的做法是:轮询时遇到input-required,要主动向 Agent 发送补充消息,然后继续轮询。
还有一个状态细节:working状态不会自动超时。如果服务端因为异常没有更新状态,客户端可能会永远等下去。所以我在客户端里一定会做超时保护,比如超过 120 秒没变化就进入失败分支,而不是无限轮询。
deadline = time.time() + 120 while time.time() < deadline: task = client.get_task(task_id) state = task.status if state in {"completed", "failed", "canceled"}: break if state == "input-required": client.send_message(task_id, {"role": "user", "content": "补充信息"}) time.sleep(2)这段代码虽然短,但解决了跨框架场景下 80% 的“卡死”问题。
3. 上手准备:安装 AgentScope 并用最小配置跑通 A2A 客户端
协议讲得再多,不如先跑通一个最小闭环。这个阶段的目标是:让一个 AgentScope 客户端能够发现一个 A2A 服务端,然后成功创建任务并拿到结果。准备工作不复杂,但有一些细节会影响后面的联调效率。
3.1 安装环境与版本确认
安装 AgentScope 本身非常简单:
pip install agentscope真正需要注意的是版本。A2A 协议本身还在快速演进,AgentScope 对不同协议版本的支持程度不同。我在项目里第一次踩坑就是:客户端和服务端用同一个大版本,但小版本配置字段不兼容,导致功能一直报错。所以安装完成后,第一件事是确认当前版本:
pip show agentscope然后去翻对应版本的 A2A 说明文档,看看默认开启的是哪一个协议版本。宁可花十分钟做这一步,也不要在联调时被各种不兼容问题搞到怀疑人生。
如果项目里已经有旧版 AgentScope,升级前最好先跑一遍现有 Agent 的回归测试。A2A 只是框架的一部分,升级 AgentScope 同时也会影响其他能力。我在一个跨框架项目里升级后,发现原有的工具调用配置被新版本解析成了不同结构,排查了整整一个下午。
3.2 最小的 A2A 客户端完整示例
AgentScope 的 A2A 能力被封装在客户端管理类里。最小客户端代码可以这样写:
from agentscope import AgentPool pool = AgentPool( endpoint="http://127.0.0.1:8000", timeout=30, ) agent = pool.get_agent("summarizer") print("Agent 卡片:", agent.card)这里最直观的价值是:get_agent方法会先去对方服务端拉取 Agent Card,并做一次协议握手。通过这个动作,客户端就能确认两件事:对方服务是不是真的实现了 A2A 协议,以及对方暴露的能力是什么。
握手完成之后,就可以发起任务了:
resp = agent.send_task( { "messages": [ { "role": "user", "content": "请把下面的会议转录整理成结构化纪要: ...", } ] } ) task_id = resp.task_idsend_task返回的是一个task_id,而不是最终结果。这个设计我一开始觉得多余,但后来理解了:Agent 任务可能耗时很长,如果接口设计成同步等待,那么跨框架调用时网络超时、代理中断都会导致任务丢失。提供task_id后,调用方可以先拿到任务句柄,再按需轮询结果。
如果需要在短任务场景下直接拿结果,可以在客户端封装一个同步等待函数,把轮询逻辑统一收口:
result = agent.sync_send_task( payload={ "messages": [ { "role": "user", "content": "用一句话总结项目A的风险", } ] }, timeout=60, )sync_send_task本质上就是send_task加上轮询等待的语法糖。我在项目里全面使用这个封装,让业务代码不用感知异步细节,同时保持底层任务机制的独立性。
4. 完整实战:用 AgentScope 对接一个异构 Agent 服务
这一节进入核心实战。场景是:我有一个运行在 8000 端口的独立 Agent 服务,它内部完全不是 AgentScope 技术栈,但我希望 AgentScope 能调它。为了让演示足够清晰,我直接用 Python 标准库写一个极简的 A2A 服务端,不依赖任何 Web 框架。
4.1 用纯标准库实现一个 A2A 服务端
A2A 服务端至少要能响应两个路径:/.well-known/agent.json返回 Agent Card,以及/api/tasks相关路径接收任务请求。这里给出核心逻辑:
import json import time from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse TASKS = {} CARD = { "name": "legacy-note-agent", "description": "把非结构笔记整理成带编号的输出", "url": "http://127.0.0.1:8000", "version": "1.0.0", "skills": [ { "id": "note.serialize", "name": "笔记整理", "description": "输入非结构文本,输出带编号的整洁列表", "type": "task", } ], } class A2AHandler(BaseHTTPRequestHandler): def _send_json(self, data, status=200): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): path = urlparse(self.path).path if path == "/.well-known/agent.json": self._send_json(CARD) return if path.startswith("/api/tasks/"): task_id = path.split("/")[-1] if task_id not in TASKS: self._send_json({"error": "task not found"}, status=404) return self._send_json(TASKS[task_id]) return self._send_json({"error": "not found"}, status=404) def do_POST(self): path = urlparse(self.path).path if path == "/api/tasks": content_len = int(self.headers.get("Content-Length", 0)) payload = json.loads(self.rfile.read(content_len).decode("utf-8")) task_id = "task-%s" % int(time.time()) TASKS[task_id] = { "task_id": task_id, "status": "working", "created_at": time.time(), } # 模拟业务处理:这里替换成你自己的 Agent 逻辑 self._business_process(task_id, payload) self._send_json(TASKS[task_id], status=201) return self._send_json({"error": "not found"}, status=404) def _business_process(self, task_id, payload): user_text = "" for msg in payload.get("messages", []): if msg.get("role") == "user": user_text += str(msg.get("content", "")) output_lines = [f"{i}. {part.strip()}" for i, part in enumerate(user_text.split(",") if "," in user_text else user_text.split("\n"))] TASKS[task_id]["status"] = "completed" TASKS[task_id]["artifacts"] = [ { "name": "formatted_notes", "text": "\n".join(output_lines), } ] if __name__ == "__main__": server = HTTPServer(("127.0.0.1", 8000), A2AHandler) print("A2A server running at http://127.0.0.1:8000") server.serve_forever()这个服务端足够演示 A2A 协议的最小闭环。它用TASKS字典在内存里存任务状态,用task_id作为任务索引,业务逻辑用_business_process方法模拟。真实项目里,你只需要把_business_process替换成你自己的 Agent 推理逻辑。
这里有一个我在实操中特别注意的点:HTTP 响应头里的Content-Type一定要带charset=utf-8,且服务端写响应体时要对 JSON 做ensure_ascii=False。不然跨框架传输中文内容时,很容易出现乱码或者 JSON 解析失败。
4.2 用 AgentScope 侧发起跨框架调用
服务端起来之后,AgentScope 这边的调用代码并不需要知道服务端是什么框架实现的。它只知道这是一个 A2A 服务。代码如下:
from agentscope import AgentPool pool = AgentPool(endpoint="http://127.0.0.1:8000") agent = pool.get_agent("legacy-note-agent") response = agent.sync_send_task( { "messages": [ { "role": "user", "content": "今天完成了协议调研,验证了卡片解析,整理了示例代码", } ] }, timeout=30, ) print(response.artifact_text)如果一切正常,你会看到输出变成带编号的列表:
1. 今天完成了协议调研 2. 验证了卡片解析 3. 整理了示例代码这一步跑通,说明协议层面的握手、任务创建、任务状态轮询、结果返回这四个环节全都正常。从工程视角看,这比“两个 Agent 能互相聊天”重要得多,因为握手和状态机才是跨框架集成的骨架。
我把这套流程沉淀成了一个脚本,每次进行新的跨框架接入时,先跑一遍这个最小闭环,再进入复杂业务逻辑。几乎每次都能快速暴露协议版本、路径、字段名这类低级问题。
4.3 把非 A2A 协议的旧接口包装成 A2A 服务
更多时候,我们手头已经有一个老 Agent 服务,它是普通 HTTP 接口,不是 A2A。我不会立刻去改造老服务,而是写一层适配器,把老接口包装成 A2A 暴露出来。
适配器要做的事情本质上有三步:
- 对外暴露
/.well-known/agent.json,让别人知道可以按 A2A 协议来调。 - 接收
POST /api/tasks请求,把 A2A 格式的任务请求转换成老接口自己的内部参数。 - 将老接口的返回结果包装成 A2A 的
artifacts格式。
这样的好处是:老服务完全不需要动,只加一层薄薄的适配器。我在模拟项目里就是这样把一个只支持“一问一答”的旧接口接进来的,整个适配过程只花了一个晚上。改完以后,其他框架的 Agent 也能通过标准协议来调用这个老服务,集成成本从原来的“读文档、写对接代码”降为“拿 Agent Card 配置一下”。
5. 异构生态集成落地:多框架 Agent 编排方案与取舍
打通单个 Agent 只是第一步。真正有挑战的是把多个不同框架开发的 Agent 编到同一条业务链路上。这一节是实操项目中的编排经验总结。
5.1 不要让“异构”变成“异形”
多框架集成最怕的是每个 Agent 都有一套自己的任务表达方式。A2A 协议解决了“对话”的标准化,但没有替你解决“流程”的标准化。所以我的经验是:在上层引入一个统一的编排层,让业务代码只跟编排层对话,不直接跟底层 Agent 耦合。
在 AgentScope 项目里,这个编排层可以表现为一个 Agent 池配置:
pipeline_config = { "steps": [ {"name": "classifier", "endpoint": "http://agent-a:8001", "type": "a2a"}, {"name": "retriever", "endpoint": "http://agent-b:8002", "type": "a2a"}, {"name": "generator", "endpoint": "http://agent-c:8003", "type": "a2a"}, ] }每个 step 都是一个通过 A2A 协议暴露的 Agent,内部实现框架可以完全不同。编排层只关心地址和协议,不关心模型和框架。
5.2 编排一个真实的多 Agent 联动场景
我做过一个模拟的智能客服场景,由三个 Agent 协作:
| 步骤 | 职能 | 承载框架 |
|---|---|---|
| 分类 Agent | 识别用户意图,输出意图标签 | 某对话编排框架 |
| 检索 Agent | 根据意图查询知识库,返回候选答案 | 某内存型 Agent 实现 |
| 生成 Agent | 把候选答案整理成最终回复 | AgentScope |
整个调用链路是串行的:先分类,再检索,最后生成。我用编排代码实现:
intent = classifier.sync_send_task(payload_build(user_query)) candidates = retriever.sync_send_task(payload_build(intent)) reply = generator.sync_send_task(payload_build(candidates))这段代码的语义非常直观。每个 Agent 返回的内容都会成为下一个 Agent 的输入。看起来简单,但基于 A2A 协议,每个sync_send_task调用本质上都是:发现 Agent Card、创建任务、轮询状态、拉取结果。协议把这些复杂性全部标准化了。
实际运行中,我发现一个很现实的性能问题:A2A 的轮询模式在串行链条上会因为网络往返产生额外延迟。尤其是当每个 Agent 处理时间都在几秒时,轮询间隔会累积放大。
为了权衡,我做了两个优化:
- 客户端轮询间隔使用指数退避:前两次间隔 1 秒,后续变成 2 秒、4 秒,减少无效请求。
- 对于单次处理时间特别长的 Agent,改用“任务回调通知”模式。回调意味着服务端处理完成后主动通知客户端,避免客户端空转等待。
A2A 协议的任务模型天然支持这两种方式,AgentScope 侧可以配置偏好。这一点在我的项目里是非常重要的取舍:不是所有场景都适合标准轮询,一定要根据 Agent 的平均响应时间动态调整。
5.3 事务与补偿的取舍
跨框架与跨服务一样,都没有“全局事务”这种好东西。多个 Agent 组成一条链路时,如果中间某个 Agent 失败,前面 Agent 已经产生的副作用就不会自动回滚。
我处理的基本原则是:业务层自己做补偿,协议层只保证状态可查。也就是说,每个 Agent 暴露的能力尽量设计成“幂等的”,这样重试才安全。比如检索 Agent 只做只读查询,生成 Agent 不直接写库,而是把结果返回给编排层,由编排层统一落库。这样即使中途失败,也不会产生脏数据。
try: result2 = retriever.sync_send_task(payload_build(result1)) except Exception: # 重试前先确认上游任务是否已完成 task_info = retriever.get_task(task_id) if task_info.status == "completed": result2 = task_info.result else: raise这个模式的核心是“重试时先查询而不是直接重造”。结合 A2A 的任务状态,你可以安全地判断任务是否真的失败了,而不是因为网络抖动误判。
6. A2A 协议实战中的常见问题与排查技巧实录
最后这部分,我把在项目里实际遇到的高频问题整理成了一份排查速查表。每一条都是我亲手踩过坑之后总结出来的。
| 现象 | 常见原因 | 处理办法 |
|---|---|---|
| 客户端拉不到 Agent Card | .well-known路径写错,或服务端未设置正确的 Content-Type | 先用浏览器直接访问.well-known/agent.json验证 |
| 创建任务后状态一直 working | 服务端业务逻辑异常,没有更新任务状态 | 给客户端加超时保护,同时在服务端做异常兜底 |
| 中文内容乱码 | JSON 响应未使用ensure_ascii=False或 Content-Type 缺少 charset | 统一 UTF-8 编码,JSON 序列化关闭 ASCII 转义 |
| 跨框架工具调用失败 | 直接把内部 function_call 数据结构透传给不同框架 | 在适配层把工具调用转为文本/结构化 artifacts |
| 轮询频繁导致服务端压力大 | 客户端固定间隔轮询 | 改用指数退避或回调模式 |
| 任务幂等性差导致重试重复执行 | 没有用任务 ID 做去重 | 服务端保存已处理的 task_id,重试时直接返回原结果 |
下面详细说两类最值得展开的问题。
6.1 Agent Card 获取失败往往不是“协议”问题
第一次对接时,我花了很长时间查 A2A 服务端代码,反复确认协议字段,但客户端就是报“Agent 卡片获取失败”。后来才发现,问题出在 HTTP 路径上:A2A 规范要求卡片放在/.well-known/agent.json,我的服务端代码写的是/agent.json,路径差了一个层级。
这个类型的错误很隐蔽,因为单独访问/agent.json也能返回 JSON,看起来像“协议生效了”,但客户端按照标准路径去请求时,却拿到 404。所以排查时优先确认标准路径是否能直接访问,远比反复看协议文档快。
另外,如果客户端和服务端不在同一个域,服务端必须正确处理跨域请求。很多“握手失败”其实是浏览器同源策略导致的,跟 A2A 协议本身无关。我的经验是:服务端在返回卡片时,主动加上宽松的 CORS 头,会让调试轻松很多。
6.2 工具调用跨框架失效的根源
A2A 协议兴起之前,各个框架对工具调用的表达方式差异巨大。有的框架把工具调用表达成特殊数组结构,有的则是文本约定。直接把这些内部结构塞进 A2A 消息里透传给另一个框架,大概率会失效。
我的解决方案是:在适配层进行“降级表达”。把工具调用统一转成文本块或结构化 artifact。比如检索 Agent 不用显示声明“我需要调用 search_tool”,而是直接在回复里说“已查询知识库,获得候选集”。对端 Agent 不需要理解工具语义,只需要消费结果文本。
这个办法牺牲了一些底层精确性,但跨框架协作的稳定性提升非常明显。不要让不同框架的 Agent 去理解对方专有数据结构,让它们只通过协议里通用的text和artifacts字段交换结果就好。
最后说一个我个人的体会。A2A 协议能解决“怎么连通”的问题,但解决不了“要不要连通”和“连通后怎么保证数据一致”的问题。在接任何一个异构 Agent 之前,一定要先把技能边界、消息约束、幂等策略白纸黑字写清楚。协议不是万能胶,它只是一套双方都认的对话格式。只要这一步做踏实了,后面无论接多少框架,本质上都是重复劳动。希望这篇实战记录能让你在跨框架集成时少踩几个坑,也建议你动手第一步就先做一个最小闭环测试,把状态机和消息格式跑通,再往里面填业务逻辑。