news 2026/10/2 6:46:29

【大模型篇】A2A协议实战:用TaoToken统一Key跑通Agent Card与Task协作

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
【大模型篇】A2A协议实战:用TaoToken统一Key跑通Agent Card与Task协作

1. 从单Agent塞满上下文说起:A2A协议到底解决什么问题

如果你最近在折腾多Agent协作,大概率会遇到一个很具体的困境:一个Agent既要查资料、又要写代码、还要做代码审查,跑着跑着上下文窗口就爆了。我试过把十几个工具全塞给一个Agent,结果模型在工具选择上开始犯迷糊,明明该调代码执行器,它却去查了数据库。这不是模型不行,而是单Agent的架构本身有天花板。

A2A协议(Agent-to-Agent)要解决的就是这个天花板问题。它的核心思路很朴素:既然一个Agent干不完,那就拆成多个专业Agent,每个Agent只负责一类任务,彼此之间通过标准协议通信。这里的关键词是「标准协议」——不是你自己拍脑袋定义的HTTP接口,而是一套让不同团队、不同框架做出来的Agent能互相发现、互相委托任务的约定。

具体来说,A2A协议里有两个最核心的概念你需要先建立直觉。第一个是Agent Card,你可以把它理解成Agent的「名片」或者「简历」。每个A2A Agent都会在一个约定位置发布一张JSON格式的名片,里面写清楚自己叫什么、能做哪类任务(skill列表)、支不支持流式返回、支不支持异步回调。第二个是Task,它是A2A中任务协作的基本单位。调度Agent把一段任务委托给另一个Agent,就是创建一个Task;接收方执行完,把结果作为artifacts返回。

这套机制适合谁?我认为三类人最该关注。一是正在做多Agent编排的开发者,你迟早要面对Agent之间怎么分工的问题;二是做企业级AI应用的团队,需要把不同部门、不同供应商的Agent串起来;三是想理解A2A和MCP分工边界的人——MCP解决的是单个Agent怎么连工具和数据,A2A解决的是多个Agent之间怎么分工协作,一个向下连工具,一个向上连Agent,两者互补而非替代。

但这里有个现实问题:多Agent协作意味着你要同时管理多个Agent的API通道、多个Key、多套计费。如果每个Agent都单独配一套凭证,调试成本会高得离谱。这也是我后面要引入TaoToken统一Key的原因——先用一个通道把多Agent的模型调用统一起来,再谈A2A的协作逻辑,落地会顺很多。

2. TaoToken前置准备:统一Key与API通道怎么配

在真正跑A2A协作之前,你得先解决一个基础设施问题:多个Agent背后可能调用不同的模型,如果每个Agent都单独申请Key、单独配Base URL,光是环境变量就能把你绕晕。TaoToken在这里的角色是一个统一的API通道,你申请一个Key,就能通过同一个Base URL调用多种模型,这对多Agent场景特别友好——调度Agent用推理强的模型,生码Agent用代码能力强的模型,审查Agent用长上下文模型,但底层走的是同一套凭证体系。

先明确几个地址,后面配置会反复用到。官网入口是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API的Base URL是 https://taotoken.net/api ,注意这个API地址后面不加任何UTM参数,保持干净。你需要去控制台创建Key,控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ,Key的管理页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。如果你对某个模型的能力不确定,可以先去模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 快速试一下,确认模型ID和响应格式再写进Agent配置。

拿到Key之后,我建议你先用最简方式验证通道是否通。以OpenAI兼容格式为例,你可以用curl直接打一次:

curl https://taotoken.net/api/v1/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer sk-你的Key" \ -d '{ "model": "gpt-4o-mini", "messages": [{"role": "user", "content": "只回复OK两个字"}] }'

如果返回的JSON里choices[0].message.content是「OK」,说明通道没问题。这一步很重要,因为后面A2A协作里每个Agent都要调模型,如果通道本身不通,你会在A2A的报错里绕很久。

接下来是环境变量的统一管理。我习惯把TaoToken的配置抽成一组环境变量,所有Agent共享:

export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_API_KEY="sk-你的Key" export SCHEDULER_MODEL="gpt-4o" export CODER_MODEL="claude-3-5-sonnet" export REVIEWER_MODEL="gpt-4o-mini"

这样调度Agent、生码Agent、审查Agent各自读自己的模型ID,但Base URL和Key是同一套。如果你用的是Claude Code这类工具,配置方式略有不同,需要走Anthropic兼容通道,文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 有说明。对于长期跑编码类Agent的场景,可以考虑Coding Plan,入口是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ,它更适合高频调用的Agent工作流。

