☰
Agent to Agent协议实战:用TaoToken统一Key打通智能体协作链路
2026/10/8 2:29:57 网站建设 项目流程

1. 从单体智能体到协作网络:A2A协议到底解决什么问题

如果你最近在折腾多智能体系统,大概率会遇到一个很现实的场景:你写了一个负责查资料的 Agent,又写了一个负责写代码的 Agent,还想再加一个负责跑测试的 Agent。三个 Agent 各自跑得挺好,但一旦让它们互相调用,问题就来了——A 用 JSON 传参,B 只认自然语言,C 要求 OAuth 鉴权,光是打通通信就耗掉大半天。

这就是 Agent to Agent(A2A)协议要解决的核心痛点。A2A 是 Google 在 2025 年 4 月发布的开放协议,目标很明确:让不同厂商、不同框架构建的智能体能够像人类团队一样分工协作。它定义了一套标准通信方式,包括 Agent Card(代理卡片,放在/.well-known/agent.json路径下描述能力)、Task(任务生命周期管理)、Message(消息载体)和 Artifact(产物交付)等核心组件。

A2A 和 MCP 的关系经常被搞混。简单说,MCP 管的是"模型怎么调用工具和数据源",A2A 管的是"智能体之间怎么互相派活和汇报"。一个典型链路是:客服 Agent 通过 A2A 把"修改配送地址"的任务派给物流 Agent,物流 Agent 再通过 MCP 调用外部 API 拿到最新配送信息。两者互补,不冲突。

那为什么需要 TaoToken?因为当你真的把多个 Agent 串起来跑的时候,每个 Agent 都要调模型、都要鉴权。如果每个 Agent 单独申请一套 Key、单独配一套通道,管理成本会迅速膨胀。TaoToken 提供统一 Key 和统一 API 通道,让所有 Agent 共用一套调用凭证,同时保持各自的模型选择灵活。这篇教程就带你从零跑通一条 A2A 风格的智能体协作链路,用 TaoToken 统一管理所有 Agent 的模型调用。

适合谁看:已经写过至少一个 Agent demo、想往多智能体协作方向推进的开发者;或者正在评估 A2A 协议落地方式、想先跑通最小可行链路的团队。

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

在开始搭多智能体协作链路之前,先把 TaoToken 的调用凭证和通道准备好。这一步看起来简单,但后面所有 Agent 的模型调用都依赖它,配错了会浪费很多排查时间。

2.1 获取 API Key 与确认 Base URL

首先到 TaoToken 控制台创建一个 API Key。地址是https://taotoken.net/api-keys,登录后点创建,复制出来的 Key 格式类似sk-xxxxxxxx。这个 Key 就是你所有 Agent 共用的统一凭证。

Base URL 固定为https://taotoken.net/api,注意不要加 UTM 参数,也不要加尾部斜杠。后面在代码里配置的时候,OpenAI 兼容接口的 base_url 就填这个。

注意:Key 只在创建时完整显示一次,复制后存到环境变量或密钥管理工具里,不要硬编码进代码提交到 Git。

2.2 确认可用模型 ID

TaoToken 支持多种模型,你需要确认自己要用哪个 Model ID。常见的有claude-sonnet-4-20250514、gpt-4o、deepseek-chat等。具体可用列表在控制台的模型页面能看到。多智能体场景下,不同 Agent 可以用不同模型——比如规划 Agent 用推理强的,执行 Agent 用速度快的,但都走同一个 Key 和 Base URL。

2.3 环境变量配置

推荐用环境变量管理,避免 Key 泄露。在项目根目录创建.env文件:

TAOTOKEN_API_KEY=sk-你的实际Key TAOTOKEN_BASE_URL=https://taotoken.net/api

然后在代码里用os.environ或dotenv读取。如果你用 Claude Code 或 Cline 这类工具,它们的配置文件里也要填这三件套:Base URL、API Key、Model ID。

2.4 验证 Key 是否可用

在正式搭多 Agent 之前,先用一条最简单的请求确认 Key 和通道没问题:

import os from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"] ) resp = client.chat.completions.create( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "回复OK两个字母"}] ) print(resp.choices[0].message.content)

