1. 从单体 Agent 到 AgentMesh:为什么需要统一 Key 与 A2A 网络
如果你已经用 Claude Code、Cursor 或 OpenCode 跑通过单个 Agent 的任务,接下来大概率会遇到一个瓶颈:多个 Agent 之间怎么互相调用。比如一个负责查数据库的 Agent 需要把结果交给负责生成报告的 Agent,再交给负责发送通知的 Agent——这三个 Agent 可能跑在不同机器、不同框架、甚至不同团队维护的环境里。
AgentMesh 要解决的就是这个问题。它不是一个具体的开源项目,而是一种多 Agent 协作的架构模式:每个 Agent 通过 A2A(Agent-to-Agent)协议暴露自己的能力,由一个统一的入口管理身份、路由和调用凭证。你可以把它理解成微服务架构里的服务网格(Service Mesh),只不过网格里跑的不是 HTTP 微服务,而是具备自主决策能力的 Agent。
在实际搭建 AgentMesh 原型时,最先卡住大多数人的不是协议本身,而是模型调用的凭证管理。每个 Agent 都要调模型,如果每个 Agent 各自维护一套 API Key,很快就会变成:Key 散落在十几个配置文件里、额度无法统一监控、换模型时要改 N 个地方。所以这篇教程的切入点是:用 TaoToken 的统一 Key 作为所有 Agent 的模型调用入口,在此基础上搭出 AgentMesh 的配置骨架。
适合谁看:已经跑通过至少一个 Agent 工具、想进一步做多 Agent 协作的开发者;或者正在设计 Agent 工厂平台、需要一套可复制的配置模板的团队。下面给出的settings.json和config.toml都是可以直接复制修改的骨架,跑通最小原型大概需要 20 分钟。
2. TaoToken 前置:统一 Key 与 API 通道准备
TaoToken 在这里扮演的角色是模型调用的统一网关。所有 Agent 不管用什么框架、跑在哪里,模型请求都发往同一个 API 地址,用同一个 Key 认证。这样 AgentMesh 里新增一个 Agent 时,不需要再申请新的模型凭证,只需要在配置里引用已有的 Key。
先拿到 Key。访问控制台创建 API Key:
# 控制台地址(创建和管理 API Key) https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite创建后你会得到一个形如sk-xxxxxxxx的 Key。把它存到环境变量里,不要硬编码进配置文件:
# Linux / macOS export TAOTOKEN_API_KEY="sk-你的实际Key" # Windows PowerShell $env:TAOTOKEN_API_KEY="sk-你的实际Key"API 的基础地址是:
https://taotoken.net/api这个地址在后面的settings.json和config.toml里都会用到。注意 API 地址不带任何查询参数,Key 通过请求头传递。
提示:如果你打算让多个 Agent 共享同一个 Key,建议在控制台里给这个 Key 起一个明确的名字,比如
agentmesh-shared,方便后续在用量面板里区分是哪个项目在消耗额度。
模型对话的调试入口在这里,配置完成后可以先用它验证 Key 是否可用:
https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite3. 可复制配置:settings.json 与 config.toml 骨架
AgentMesh 的配置分两层:一层是全局层,定义模型通道、Key 引用、A2A 网络的基础参数;另一层是Agent 层,每个 Agent 声明自己的角色、技能、以及它在 A2A 网络里的端点信息。
3.1 settings.json:全局通道与 A2A 网络参数
这个文件放在项目根目录,所有 Agent 共享。核心是把模型调用统一指向 TaoToken,同时定义 A2A 网络的注册与发现方式。
{ "mesh": { "name": "agentmesh-local", "version": "0.1.0", "a2a": { "registry": "http://127.0.0.1:8080/registry", "heartbeat_interval_ms": 10000, "discovery_mode": "registry", "task_timeout_ms": 60000 } }, "model_gateway": { "provider": "taotoken", "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "default_model": "claude-sonnet-4-20250514", "fallback_model": "gpt-4o-mini", "timeout_ms": 30000, "max_retries": 3 }, "agents": [ { "id": "agent-router", "name": "路由 Agent", "role": "router", "endpoint": "http://127.0.0.1:9101/a2a", "skills": ["task-routing", "intent-classification"], "model_override": null }, { "id": "agent-data", "name": "数据查询 Agent", "role": "worker", "endpoint": "http://127.0.0.1:9102/a2a", "skills": ["sql-query", "data-format"], "model_override": "claude-sonnet-4-20250514" }, { "id": "agent-report", "name": "报告生成 Agent", "role": "worker", "endpoint": "http://127.0.0.1:9103/a2a", "skills": ["report-writing", "markdown-format"], "model_override": null } ] }几个关键字段说明:
model_gateway.base_url固定为 TaoToken 的 API 地址,api_key_env指向环境变量名而不是 Key 本身,这样配置文件可以安全地提交到版本库。agents数组里每个 Agent 的endpoint就是它在 A2A 网络里的调用地址,skills是它对外声明能处理的任务类型——路由 Agent 靠这个来做任务分发。
3.2 config.toml:单个 Agent 的运行时配置
每个 Agent 有自己的config.toml,放在各自的目录下。以数据查询 Agent 为例:
[agent] id = "agent-data" name = "数据查询 Agent" role = "worker" version = "0.1.0" [agent.a2a] listen_addr = "127.0.0.1:9102" registry_url = "http://127.0.0.1:8080/registry" heartbeat_interval_ms = 10000 max_concurrent_tasks = 4 [agent.model] provider = "taotoken" base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" model = "claude-sonnet-4-20250514" temperature = 0.2 max_tokens = 4096 [agent.skills] sql-query = { description = "执行 SQL 查询并返回结构化结果", input_schema = "schemas/sql-query.json" }>curl -X POST https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "claude-sonnet-4-20250514", "messages": [{"role": "user", "content": "回复 OK 两个字母"}], "max_tokens": 10 }'如果返回的 JSON 里choices[0].message.content包含OK,说明通道正常。这一步不通的话,后面所有 Agent 都跑不起来,先排查 Key 和网络。
4.2 启动本地注册中心
原型阶段用一个极简的注册中心即可。可以用 Python 起一个:
# registry/server.py from http.server import HTTPServer, BaseHTTPRequestHandler import json AGENTS = {} class RegistryHandler(BaseHTTPRequestHandler): def do_POST(self): if self.path == "/registry": length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length)) AGENTS[body["id"]] = body self.send_response(200) self.end_headers() self.wfile.write(json.dumps({"status": "registered"}).encode()) else: self.send_response(404) self.end_headers() def do_GET(self): if self.path == "/registry": self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(list(AGENTS.values())).encode()) else: self.send_response(404) self.end_headers() if __name__ == "__main__": HTTPServer(("127.0.0.1", 8080), RegistryHandler).serve_forever()启动:
python registry/server.py4.3 注册 Agent 并验证发现
用 curl 模拟一个 Agent 向注册中心注册自己:
curl -X POST http://127.0.0.1:8080/registry \ -H "Content-Type: application/json" \ -d '{ "id": "agent-data", "name": "数据查询 Agent", "endpoint": "http://127.0.0.1:9102/a2a", "skills": ["sql-query", "data-format"] }'然后查询注册中心,确认 Agent 已被发现:
curl http://127.0.0.1:8080/registry返回的列表里应该包含agent-data。这一步验证的是 A2A 网络的发现机制——路由 Agent 就是通过这个接口知道有哪些下游 Agent 可用的。
4.4 模拟一次 A2A 调用
路由 Agent 收到任务后,根据技能匹配找到agent-data,然后向它的端点发起调用:
curl -X POST http://127.0.0.1:9102/a2a \ -H "Content-Type: application/json" \ -H "X-A2A-From: agent-router" \ -d '{ "task_id": "task-001", "skill": "sql-query", "input": {"query": "SELECT count(*) FROM orders WHERE status = '\''pending'\''"}, "callback": "http://127.0.0.1:9101/a2a/callback" }'如果agent-data的运行时正常,它会执行查询、调用 TaoToken 的模型通道做结果格式化,然后通过callback地址把结果回传给路由 Agent。整个链路跑通,最小 AgentMesh 原型就成立了。
5. 本篇常见错排查
5.1 模型请求返回 401
最常见的原因是环境变量没生效。settings.json和config.toml里写的是api_key_env,运行时需要确保TAOTOKEN_API_KEY在当前 shell 会话里确实存在。用echo $TAOTOKEN_API_KEY确认一下,如果为空,重新 export 一次。另一个可能是 Key 被复制时带了空格或换行,检查一下。
5.2 Agent 注册后查不到
先确认注册中心的POST /registry返回了 200。如果返回 404,检查settings.json里a2a.registry的路径是否和注册中心实际监听的路径一致。另外,注册中心重启后内存里的AGENTS会清空,原型阶段这是正常的,重新注册即可。
5.3 A2A 调用超时
task_timeout_ms默认 60 秒。如果下游 Agent 处理的任务涉及多次模型调用,可能超时。两个调整方向:一是把task_timeout_ms调大,二是检查下游 Agent 的max_concurrent_tasks是否被占满。本地原型阶段,把max_concurrent_tasks设为 1 反而更容易排查问题,因为任务会排队而不是并发。
5.4 模型返回内容被截断
检查config.toml里的max_tokens。不同模型的默认上限不同,如果任务需要生成较长的报告,把max_tokens调到 8192 或更高。同时确认 TaoToken 通道的timeout_ms足够长,长文本生成容易触发超时。
5.5 多个 Agent 共用 Key 时额度混乱
这是统一 Key 方案的固有代价。解决办法是在 TaoToken 控制台里给不同项目创建不同的 Key,然后在settings.json里通过api_key_env区分。比如 AgentMesh 用TAOTOKEN_API_KEY_MESH,其他项目用各自的变量名。这样用量面板里能按 Key 拆分统计。
6. 下一步:从原型到可用的 AgentMesh
跑通上面的骨架后,你手里已经有一套可工作的多 Agent 协作配置。接下来可以往三个方向扩展。
第一,把注册中心换成带健康检查的版本。原型里的注册中心只存不查,生产环境需要心跳检测和故障摘除。可以在settings.json的a2a.heartbeat_interval_ms基础上,让每个 Agent 定期向注册中心发心跳,超时未心跳的从可用列表里移除。
第二,给 A2A 调用加上任务级凭证。目前X-A2A-From只是一个标识,没有验证。可以在 TaoToken 的 Key 体系之上,为每次 A2A 调用生成一个短期 token,下游 Agent 验证 token 后才执行任务。这样即使注册中心被伪造,也无法调用下游 Agent。
第三,把配置骨架接入实际的 Agent 运行时。settings.json和config.toml定义的是声明式配置,需要有一个加载器把它们读进内存,然后启动对应的 Agent 进程。如果你用的是 Claude Code 或 OpenCode 这类工具,可以把config.toml里的agent.model段映射到它们的模型配置里,让它们也走 TaoToken 的统一通道。
长期做编码类 Agent 和 Agent 编排的话,Coding Plan 里有更完整的模型路由和额度管理方案,适合把原型推进到团队可用:
https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite接入文档里有 A2A 端点鉴权和任务回调的详细说明,扩展注册中心之前建议先过一遍:
https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite我试过把三个 Agent 跑在同一台机器上,注册中心用内存版,整个原型从零到跑通大概花了 25 分钟,其中大部分时间花在调config.toml的缩进和确认环境变量上。真正跑通之后,新增一个 Agent 只需要复制一份config.toml、改id和listen_addr、在settings.json的agents数组里加一项,然后重启注册中心。这个复制成本足够低,才值得继续往上叠功能。