这里有个坑要提前说:很多人配多Agent时喜欢给每个Agent单独写一份配置文件,结果Key一换就要改五六个地方。统一Key的价值不只是省钱,更是让凭证管理收敛到一个点。你换Key只改一处,所有Agent自动生效。

3. 可复制配置:Agent Card与Task请求的JSON样例

这一节是全文最实操的部分,我会给出可以直接复制修改的Agent Card配置、Task请求体和回调处理的JSON样例。你不需要一次全懂,先把结构跑通,再按自己的业务改字段。

先看Agent Card。按照A2A协议的约定,Agent Card发布在/.well-known/agent-card.json路径下(早期版本是agent.json,现在建议用agent-card.json)。一个生码Agent的名片大概长这样:

{ "name": "code-generator-agent", "description": "根据需求描述生成代码草稿,支持多语言", "url": "http://localhost:8001/a2a", "version": "1.0.0", "capabilities": { "streaming": true, "pushNotifications": true }, "defaultInputModes": ["text/plain"], "defaultOutputModes": ["text/plain", "application/json"], "skills": [ { "id": "generate-code", "name": "生成代码", "description": "根据自然语言需求生成可运行的代码片段", "examples": [ "用Python写一个快速排序函数", "生成一个Express路由处理用户登录" ], "inputModes": ["text/plain"], "outputModes": ["text/plain"] } ] }

这里最关键的是skills数组。调度Agent拿到这张名片后,会拿用户任务去和每个skill的description、examples做匹配,决定把任务路由给谁。所以skill的description要写得具体,别写「处理各种任务」这种废话,否则路由会失准。

再看审查Agent的名片,结构一样,只是skill不同:

{ "name": "code-reviewer-agent", "description": "对代码进行规范审查与安全审查", "url": "http://localhost:8002/a2a", "version": "1.0.0", "capabilities": { "streaming": false, "pushNotifications": true }, "skills": [ { "id": "review-code", "name": "审查代码", "description": "检查代码规范、潜在bug和安全问题,输出审查报告", "examples": ["审查这段Python代码的异常处理是否完整"], "inputModes": ["text/plain"], "outputModes": ["application/json"] } ] }

接下来是Task请求。调度Agent委托任务时,发的是一个JSON-RPC风格的请求,核心字段是task和message:

{ "jsonrpc": "2.0", "id": "task-20250101-001", "method": "tasks/send", "params": { "id": "task-20250101-001", "message": { "role": "user", "parts": [ { "type": "text", "text": "用Python写一个带重试机制的HTTP请求函数,超时3秒,最多重试3次" } ] }, "metadata": { "callbackUrl": "http://localhost:8000/a2a/callback", "priority": "normal" } } }

注意metadata里的callbackUrl,这是异步回调的地址。如果接收方支持pushNotifications,任务完成后会主动POST结果到这个地址,调度Agent就不用轮询了。Task的生命周期状态是submitted → working → completed/failed,你可以在回调体里看到最终状态:

{ "jsonrpc": "2.0", "id": "task-20250101-001", "result": { "id": "task-20250101-001", "status": { "state": "completed", "timestamp": "2025-01-01T10:00:05Z" }, "artifacts": [ { "name": "generated_code", "parts": [ { "type": "text", "text": "import requests\nimport time\n\ndef fetch_with_retry(url, retries=3, timeout=3):\n ..." } ] } ] } }

artifacts就是任务的产出,可以是文本、文件、结构化数据。调度Agent拿到artifacts后,可以再创建一个新Task,把这段代码委托给审查Agent。整个链路里,调度Agent始终是轻量的,它不关心生码Agent内部调了几次模型、用了什么工具,只关心Task的状态和产出。

如果你用的是Cline MCP或者Codex这类工具做Agent宿主,配置里要写全三件套:Base URL填https://taotoken.net/api,Key填你的TaoToken Key,Model ID填你选定的模型。三者缺一,Agent就调不通模型,A2A协作也就无从谈起。

4. 端到端验证:一次调度Agent到生码Agent的协作实测

配置写完了,得跑一次真实的协作验证,否则你不知道Agent Card发现和Task分发到底通没通。我下面用一个最小可跑的Python示例,演示调度Agent如何发现生码Agent、委托任务、拿到结果。你可以直接复制到本地改。

