1. 从单 Agent 到多 AgentA2A 协议到底解决什么问题如果你已经用 LangGraph、CrewAI 或者 ADK 写过单个 Agent大概率会遇到一个天花板一个 Agent 再强也很难同时把「查资料、写代码、跑测试、发通知」全干完。你可能会想那我起两个 Agent 不就行了问题恰恰出在这里——两个用不同框架写的 Agent怎么互相说话A2AAgent2Agent协议就是冲着这件事来的。它是一个开放标准定义了一套通用的通信语言让不同厂商、不同框架构建的 Agent 能够互相发现能力、委派任务、交换结果。你可以把它理解成 Agent 世界的 HTTP不管对面是 Python 写的还是 Java 写的只要遵循 A2A就能对话。它和 MCP 的关系经常被搞混。简单说MCP 解决的是「Agent 怎么用工具」A2A 解决的是「Agent 怎么找 Agent 帮忙」。一个对内连接能力一个对外连接同伴。两者是互补的不是替代关系。这篇文章面向想真正跑通多智能体协作的开发者。我会带你从环境准备开始写一个 A2A 服务端 Agent再写一个客户端 Agent最后用一个「任务分派」的双 Agent 案例把整条链路验证一遍。全程可复制踩过的坑我也会标出来。适合谁看已经会写基础 LLM 调用、想往多 Agent 架构走的人或者手上有一堆零散 Agent想用统一协议把它们串起来的人。不需要你提前懂 A2A但需要你能跑 Python 和看懂 JSON。2. 前置准备A2A SDK 安装与 TaoToken 接入配置A2A 本身是协议标准不是一个 pip 装完就能用的软件。真正落地要靠 SDK 和框架实现。目前官方提供了 Python、JavaScript、Java 的 SDKPython 生态最成熟我们以 Python 为主线。先说模型接入这一层。多 Agent 协作会频繁调用模型如果每个 Agent 都单独配一套 Key管理起来很痛苦。我的做法是统一走一个兼容 OpenAI 协议的入口把 Base URL 和 Key 集中管理。这里用 TaoToken 作为模型接入层它的 API 地址是https://taotoken.net/api兼容 OpenAI 的调用格式Agent 里改一下 base_url 就能用。先建虚拟环境把依赖装好python -m venv a2a-demo source a2a-demo/bin/activate # Windows 用 a2a-demo\Scripts\activate pip install a2a-sdk openai uvicorn fastapi httpx如果你用的是官方 samples 仓库可以直接克隆下来对照git clone https://github.com/google/A2A.git cd A2A pip install -r requirements.txt接下来配置模型凭证。我习惯用环境变量避免 Key 写死在代码里export TAOTOKEN_API_KEY你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/apiKey 在控制台的 API Keys 页面生成地址是https://taotoken.net/api-keys。生成后复制一次页面刷新就看不到了记得存好。注意Base URL 结尾不要带/v1SDK 内部会自己拼。带了会出现路径重复报 404。模型 ID 这块多 Agent 场景建议选一个指令跟随稳定的模型比如gpt-4o-mini或者claude-3-5-sonnet这类。具体可用列表在模型对话页面能查到地址https://taotoken.net/models。选模型的原则是Agent 之间传的是结构化任务描述模型要能稳定输出 JSON别选那种爱自由发挥的。到这里前置就齐了Python 环境、A2A SDK、模型接入三件套Base URL Key Model ID。下一节开始写真正的服务端配置。3. 可复制配置A2A 服务端 Agent 的 AgentCard 与执行器A2A 服务端的核心是两样东西AgentCard 和 AgentExecutor。AgentCard 是 Agent 的「名片」告诉别人我是谁、我能干什么、怎么调用我。AgentExecutor 是「大脑」真正处理收到的任务。先写 AgentCard。它本质是一个 JSON 结构描述 Agent 的元数据。我把它单独放一个文件agent_card.json{ name: research_agent, description: 负责信息检索与摘要的 Agent, url: http://localhost:8001, version: 1.0.0, capabilities: { streaming: true, pushNotifications: false }, defaultInputModes: [text/plain], defaultOutputModes: [text/plain], skills: [ { id: search_and_summarize, name: 检索并摘要, description: 接收一个主题返回结构化摘要, inputModes: [text/plain], outputModes: [text/plain] } ] }几个字段容易踩坑。url必须和实际监听地址一致客户端靠它来发请求。skills里的id是任务分派时的匹配依据命名要语义化。capabilities.streaming如果设成 true服务端就得实现流式返回否则客户端会等不到数据。然后是执行器。它继承 SDK 的AgentExecutor实现execute方法import os from openai import OpenAI from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.events import EventQueue from a2a.utils import new_agent_text_message client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) class ResearchExecutor(AgentExecutor): async def execute(self, context: RequestContext, event_queue: EventQueue): task_text context.get_user_input() resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是检索摘要助手输出简洁要点。}, {role: user, content: task_text}, ], ) answer resp.choices[0].message.content await event_queue.enqueue_event(new_agent_text_message(answer))这里context.get_user_input()拿到的是客户端发来的任务文本。event_queue是 A2A 的事件通道你把结果塞进去SDK 负责按协议格式回给客户端。注意execute是 async 的别写成同步函数否则事件队列不会 flush。最后把它们组装成服务import uvicorn from a2a.server.apps import A2AStarletteApplication from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.tasks import InMemoryTaskStore from a2a.types import AgentCard card AgentCard.model_validate_json(open(agent_card.json).read()) handler DefaultRequestHandler( agent_executorResearchExecutor(), task_storeInMemoryTaskStore(), ) app A2AStarletteApplication(agent_cardcard, http_handlerhandler).build() if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8001)InMemoryTaskStore适合本地调试生产环境要换成持久化实现否则重启后任务状态全丢。跑起来后访问http://localhost:8001/.well-known/agent.json应该能看到你的 AgentCard这就是 A2A 的「发现」机制——客户端靠这个固定路径找到你的名片。4. 验证请求双 Agent 任务分派案例跑通全流程服务端有了现在写客户端再起第二个 Agent做一个「主 Agent 分派任务给研究 Agent」的案例。先起第二个 Agent叫writer_agent监听 8002职责是把摘要写成一段通顺的说明文。它的 AgentCard 和 executor 跟上面结构一样只改 name、url、skill 和 system prompt。为了省事我把 executor 抽成通用类用参数区分角色class RoleExecutor(AgentExecutor): def __init__(self, system_prompt: str): self.system_prompt system_prompt async def execute(self, context, event_queue): task_text context.get_user_input() resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: self.system_prompt}, {role: user, content: task_text}, ], ) await event_queue.enqueue_event( new_agent_text_message(resp.choices[0].message.content) )研究 Agent 用「输出要点」写作 Agent 用「把要点扩写成段落」。两个服务分别用 8001、8002 端口启动。客户端这边A2A SDK 提供了A2AClient但更直观的方式是直接用 httpx 按协议发请求方便你看清每一步。核心是构造一个message/send请求import httpx, uuid, asyncio async def send_task(agent_url: str, text: str): payload { jsonrpc: 2.0, id: str(uuid.uuid4()), method: message/send, params: { message: { role: user, parts: [{kind: text, text: text}], messageId: str(uuid.uuid4()), } }, } async with httpx.AsyncClient(timeout60) as c: r await c.post(agent_url, jsonpayload) r.raise_for_status() return r.json() async def main(): topic A2A 协议在多智能体协作中的作用 # 第一步研究 Agent 出摘要 research await send_task(http://localhost:8001, topic) summary research[result][parts][0][text] print(研究 Agent 返回, summary) # 第二步把摘要交给写作 Agent writer await send_task(http://localhost:8002, f请扩写{summary}) print(写作 Agent 返回, writer[result][parts][0][text]) asyncio.run(main())跑起来后你应该看到两段输出第一段是研究 Agent 的要点第二段是写作 Agent 基于要点扩写的段落。这就是一次完整的 A2A 任务分派——主流程先调 8001拿到结果再调 8002两个 Agent 之间没有共享内存只通过协议传文本。如果你想验证「发现」机制可以在客户端先 GEThttp://localhost:8001/.well-known/agent.json解析出 skills再决定把任务发给谁。这样客户端就不需要硬编码 Agent 地址扩展性更好。实测下来整个链路最耗时的不是模型调用而是服务启动时的端口占用检查。两个 Agent 一定要用不同端口否则第二个起不来客户端会连到错误的 Agent 上返回的结果驴唇不对马嘴。5. 常见报错排查401、local proxy failed 与 reading choices多 Agent 跑不通八成是下面几个错。我按出现频率排一下。401 Unauthorized。这个基本是 Key 或 Base URL 的问题。先确认环境变量有没有真正加载echo $TAOTOKEN_API_KEY echo $TAOTOKEN_BASE_URL如果输出为空说明 export 没生效或者你在新的终端窗口里跑代码。另一个常见原因是 Base URL 写成了https://taotoken.net/api/v1SDK 再拼一次/v1就变成/api/v1/v1服务端直接拒。正确写法是https://taotoken.net/api。local proxy failed / connection refused。这个错通常出现在客户端连服务端的时候。检查三件事服务端是否真的在跑curl http://localhost:8001/.well-known/agent.json能不能返回 JSON端口是否被占用lsof -i :8001AgentCard 里的url和实际监听地址是否一致。我遇到过一次AgentCard 写的是127.0.0.1服务监听的是0.0.0.0本地能通但容器里连不上改成一致就好了。reading choices of undefined。这个错来自模型返回体解析。resp.choices[0]报 undefined说明返回结构里没有 choices。原因可能是模型 ID 写错了服务端返回了错误对象或者请求被限流返回了空。加一行防御性打印print(resp.model_dump())看实际返回什么。如果是{error: ...}那就是模型侧的问题检查 Model ID 是否在可用列表里。多 Agent 场景下每个 Agent 的模型 ID 要单独确认别复制粘贴漏改。OAuth / token 过期。如果你用的是需要 OAuth 的模型服务token 有有效期。A2A 服务端是长驻进程token 过期后所有请求都会 401。解决办法是在 executor 里做 token 刷新或者用不会过期的 API Key。TaoToken 的 Key 是长期有效的省了这块麻烦。任务卡住不返回。如果客户端一直等服务端日志也没输出大概率是execute方法写成了同步函数事件队列没被 await。检查方法签名有没有asyncenqueue_event前面有没有await。排查顺序建议先 curl 服务端发现接口再单独测模型调用最后测完整链路。分层定位比一上来就 debug 全流程快得多。6. 把 A2A 用起来从本地 Demo 到长期协作的落地建议跑通双 Agent 只是起点。真正要落地有几个地方值得提前想清楚。第一是 Agent 的粒度。别把 Agent 切得太碎否则一次任务要串五六个 Agent延迟叠加起来很难受。我的经验是一个 Agent 对应一个明确的职责边界比如「检索」「写作」「审核」而不是「处理第一步」「处理第二步」。职责边界稳定的 Agent 才能被复用。第二是任务状态管理。本地用InMemoryTaskStore没问题但多 Agent 协作往往涉及长任务中间可能断线。生产环境要换成持久化存储并且利用 A2A 的 push notification 机制让服务端在任务完成时主动通知客户端而不是客户端一直轮询。第三是模型接入的集中管理。Agent 数量一多每个都配 Key 会失控。统一走一个兼容 OpenAI 协议的入口改 base_url 就能切换模型维护成本低很多。TaoToken 的接入文档在https://taotoken.net/doc里面有各语言的调用示例照着改就行。如果你打算把多 Agent 协作做成长期跑的服务比如每天定时分派任务、多个 Agent 协同处理那用按量计费的方式会比单独买模型额度更灵活。Coding Plan 那套适合长期编码和 Agent 场景地址是https://taotoken.net/coding-plan可以先看看额度模型再决定。最后给一个实用技巧在 AgentCard 的 description 里写清楚「什么任务该发给我」客户端做任务路由时可以先用模型读一遍所有 Agent 的 description再决定分派给谁。这比硬编码 if-else 灵活得多也是 A2A「发现」机制真正的价值所在。