如果输出OK,说明 Key 和通道都正常。如果报 401,检查 Key 是否复制完整;如果报连接错误,检查 Base URL 是否写成了https://taotoken.net/api(不要带路径后缀)。

这一步完成后,你就有了一个所有 Agent 都能共用的模型调用入口。接下来进入 A2A 协作链路的实际搭建。

3. 可复制配置:多智能体 A2A 协作链路搭建

这一节是整篇的核心。我会用一个具体场景来演示:一个"研究 Agent"负责查资料并生成摘要,一个"写作 Agent"负责根据摘要写技术短文,两者通过 A2A 风格的消息传递协作。所有 Agent 的模型调用都走 TaoToken 统一 Key。

3.1 项目结构与依赖

先建目录结构:

mkdir a2a-demo && cd a2a-demo mkdir agents shared touch agents/research_agent.py agents/writer_agent.py shared/a2a_message.py shared/llm_client.py

安装依赖:

pip install openai python-dotenv fastapi uvicorn requests

3.2 统一 LLM 客户端封装

在shared/llm_client.py里封装一个所有 Agent 共用的调用函数:

import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() _client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"] ) def call_llm(model: str, system_prompt: str, user_prompt: str) -> str: resp = _client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.7 ) return resp.choices[0].message.content

这个封装让每个 Agent 只需要传 model 和 prompt,不用关心 Key 和 Base URL。换模型也只改一个参数。

3.3 A2A 消息结构定义

参照 A2A 协议的 Message 和 Task 概念,在shared/a2a_message.py里定义轻量消息结构:

from dataclasses import dataclass, field from typing import Any import uuid import time @dataclass class A2AMessage: role: str # "user" or "agent" parts: list[dict] message_id: str = field(default_factory=lambda: str(uuid.uuid4())) timestamp: float = field(default_factory=time.time) @dataclass class A2ATask: task_id: str status: str # submitted / working / completed / failed skill: str input_message: A2AMessage artifact: Any = None

这里简化了 A2A 的完整规范,但保留了核心字段:任务 ID、状态流转、技能标识、输入消息和产物。实际生产环境可以扩展成完整的 JSON-RPC 格式。

3.4 Agent Card 配置

每个 Agent 需要一个 Agent Card 来描述自己的能力。在agents/research_agent.py里定义:

RESEARCH_AGENT_CARD = { "name": "research-agent", "description": "负责根据主题检索资料并生成结构化摘要", "url": "http://localhost:8001", "version": "1.0.0", "capabilities": { "skills": [ { "id": "summarize", "name": "资料摘要", "description": "输入一个主题,输出该主题的核心要点摘要", "input_modes": ["text"], "output_modes": ["text"] } ] }, "authentication": { "schemes": ["apiKey"], "apiKeyHeader": "X-TaoToken-Key" } }

注意authentication字段里我们声明用 API Key 方案,实际调用时通过 Header 传递 TaoToken 的 Key。这样每个 Agent 对外暴露的鉴权方式统一,内部调模型也统一。

3.5 研究 Agent 实现

from shared.llm_client import call_llm from shared.a2a_message import A2AMessage, A2ATask import uuid RESEARCH_AGENT_CARD = { ... } # 同上 def handle_task(task: A2ATask) -> A2ATask: task.status = "working" topic = task.input_message.parts[0]["text"] system = "你是一个研究助手。根据用户给出的主题,输出3-5条核心要点,每条不超过50字。" result = call_llm( model="claude-sonnet-4-20250514", system_prompt=system, user_prompt=f"主题:{topic}" ) task.artifact = result task.status = "completed" return task def create_research_task(topic: str) -> A2ATask: msg = A2AMessage(role="user", parts=[{"type": "text", "text": topic}]) return A2ATask( task_id=str(uuid.uuid4()), status="submitted", skill="summarize", input_message=msg )

3.6 写作 Agent 实现

写作 Agent 接收研究 Agent 的产物作为输入:

from shared.llm_client import call_llm from shared.a2a_message import A2AMessage, A2ATask import uuid WRITER_AGENT_CARD = { "name": "writer-agent", "description": "根据研究摘要撰写技术短文", "url": "http://localhost:8002", "version": "1.0.0", "capabilities": { "skills": [ { "id": "write_article", "name": "技术短文写作", "description": "输入研究摘要,输出一篇300字左右的技术短文", "input_modes": ["text"], "output_modes": ["text"] } ] }, "authentication": { "schemes": ["apiKey"], "apiKeyHeader": "X-TaoToken-Key" } } def handle_task(task: A2ATask) -> A2ATask: task.status = "working" summary = task.input_message.parts[0]["text"] system = "你是一个技术写作助手。根据给定的研究摘要,写一篇300字左右的技术短文,语言简洁专业。" result = call_llm( model="gpt-4o", system_prompt=system, user_prompt=f"研究摘要:\n{summary}" ) task.artifact = result task.status = "completed" return task def create_writer_task(summary: str) -> A2ATask: msg = A2AMessage(role="agent", parts=[{"type": "text", "text": summary}]) return A2ATask( task_id=str(uuid.uuid4()), status="submitted", skill="write_article", input_message=msg )

注意写作 Agent 用的是gpt-4o,研究 Agent 用的是claude-sonnet-4-20250514,但两者都通过同一个 TaoToken Key 和 Base URL 调用。这就是统一 Key 的价值——模型可以不同,通道和凭证统一。

3.7 协作编排脚本

在项目根目录创建orchestrator.py:

from agents.research_agent import create_research_task, handle_task as research_handle from agents.writer_agent import create_writer_task, handle_task as writer_handle def run_a2a_pipeline(topic: str): print(f"[Orchestrator] 启动 A2A 协作链路,主题:{topic}") # 第一步:研究 Agent 处理 research_task = create_research_task(topic) print(f"[Research Agent] 任务 {research_task.task_id} 状态:{research_task.status}") research_task = research_handle(research_task) print(f"[Research Agent] 任务完成,状态:{research_task.status}") print(f"[Research Agent] 产物:\n{research_task.artifact}\n") # 第二步:将研究产物作为输入传给写作 Agent writer_task = create_writer_task(research_task.artifact) print(f"[Writer Agent] 任务 {writer_task.task_id} 状态:{writer_task.status}") writer_task = writer_handle(writer_task) print(f"[Writer Agent] 任务完成,状态:{writer_task.status}") print(f"[Writer Agent] 最终产物:\n{writer_task.artifact}") return writer_task.artifact if __name__ == "__main__": run_a2a_pipeline("A2A 协议与 MCP 协议的区别")

运行python orchestrator.py,你会看到研究 Agent 先输出摘要,写作 Agent 再基于摘要输出短文。整条链路里,两个 Agent 各自调用了不同的模型,但都走 TaoToken 统一通道。

3.8 用 FastAPI 暴露 Agent 端点(可选进阶)

如果你想让 Agent 真正通过网络互调,可以用 FastAPI 把每个 Agent 包成 HTTP 服务:

from fastapi import FastAPI, Header, HTTPException from agents.research_agent import handle_task, create_research_task app = FastAPI() @app.post("/tasks") def create_and_run(topic: str, x_taotoken_key: str = Header(...)): if not x_taotoken_key.startswith("sk-"): raise HTTPException(status_code=401, detail="Invalid API Key") task = create_research_task(topic) return handle_task(task)

这样研究 Agent 跑在 8001 端口,写作 Agent 跑在 8002 端口,编排器通过 HTTP 调用它们。鉴权统一用X-TaoToken-KeyHeader,内部调模型也用同一个 Key。

4. 验证请求与成功结果:跑通完整协作链路

配置写完后,最关键的一步是验证整条链路真的能跑通。很多人卡在这里,因为多 Agent 场景下出错的地方比单 Agent 多得多。

4.1 分步验证策略

不要一上来就跑完整链路。按这个顺序逐步验证:

第一步,单独验证 TaoToken 通道。运行第 2.4 节那段最简单的请求代码,确认能拿到模型回复。这一步不通,后面全白搭。

第二步,单独验证研究 Agent。在orchestrator.py里只跑研究 Agent 部分:

from agents.research_agent import create_research_task, handle_task task = create_research_task("A2A 协议的核心组件") task = handle_task(task) print(task.status) print(task.artifact)

预期输出:completed和一段 3-5 条要点的摘要。如果状态是failed或者报错,看第 5 节的排查。

第三步,单独验证写作 Agent。手动传一段摘要进去:

from agents.writer_agent import create_writer_task, handle_task task = create_writer_task("A2A 协议包含 Agent Card、Task、Message、Artifact 等核心组件。") task = handle_task(task) print(task.status) print(task.artifact)

预期输出:completed和一篇 300 字左右的短文。

第四步,跑完整编排链路。运行python orchestrator.py,观察两个 Agent 的状态流转和产物传递。

4.2 成功结果长什么样

完整链路跑通后,终端输出大致是这样:

[Orchestrator] 启动 A2A 协作链路,主题:A2A 协议与 MCP 协议的区别 [Research Agent] 任务 3f2a... 状态:submitted [Research Agent] 任务完成,状态:completed [Research Agent] 产物: 1. A2A 专注智能体间协作,MCP 专注模型与工具连接 2. A2A 采用去中心化架构,MCP 是客户端-服务器架构 3. A2A 用自然语言描述任务,MCP 用结构化参数调用 4. 两者互补,共同构建完整 AI 协作生态 [Writer Agent] 任务 8b1c... 状态:submitted [Writer Agent] 任务完成,状态:completed [Writer Agent] 最终产物: A2A 协议与 MCP 协议是当前 AI 智能体领域两个重要的通信标准...

关键验证点:研究 Agent 的artifact非空且内容合理;写作 Agent 的输入确实是研究 Agent 的输出;两个 Agent 的状态都从submitted流转到completed。

4.3 用日志确认模型调用走了 TaoToken

如果你想确认所有调用确实走了 TaoToken 通道,可以在shared/llm_client.py里加一行日志:

import logging logging.basicConfig(level=logging.INFO) def call_llm(model: str, system_prompt: str, user_prompt: str) -> str: logging.info(f"Calling model={model} via base_url={_client.base_url}") resp = _client.chat.completions.create(...) return resp.choices[0].message.content

运行后你会看到类似Calling model=claude-sonnet-4-20250514 via base_url=https://taotoken.net/api的日志。两个 Agent 的日志里 base_url 一致,model 不同,说明统一通道生效。

4.4 验证任务状态流转

A2A 协议强调任务生命周期管理。你可以在编排器里加一个状态检查:

def run_a2a_pipeline(topic: str): research_task = create_research_task(topic) assert research_task.status == "submitted" research_task = research_handle(research_task) assert research_task.status == "completed", f"研究任务失败:{research_task.status}" writer_task = create_writer_task(research_task.artifact) assert writer_task.status == "submitted" writer_task = writer_handle(writer_task) assert writer_task.status == "completed", f"写作任务失败:{writer_task.status}" return writer_task.artifact

这样任何一步状态不对都会立刻报错,方便定位问题。

5. 本篇常见错误排查:401、连接失败与产物为空

多 Agent 协作链路的报错往往比单 Agent 更隐蔽,因为错误可能发生在任何一个 Agent 的模型调用、消息传递或状态流转环节。下面按真实报错场景逐一排查。

5.1 401 Unauthorized

这是最常见的错误。报错信息通常是:

openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key'}}

排查步骤:第一,确认.env文件里的TAOTOKEN_API_KEY是完整的,没有多余空格或换行。第二,确认 Key 没有过期或被删除,到控制台https://taotoken.net/api-keys检查。第三,确认代码里读取环境变量的方式正确,load_dotenv()要在OpenAI()初始化之前调用。第四,如果你在 FastAPI 里用 Header 传 Key,确认 Header 名称和 Agent Card 里声明的一致。

一个容易忽略的点:如果你把 Key 写在了docker-compose.yml或 CI 配置里,确认没有转义字符问题。

5.2 Connection Error / local proxy failed

报错信息类似:

openai.APIConnectionError: Connection error.

或者:

httpx.ConnectError: [Errno 111] Connection refused