先写一个极简的A2A服务端,模拟生码Agent。它做三件事:暴露Agent Card、接收Task、返回artifacts。

from fastapi import FastAPI, Request import uvicorn import time app = FastAPI() AGENT_CARD = { "name": "code-generator-agent", "description": "根据需求生成代码", "url": "http://localhost:8001/a2a", "version": "1.0.0", "capabilities": {"streaming": False, "pushNotifications": False}, "skills": [ { "id": "generate-code", "name": "生成代码", "description": "根据自然语言需求生成代码片段", "examples": ["写一个快速排序"] } ] } @app.get("/.well-known/agent-card.json") def agent_card(): return AGENT_CARD @app.post("/a2a") async def handle_task(request: Request): body = await request.json() task_id = body["params"]["id"] user_text = body["params"]["message"]["parts"][0]["text"] # 这里真实场景会调TaoToken的模型接口生成代码 generated = f"# 根据需求生成:{user_text}\nprint('hello a2a')" return { "jsonrpc": "2.0", "id": task_id, "result": { "id": task_id, "status": {"state": "completed", "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ")}, "artifacts": [ {"name": "generated_code", "parts": [{"type": "text", "text": generated}]} ] } } if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8001)

启动这个服务后,访问http://localhost:8001/.well-known/agent-card.json,你应该能看到完整的Agent Card JSON。这一步验证的是「名片发布」是否正常。

然后写调度Agent的客户端逻辑,它先拉取Agent Card,再发Task:

import requests import json # 第一步:发现Agent Card card_resp = requests.get("http://localhost:8001/.well-known/agent-card.json") card = card_resp.json() print("发现Agent:", card["name"]) print("可用技能:", [s["id"] for s in card["skills"]]) # 第二步:构造Task请求 task_payload = { "jsonrpc": "2.0", "id": "task-demo-001", "method": "tasks/send", "params": { "id": "task-demo-001", "message": { "role": "user", "parts": [{"type": "text", "text": "写一个Python函数计算斐波那契数列"}] } } } # 第三步:发送Task resp = requests.post("http://localhost:8001/a2a", json=task_payload) result = resp.json() print("任务状态:", result["result"]["status"]["state"]) print("产出:", result["result"]["artifacts"][0]["parts"][0]["text"])

跑通后你会看到类似输出:发现Agent: code-generator-agent,可用技能: ['generate-code'],任务状态: completed,产出里是生成的代码文本。这就是一次完整的A2A协作:发现名片 → 匹配技能 → 委托Task → 接收artifacts。

在真实场景里,生码Agent的handle_task内部会调用TaoToken的模型接口。你可以把上面服务端里的generated那行替换成真实的模型调用:

import os resp = requests.post( f"{os.environ['TAOTOKEN_BASE_URL']}/v1/chat/completions", headers={"Authorization": f"Bearer {os.environ['TAOTOKEN_API_KEY']}"}, json={ "model": os.environ.get("CODER_MODEL", "gpt-4o-mini"), "messages": [{"role": "user", "content": user_text}] } ) generated = resp.json()["choices"][0]["message"]["content"]

这样生码Agent就真正用上了统一Key通道。调度Agent、生码Agent、审查Agent可以各自跑在不同端口,但都通过TaoToken的Base URL调模型,凭证只有一套。验证成功后,你可以再起一个审查Agent,让调度Agent把生码结果作为新Task委托过去,形成「生成→审查」的两级协作链路。

5. 常见报错排查:401、local proxy failed与choices读取失败

多Agent协作跑不起来,十有八九卡在几个固定报错上。我把踩过的坑按报错类型整理出来,你对照着查会快很多。

401 Unauthorized是最常见的。表现是Agent调模型时返回401,或者A2A服务端返回鉴权失败。原因通常有三个:Key没填对、Key前面多了空格、环境变量没生效。排查时先确认你export的TAOTOKEN_API_KEY和实际Key一致,注意别把Bearer前缀重复写进Key里。如果你在Docker里跑Agent,环境变量可能没传进去,用docker exec -it 容器名 env | grep TAOTOKEN确认一下。还有一种情况是Key被复制时带了换行符,用echo -n $TAOTOKEN_API_KEY | wc -c看长度对不对。

