1. 从两个 Agent 互相“装死”说起:A2A 协作链路到底卡在哪
如果你已经用谷歌 Agent2Agent 协议(A2A)搭过两个智能体,大概率遇到过这种场面:Agent A 明明发出了tasks/send,Agent B 的日志里却什么都没有;或者 B 回了结果,A 那边一直卡在submitted状态不动。我试过最离谱的一次,两个 Agent 各自跑得好好的,一连起来就互相“装死”,排查半天发现是两边的模型通道 Key 不统一,一个走的是 A 供应商,一个走的是 B 供应商,返回结构对不上,解析直接抛异常。
这就是 A2A 落地时最真实的痛点。协议本身解决的是“智能体之间怎么说话”的问题,它定义了 Agent Card、Task、Message、Artifact 这些概念,让不同框架的 Agent 能互相发现、协商、回传结果。但协议不解决“每个 Agent 背后调用的模型从哪来、Key 怎么管、Base URL 怎么配”的问题。当你有三个、五个甚至更多 Agent 时,每个 Agent 都去单独申请一套模型凭证,维护成本会迅速失控,而且一旦某个供应商的接口格式有差异,协作链路就会在某个环节断掉。
所以这篇内容聚焦一个具体场景:用 TaoToken 作为统一的模型调用入口,给多个 A2A Agent 提供一致的 Base URL 和 Key,让任务分发和结果回传这条链路真正跑通。适合谁看?已经在写 Agent 代码、想让两个以上智能体协作、但被多套凭证和多套接口格式折腾过的开发者。你不需要先精通 A2A 的全部规范,只要能把两个 HTTP 服务跑起来,就能跟着下面的步骤验证链路。
核心检索词先明确:Agent2Agent 协议是谷歌推出的开源智能体互操作协议,TaoToken 是统一模型 API 通道,两者结合解决的是“多 Agent 协作时模型调用层不统一”的问题。下面从环境准备开始,一步步给出可复制的配置和验证方法。
2. 前置准备:TaoToken 统一 Key 与 A2A Agent 环境搭建
在动手写 Agent 互调代码之前,先把两件事准备好:一个是 TaoToken 的 API Key 和 Base URL,另一个是本地能跑起来的 A2A Agent 骨架。这两件事都不复杂,但顺序不能乱,否则后面验证时会分不清是 Key 的问题还是 Agent 代码的问题。
先说 TaoToken 这边。你需要拿到一个 API Key,这个 Key 会同时给多个 Agent 使用,所以不要把它硬编码在某个 Agent 的源码里,而是通过环境变量注入。Base URL 统一用https://taotoken.net/api,注意这个地址不带任何查询参数,直接作为 OpenAI 兼容接口的根路径使用。Key 的获取入口在控制台的 API Keys 页面,登录后创建一个新 Key,复制出来先存到本地临时文件里,后面配置环境变量时要用。
这里有个容易踩的坑:很多人会把 Base URL 写成带/v1的完整路径,然后在代码里又拼一次/v1,结果请求发到了/v1/v1/chat/completions,直接 404。TaoToken 的 Base URL 就是https://taotoken.net/api,至于要不要加/v1,取决于你用的 SDK。如果你用的是 OpenAI 官方 Python SDK,它内部会自动拼/chat/completions,所以 Base URL 保持https://taotoken.net/api即可;如果你用的是裸requests发请求,那完整路径就是https://taotoken.net/api/v1/chat/completions。这一点在后面的配置片段里会再强调。
再说 A2A Agent 骨架。你不需要从零实现 A2A 协议的全部方法,先用一个最小化的 HTTP 服务模拟两个 Agent 即可。Agent A 负责接收用户任务,然后通过 A2A 的tasks/send把子任务转发给 Agent B;Agent B 处理完后,把结果作为 Artifact 回传给 Agent A。两个 Agent 背后都调用同一个 TaoToken 通道来生成内容。这样你就能在一个可控的环境里观察任务分发和结果回传的完整日志。
环境变量建议统一放在一个.env文件里,两个 Agent 启动时都加载同一个文件。这样做的目的是保证两个 Agent 用的是同一套凭证和同一个 Base URL,排除“两边配置不一致”这个最常见的干扰因素。具体变量名和值在下一节给出,你可以直接复制。
另外提醒一点:A2A 协议里的 Agent Card 通常放在/.well-known/agent.json,里面会声明 Agent 的能力和端点 URL。在本地验证阶段,你可以先手动写一个静态的 JSON 文件,不需要动态生成。等链路跑通之后,再考虑把 Agent Card 做成动态接口。这个顺序很重要,先通链路,再补规范细节。
3. 可复制配置:环境变量、Base URL 与 Agent 调用片段
这一节给出可以直接复制到项目里的配置片段。我按文件类型分开写,你照着放到对应位置即可。所有片段里的 Key 都用占位符sk-taotoken-xxxxxxxx表示,你替换成自己创建的那个 Key。
首先是.env文件,放在项目根目录,两个 Agent 共用:
# TaoToken 统一通道配置 TAOTOKEN_API_KEY=sk-taotoken-xxxxxxxx TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL_ID=claude-sonnet-4-20250514 # Agent A 配置 AGENT_A_PORT=8001 AGENT_A_CARD_URL=http://localhost:8001/.well-known/agent.json # Agent B 配置 AGENT_B_PORT=8002 AGENT_B_CARD_URL=http://localhost:8002/.well-known/agent.json注意TAOTOKEN_MODEL_ID这一项,它决定了两个 Agent 背后调用的具体模型。你可以在 TaoToken 的模型列表里选一个支持长上下文和工具调用的模型,这里用claude-sonnet-4-20250514作为示例。两个 Agent 用同一个 Model ID,是为了保证返回结构一致,减少解析层的分支判断。
接下来是 Agent B 的调用片段,用 Python 的 OpenAI SDK 写。Agent B 收到任务后,调用 TaoToken 生成内容,然后把结果封装成 A2A 的 Artifact 返回:
import os from openai import OpenAI from fastapi import FastAPI, Request from pydantic import BaseModel app = FastAPI() client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) class TaskRequest(BaseModel): task_id: str message: str @app.post("/tasks/send") async def handle_task(req: TaskRequest): completion = client.chat.completions.create( model=os.environ["TAOTOKEN_MODEL_ID"], messages=[ {"role": "system", "content": "你是一个负责子任务的智能体,请简洁回答。"}, {"role": "user", "content": req.message}, ], temperature=0.3, ) result_text = completion.choices[0].message.content return { "task_id": req.task_id, "status": "completed", "artifacts": [ {"type": "text", "content": result_text} ], }这段代码的关键点有三个。第一,base_url直接读环境变量,不写死,方便切换。第二,model也读环境变量,和 Agent A 保持一致。第三,返回结构里带了task_id和status,这是 A2A 协议里 Task 生命周期的一部分,客户端可以根据status判断任务是否完成。
然后是 Agent A 的调用片段,它负责把用户任务分发给 Agent B,并接收回传结果:
import os import httpx from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() AGENT_B_URL = f"http://localhost:{os.environ['AGENT_B_PORT']}/tasks/send" class UserTask(BaseModel): task_id: str query: str @app.post("/dispatch") async def dispatch(req: UserTask): async with httpx.AsyncClient(timeout=60) as http: resp = await http.post( AGENT_B_URL, json={"task_id": req.task_id, "message": req.query}, ) resp.raise_for_status() data = resp.json() return { "task_id": data["task_id"], "status": data["status"], "final_output": data["artifacts"][0]["content"], }Agent A 本身不直接调用模型,它只负责转发和汇总。这样做的好处是职责清晰:Agent A 是协调者,Agent B 是执行者,模型调用统一在 Agent B 这一层通过 TaoToken 完成。如果你想让 Agent A 也具备模型能力,可以再给它加一个独立的调用函数,但 Base URL 和 Key 仍然复用同一套环境变量。
最后附一个 Agent Card 的静态 JSON 示例,放在/.well-known/agent.json路径下,两个 Agent 各一份,内容按各自能力改一下:
{ "name": "agent-b-executor", "description": "负责执行子任务并返回文本结果", "url": "http://localhost:8002", "capabilities": ["text-generation", "task-execution"], "endpoints": { "tasks/send": "http://localhost:8002/tasks/send" } }这个 JSON 在本地验证阶段不是必须的,但加上之后,后面如果你想用 A2A 的发现机制去动态获取 Agent 能力,就有地方可读。先放着,不影响链路跑通。
4. 验证请求:两个 Agent 互调后的日志与成功结果判断
配置写完之后,最关键的一步是验证链路是否真的通了。很多人到这里会直接看 Agent A 的返回,看到有内容就以为成功了,但其实可能只是 Agent A 自己编了一段话,Agent B 根本没被调用。所以验证要看三个层面的日志:Agent A 的转发日志、Agent B 的接收日志、以及 TaoToken 侧的调用记录。
先启动两个服务。Agent B 先起,因为它要监听端口等请求:
uvicorn agent_b:app --port 8002 --reload然后起 Agent A:
uvicorn agent_a:app --port 8001 --reload两个服务都起来后,用 curl 发一个测试请求给 Agent A 的/dispatch端点:
curl -X POST http://localhost:8001/dispatch \ -H "Content-Type: application/json" \ -d '{"task_id": "test-001", "query": "用一句话说明什么是A2A协议"}'预期返回类似这样:
{ "task_id": "test-001", "status": "completed", "final_output": "A2A协议是谷歌推出的开源智能体互操作协议,让不同框架的AI智能体能够相互通信和协作。" }看到这个返回,说明 Agent A 成功把请求转发给了 Agent B,Agent B 通过 TaoToken 调用了模型,并把结果回传给了 Agent A。但这还不够,你要去 Agent B 的终端看日志,确认它确实收到了task_id为test-001的请求。Agent B 的日志里应该出现类似这样的记录:
INFO: 127.0.0.1:xxxxx - "POST /tasks/send HTTP/1.1" 200 OK如果 Agent B 的日志里没有这条记录,那说明 Agent A 的转发地址配错了,或者 Agent B 没启动成功。这时候先检查AGENT_B_PORT环境变量是否和 Agent B 实际监听的端口一致。
再进一步,你可以去 TaoToken 控制台的调用记录页面,看是否有一次模型调用发生在你发请求的时间点附近。如果有,说明整条链路从 Agent A 到 Agent B 再到 TaoToken 是通的。如果 TaoToken 侧没有记录,但 Agent B 返回了 200,那可能是 Agent B 用了缓存或者 mock 数据,需要检查代码里是否真的执行了client.chat.completions.create。
还有一个验证技巧:故意把 Agent B 的TAOTOKEN_API_KEY改成一个错误的值,再发一次请求。如果 Agent A 返回的是 500 错误,并且 Agent B 日志里出现 401 相关的报错,说明链路是通的,只是 Key 不对。这个反向验证能帮你快速区分“链路不通”和“凭证错误”两种情况。
成功跑通之后,你可以把task_id换成动态生成的 UUID,连续发多个请求,观察两个 Agent 是否能并发处理。如果并发时出现结果错乱,比如task_id对不上,那说明你的 Agent B 在处理时没有正确隔离每个请求的上下文,需要检查是否有全局变量被多个请求共享。
5. 本篇常见错排查:401、local proxy failed、reading choices 与 OAuth 报错
链路跑不通时,报错信息通常集中在几个固定的位置。下面按真实遇到的频率从高到低排列,每条都给出触发条件和处理方式。
401 Unauthorized是最常见的。触发条件通常是TAOTOKEN_API_KEY没有正确注入到环境变量里,或者 Key 复制时带了多余的空格。检查方法很简单,在 Agent B 的代码里加一行print(os.environ.get("TAOTOKEN_API_KEY")),看输出的值是否和你创建的一致。如果输出是None,说明.env文件没有被加载,需要确认你用的框架是否自动读取.env,如果没有,手动用python-dotenv加载。另外注意,Key 不要写在代码里然后提交到 Git,环境变量是最低要求。
local proxy failed这个报错通常出现在你本地设置了 HTTP 代理,但代理没有正确处理taotoken.net的请求。触发条件是系统环境变量里有HTTP_PROXY或HTTPS_PROXY,而代理服务没有运行或者规则不匹配。处理方式是检查你的终端环境,临时取消代理设置再试。如果你确实需要通过代理访问外网,确保代理规则里把taotoken.net加入直连或正确转发。这个报错和 TaoToken 本身无关,是本地网络环境的问题。
reading choices 报错,完整信息通常是KeyError: 'choices'或者TypeError: 'NoneType' object is not subscriptable。触发条件是模型返回的 JSON 结构里没有choices字段,而你的代码直接取了completion.choices[0]。这种情况多半是因为请求发到了错误的端点,比如 Base URL 拼错导致返回了一个 HTML 错误页,解析 JSON 时失败。检查你的 Base URL 是否是https://taotoken.net/api,以及 SDK 是否自动拼接了正确的路径。如果你用的是裸requests,确认完整 URL 是https://taotoken.net/api/v1/chat/completions。
OAuth 相关报错,比如invalid_grant或unauthorized_client,通常出现在你尝试用 OAuth 方式获取 Token 而不是直接用 API Key 的场景。TaoToken 的 API 通道用的是 Key 认证,不需要走 OAuth 流程。如果你在代码里引入了 OAuth 库或者配置了client_id、client_secret,先去掉这些,改用api_key参数。这个报错和 A2A 协议本身无关,是认证方式选错了。
还有一个不报错但链路不通的情况:Agent A 返回了结果,但内容是空的字符串。触发条件是 Agent B 的temperature设得过高,或者max_tokens设得太小,导致模型返回了空内容。检查 Agent B 的调用参数,把temperature降到 0.3 左右,max_tokens至少设成 256。另外,如果你在 system message 里写了“请简洁回答”,模型可能会返回一个空字符串,改成“请用一句话回答”更稳妥。
最后提醒一个配置层面的坑:如果你同时用了 CC Switch 或 Cline MCP 这类工具来管理多个模型通道,确保它们指向的 Base URL、Key、Model ID 三件套和你在.env里写的一致。三件套里任何一个不一致,都会导致 Agent 之间的返回结构出现差异,进而让 A2A 的结果回传解析失败。排查时先把三件套对齐,再去看协议层的日志。
6. 把统一通道用起来:从两个 Agent 到可持续的协作链路
链路跑通之后,你可以做几件让这套东西真正可用的事。第一件是把 Agent Card 从静态 JSON 改成动态接口,让 Agent A 在分发任务前先拉取 Agent B 的卡片,确认对方具备处理该任务的能力。这样当你有多个执行型 Agent 时,协调者可以根据能力标签做路由,而不是硬编码一个 URL。
第二件是把task_id的生成和追踪做成一个轻量的任务表。每次 Agent A 分发任务时,记录task_id、目标 Agent、发起时间;Agent B 回传后,更新状态和结果。这样当某个任务卡住时,你能快速定位是哪个环节没有返回。这个任务表不需要数据库,用一个内存字典或者本地 JSON 文件就够,验证阶段够用。
第三件是给 TaoToken 的调用加上重试和超时。Agent B 在调用模型时,如果遇到网络抖动,一次失败不应该让整个 A2A 任务失败。用tenacity或者手写一个简单的重试循环,设置最多重试 2 次,每次间隔 1 秒。超时时间设成 60 秒,因为有些模型在长上下文下响应会慢一些。这些参数在.env里也可以配,方便调整。
如果你打算把这条链路用到更接近生产的场景,建议把 Agent A 和 Agent B 的日志统一收集到一个地方,按task_id串联。这样排查问题时,你能一眼看到从请求进入到模型返回的完整路径。日志里至少记录四个时间点:Agent A 收到请求、Agent A 发出转发、Agent B 收到请求、Agent B 收到模型返回。这四个时间点之间的差值,能帮你判断瓶颈在转发层还是模型层。
最后,关于模型通道的选择,TaoToken 的 API 入口是https://taotoken.net/api,Key 在控制台的 API Keys 页面创建。如果你后面要接入更多 Agent,每个 Agent 都复用同一套环境变量即可,不需要单独申请凭证。需要看接入细节的话,接入文档在https://taotoken.net/doc;想先验证模型返回是否正常,可以用模型对话页面发一条测试消息;如果打算长期跑编码类或 Agent 类任务,Coding Plan 的入口在https://taotoken.net/coding-plan。这几个入口按需取用,先把两个 Agent 的链路跑稳,再逐步扩展。