这类错误通常是 Base URL 配置问题。排查:第一,确认TAOTOKEN_BASE_URL是https://taotoken.net/api,不要加尾部斜杠,不要加/v1后缀。第二,确认本地网络能正常访问该地址,可以用curl https://taotoken.net/api测试连通性。第三,如果你在公司内网,确认没有防火墙拦截。第四,检查是否有残留的代理环境变量(HTTP_PROXY、HTTPS_PROXY)干扰,临时 unset 掉再试。

注意:不要在任何配置里使用非官方的中转地址,统一用https://taotoken.net/api。

5.3 reading 'choices' 报错

报错信息:

TypeError: Cannot read properties of undefined (reading 'choices')

或者 Python 里:

AttributeError: 'NoneType' object has no attribute 'choices'

这通常说明 API 返回了非预期结构。排查:第一,确认 Model ID 拼写正确,比如claude-sonnet-4-20250514不要写成claude-sonnet-4。第二,确认请求参数里messages格式正确,每条消息必须有role和content。第三,打印完整响应看看实际返回了什么:

resp = _client.chat.completions.create(...) print(resp) # 先看结构

如果返回的是错误对象而不是正常 completion,根据错误信息进一步排查。

5.4 OAuth 相关报错

如果你在 Agent Card 里声明了 OAuth 但实际用的是 API Key,可能会遇到:

OAuth token validation failed

或者:

Missing required authentication scheme

排查:确认 Agent Card 里的authentication.schemes和实际传递的凭证类型一致。本篇教程统一用apiKey方案,Header 名X-TaoToken-Key。如果你要改成 OAuth,需要额外实现 token 获取和刷新逻辑,建议先用 API Key 跑通再升级。

5.5 产物为空或状态卡在 working

如果 Agent 状态一直是working或者artifact为空,排查:第一,确认call_llm返回的字符串非空,可以在函数里加assert result。第二,确认handle_task里task.artifact = result这行确实执行了。第三,如果是网络调用超时,给 OpenAI 客户端加超时参数:

_client = OpenAI( api_key=..., base_url=..., timeout=60.0 )

第四,检查是否有异常被吞掉,在handle_task里加 try-except 打印完整堆栈。

5.6 CC Switch / Cline MCP / Codex auth.json 配置要点

如果你用 CC Switch 管理多个 Claude Code 配置,或者用 Cline 的 MCP 功能,或者用 Codex 的auth.json,记住三件套必须完整:Base URL 填https://taotoken.net/api,API Key 填 TaoToken 的 Key,Model ID 填你确认可用的模型。三者缺一不可,少任何一个都会报鉴权或模型不存在错误。

Codex 的auth.json示例:

{ "api_key": "sk-你的TaoToken Key", "base_url": "https://taotoken.net/api", "model": "claude-sonnet-4-20250514" }

Cline 的 MCP 配置里,如果通过 TaoToken 调用模型,同样在环境变量或配置项里填这三件套。

6. 把统一 Key 用在长期协作链路上

跑通上面这条最小链路后,你可以往几个方向扩展。第一,把 Agent 数量加到三个以上,比如加一个"审核 Agent"检查写作 Agent 的输出质量,形成研究→写作→审核的流水线。第二,把 Agent 拆成独立 HTTP 服务,用 FastAPI 暴露/tasks端点,编排器通过 HTTP 调用,这样更接近真实 A2A 的跨进程协作。第三,给每个 Agent 配不同的模型,规划类用推理强的,执行类用速度快的,但都走 TaoToken 统一 Key,管理成本不变。

如果你打算长期跑多智能体协作,建议把 Key 管理、模型选择、Agent Card 注册这几件事标准化。TaoToken 的 Coding Plan 适合需要长期编码和 Agent 调用的场景,模型对话页面可以快速验证不同模型在协作链路里的表现。接入文档里有完整的 Base URL 和参数说明,遇到配置问题可以先查文档再排查。

实际跑下来,统一 Key 最大的好处不是省事,而是让"换模型"和"加 Agent"这两个动作变得没有心理负担。你不需要为每个新 Agent 重新申请凭证、重新配通道,复制一份配置改个 Model ID 就能跑。这在多智能体协作场景里,比单 Agent 时代重要得多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询