local proxy failed这个报错通常出现在你本地配了某些网络工具,或者Agent的HTTP客户端走了系统代理。表现是请求发不出去,或者连到了错误的地址。排查思路是检查环境变量里有没有HTTP_PROXY、HTTPS_PROXY、ALL_PROXY,有的话先unset掉再试。另外确认你的Base URL写的是https://taotoken.net/api,不要多写斜杠或者拼错域名。如果你在容器里跑,容器的DNS配置也可能导致解析失败,用curl -v https://taotoken.net/api/v1/models看具体卡在哪一步。

reading choices 报错一般长这样:KeyError: 'choices'或者list index out of range。这说明你拿到的响应体里没有choices字段,通常是模型调用失败了,返回的是错误JSON。正确做法是先打印完整响应再取字段:

resp = requests.post(url, headers=headers, json=payload) data = resp.json() if "choices" not in data: print("响应异常:", json.dumps(data, ensure_ascii=False)) else: content = data["choices"][0]["message"]["content"]

常见触发原因是模型ID写错了,比如把gpt-4o写成gpt4o,或者用了当前通道不支持的模型名。另一个原因是messages格式不对,比如role写成了user带了空格。你可以在模型对话页面先手动试一次同样的模型ID,确认可用再写进Agent。

OAuth 相关报错多出现在你用Claude Code或某些需要OAuth流程的工具时。表现是提示token过期或授权失败。这类工具通常需要走Anthropic兼容通道,配置时Base URL和Key的填法跟OpenAI格式不同,具体看接入文档。如果你同时混用了OAuth工具和API Key工具,注意别把两套凭证搞混。对于长期跑的编码Agent,用Coding Plan会比反复处理OAuth刷新更省心。

还有一个容易被忽略的坑:A2A的Task ID重复。如果你用时间戳做ID,高并发下可能撞车,导致回调结果覆盖。建议用UUID或者带Agent前缀的ID。另外回调地址如果是localhost,接收方和调度方不在同一台机器时会失败,跨机部署要填真实可达的地址。

6. 把A2A和MCP的边界理清楚,再决定你的接入方式

跑通上面的验证后,你应该对A2A的Agent Card发现和Task分发有了体感。最后我想把A2A和MCP的分工边界再强调一次,因为这直接决定你的架构怎么搭。

MCP解决的是单个Agent怎么连工具和数据。比如你的生码Agent需要读数据库、执行代码、查文档,这些能力通过MCP Server暴露给Agent,Agent用Function Calling触发。A2A解决的是Agent之间怎么分工协作。调度Agent不需要知道生码Agent内部用了哪些MCP工具,它只通过A2A发Task、收artifacts。用一句话概括:MCP向下连工具,A2A向上连Agent。复杂的多Agent系统里,两者通常都要用,不是二选一。

那统一Key在这个架构里扮演什么角色?它是所有Agent调用模型的公共通道。不管你有几个Agent、每个Agent用什么模型,底层都走同一套Base URL和Key。这样做的好处是凭证管理收敛、计费清晰、切换模型不用改多处配置。你可以在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 管理你的Key,在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 查接入细节。如果你的多Agent工作流是长期跑编码和审查任务,Coding Plan会更合适,入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。想先验证某个模型在A2A场景下的表现,可以直接去模型对话页面试。

我的建议是:先用一个调度Agent加一个专业Agent跑通最小闭环,确认Agent Card能被正确发现、Task能正常返回artifacts,再逐步加Agent。每加一个Agent,先单独验证它的模型调用通道,再接入A2A协作。这样出问题时你能快速定位是通道问题还是协议问题。多Agent协作的复杂度不在协议本身,而在调试链路的长度,统一Key至少帮你砍掉了凭证这一层的变量。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/2 6:45:31

WVP-GB28181源码编译与海康大华设备接入实战

简介:本资源是基于GB28181国标开发的wvp-GB28181服务端平台完整源码工程,面向视频监控系统开发者、安防集成工程师及物联网协议学习者,解决GB28181标准在实际项目中的注册、心跳、Invite媒体协商等核心流程落地难题,适用于平安城市…

作者头像 李华
网站建设 2026/10/2 6:45:27

WGCLOUD二次开发:为监控平台添加短信验证码登录

“WGCLOUD支持短信登录系统吗”——这个问题我在不少运维群里看到过。WGCLOUD作为一款开源的服务器监控平台,很多团队拿它来做主机资源监控、告警通知,用着用着就开始琢磨登录方式的事。毕竟现在大家手机不离手,短信验证码登录几乎是企业内部…

作者头像 李华
网站建设 2026/10/2 6:45:14

deepseek接入continue插件实现代码补全:TaoToken统一Key配置与本地验证

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

作者头像 李华