1. 从两个 Agent 互相“装死”说起Python 多智能体协作的真实卡点如果你正在用 Python 写 AI 智能体大概率遇到过这种场面一个 Agent 负责查资料另一个 Agent 负责写报告你把它们分别跑起来结果第一个把结果打印在终端里第二个在另一个进程里干等最后你只能手动复制粘贴。这不是智能体协作这是人工中转。A2AAgent2Agent协议要解决的就是这件事。它是一套开放标准让不同框架、不同进程、甚至不同机器上的 AI 智能体能够互相发现、互相发消息、互相回传任务结果。你可以把它理解成智能体世界的 HTTP以前每个 Agent 都要为对方写一套私有对接代码现在大家说同一种“语言”注册、发现、调用、回传都有统一格式。这篇文章面向的是已经在用 Python 做多智能体、但被通信层卡住的开发者。我会从零给出可复制的 Agent 注册、消息路由、协作编排配置演示两个智能体接力完成同一任务的完整验证步骤并说明如何通过 TaoToken 统一管理调用凭证避免每个 Agent 各配一套 Key 的混乱。全文代码可直接跑环境是 Python 3.10依赖a2a-sdk和httpx。先说清楚 A2A 和 MCP 的关系很多人会混。MCP 解决的是“智能体怎么调用工具和数据源”偏向 Agent 到工具A2A 解决的是“智能体之间怎么对话和分工”偏向 Agent 到 Agent。两者不冲突一个 Agent 可以同时用 MCP 拿工具、用 A2A 找同伴。下面这张表是我自己梳理的对照后面配置时会反复用到。维度MCPA2A核心目的智能体调用工具/数据智能体之间协作类比AI 界的 USB-CAI 界的互联网协议解决问题如何调用 API、查数据库如何发现对方、协商任务、传结果典型场景Agent 需要获取天气数据两个 Agent 协作完成旅行规划通信方向Agent → 工具Agent ↔ Agent理解了这层再看 A2A 的核心组件就顺了。Agent Card 是智能体的“数字名片”描述名称、能力、访问地址和认证方式Task 是一次完整协作单元带唯一 ID 和生命周期状态Message 是智能体之间的对话内容Artifact 是任务完成后产出的文档、图片等交付物。其他 Agent 通过/.well-known/agent-card.json这个固定路径就能拿到名片立刻知道对方能做什么、怎么联系。我试过在没有统一协议的情况下对接三个不同框架的 Agent光是消息格式转换就写了四百多行胶水代码还经常因为字段对不上而静默失败。A2A 把这些约定标准化之后通信层代码量能降一个数量级。接下来进入实操先把调用凭证这件事理顺否则后面每个 Agent 都要重复配 Key。2. TaoToken 前置用统一 Key 管理多 Agent 的模型调用凭证多智能体协作有一个容易被忽略的工程问题每个 Agent 在调用大模型时都需要凭证。如果两个 Agent 分别用不同的 Key、不同的 Base URL协作链路一旦出问题你根本不知道是通信层挂了还是某个 Agent 的模型调用失败了。更麻烦的是Agent 之间传递任务时接收方往往需要自己再调一次模型来理解任务这时候凭证从哪来我的做法是把模型调用统一收敛到 TaoToken 的 API 通道。TaoToken 提供统一的 Key 和 API 入口多个 Agent 共用同一套凭证配置调用日志也集中在一处排查问题时能快速区分“是 A2A 通信失败”还是“模型调用失败”。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要准备的东西不多一个 TaoToken 账号、一个 API Key、以及要用的模型 ID。Key 在控制台创建地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建后复制保存后面所有 Agent 共用它。模型 ID 可以在模型对话页面确认地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 选一个你常用的即可。这里要强调一个原则A2A 负责 Agent 之间的通信TaoToken 负责 Agent 到模型的调用两者分层清晰。不要把模型 Key 塞进 A2A 的 Message 里传来传去那样既不安全也难维护。正确做法是每个 Agent 在本地读取同一份环境变量通信层只传任务内容不传凭证。环境变量这样设置Linux/macOS 用 exportWindows 用 set写进.env文件更稳妥# .env 文件所有 Agent 共用 TAOTOKEN_API_KEYsk-你的TaoToken密钥 TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_ID你的模型ID然后在 Python 里统一读取封装成一个客户端工厂避免每个 Agent 重复写# llm_client.py import os from openai import OpenAI def build_llm_client(): 所有 Agent 共用同一个模型客户端配置 api_key os.getenv(TAOTOKEN_API_KEY) base_url os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) if not api_key: raise RuntimeError(缺少 TAOTOKEN_API_KEY请检查 .env 文件) return OpenAI(api_keyapi_key, base_urlbase_url) def get_model_id(): return os.getenv(TAOTOKEN_MODEL_ID, 你的模型ID)这样两个 Agent 各自 import 这个模块凭证只有一份改 Key 只改一处。如果你后面要上 Coding Plan 做长期编码类 Agent也可以在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 了解套餐但本文的协作演示用按量 Key 就够了。凭证理顺之后A2A 的通信层就可以专心处理任务分发和结果回传不用再操心模型调用。下面进入可复制配置环节我会给出 Agent 注册、消息路由和协作编排的完整代码。3. 可复制配置Agent 注册、消息路由与协作编排这一节是全文的核心我会给出三个可直接复制的配置片段Agent Card 的 JSON 定义、A2A 服务端的注册配置、以及协作编排的 settings 片段。路径和字段都按 A2A 规范来你改掉 URL 和名称就能用。先看 Agent Card。这是每个 Agent 对外暴露的“名片”放在服务端的/.well-known/agent-card.json路径下。下面是一个“资料检索 Agent”的名片注意skills里要写清楚它能做什么authentication里声明认证方式{ name: ResearchAgent, description: 负责检索和整理资料的智能体支持关键词搜索与摘要生成, url: http://localhost:8000, version: 1.0.0, capabilities: { streaming: true, pushNotifications: false }, skills: [ { id: search_and_summarize, name: 检索并摘要, description: 根据关键词检索资料并生成结构化摘要, examples: [检索 A2A 协议的核心概念, 整理多智能体协作的要点] } ], authentication: { schemes: [apiKey] } }第二个片段是 A2A 服务端的注册配置。我用 TOML 写放在项目根目录的a2a_config.toml服务启动时读取。这样 Agent 的地址、端口、认证方式都集中管理不用散落在代码里# a2a_config.toml [server] host 0.0.0.0 port 8000 agent_card_path /.well-known/agent-card.json [agent] name ResearchAgent version 1.0.0 [authentication] scheme apiKey header X-API-Key [llm] base_url https://taotoken.net/api model_id 你的模型ID # api_key 从环境变量 TAOTOKEN_API_KEY 读取不写进配置文件第三个片段是协作编排的 settings我用 Python 的 dataclass 定义放在orchestrator_settings.py。它描述主控 Agent 要协调哪些子 Agent、每个子 Agent 的地址和职责# orchestrator_settings.py from dataclasses import dataclass, field from typing import List dataclass class SubAgentConfig: name: str url: str role: str dataclass class OrchestratorSettings: main_agent_name: str OrchestratorAgent sub_agents: List[SubAgentConfig] field(default_factorylambda: [ SubAgentConfig( nameResearchAgent, urlhttp://localhost:8000, role检索并摘要资料 ), SubAgentConfig( nameWriterAgent, urlhttp://localhost:8001, role根据摘要撰写最终报告 ), ]) timeout_seconds: int 60 max_retries: int 2这三个片段合起来就构成了 A2A 协作的配置骨架Agent Card 定义能力TOML 定义服务端和模型通道settings 定义编排关系。注意 TOML 里的base_url指向 TaoToken 的 API 入口api_key不落盘从环境变量读这是安全底线。配置写好后服务端代码这样加载并注册# research_agent_server.py import asyncio import tomllib from a2a_sdk import A2AServer, AgentCard, AgentSkill, Message, TextPart from a2a_sdk.transport.http import HTTPServerTransport from llm_client import build_llm_client, get_model_id with open(a2a_config.toml, rb) as f: config tomllib.load(f) llm build_llm_client() model_id get_model_id() async def handle_message(message: Message) - Message: 收到任务后调用模型处理再回传结果 user_text for part in message.parts: if part.type text: user_text part.text break # 通过 TaoToken 统一通道调用模型 completion llm.chat.completions.create( modelmodel_id, messages[ {role: system, content: 你是资料检索助手输出结构化摘要。}, {role: user, content: user_text}, ], ) answer completion.choices[0].message.content return Message(roleassistant, parts[TextPart(textanswer)]) agent_card AgentCard( nameconfig[agent][name], description负责检索和整理资料的智能体, versionconfig[agent][version], urlfhttp://localhost:{config[server][port]}, capabilities{streaming: True, pushNotifications: False}, skills[ AgentSkill( idsearch_and_summarize, name检索并摘要, description根据关键词检索资料并生成结构化摘要, examples[检索 A2A 协议的核心概念], ) ], authentication{schemes: [apiKey]}, ) async def main(): transport HTTPServerTransport( hostconfig[server][host], portconfig[server][port], ) server A2AServer( agent_cardagent_card, message_handlerhandle_message, transporttransport, ) print(fResearchAgent 启动名片地址: http://localhost:{config[server][port]}/.well-known/agent-card.json) await server.start() if __name__ __main__: asyncio.run(main())这段代码的关键点有三个一是从 TOML 读配置二是模型调用走 TaoToken 统一通道三是handle_message只处理任务内容、不碰凭证。第二个 AgentWriterAgent把端口改成 8001、system prompt 改成“根据摘要撰写报告”即可其余结构完全一致。4. 验证请求两个智能体接力完成同一任务配置就绪后最激动人心的部分来了让 ResearchAgent 和 WriterAgent 接力完成一个任务。主控 Agent 先把“检索 A2A 协议要点”发给 ResearchAgent拿到摘要后再把摘要作为输入发给 WriterAgent让它写成一段可读的报告。整个过程通过 A2A 消息路由结果逐级回传。先写主控编排代码这是协作的“大脑”# orchestrator.py import asyncio import httpx from a2a_sdk import A2AClient, Message, TextPart from a2a_sdk.transport.http import HTTPClientTransport from orchestrator_settings import OrchestratorSettings settings OrchestratorSettings() async def discover_agent(base_url: str) - dict: 通过固定路径发现 Agent 名片 card_url f{base_url.rstrip(/)}/.well-known/agent-card.json async with httpx.AsyncClient(timeout10) as client: resp await client.get(card_url) resp.raise_for_status() return resp.json() async def send_task(base_url: str, text: str) - str: 向指定 Agent 发送任务并取回文本结果 transport HTTPClientTransport(base_urlbase_url) client A2AClient(transporttransport) try: msg Message(roleuser, parts[TextPart(texttext)]) resp await client.send_message(msg) for part in resp.parts: if part.type text: return part.text return finally: await client.close() async def run_pipeline(topic: str): print(f任务主题: {topic}\n) # 第一步发现两个子 Agent research_cfg settings.sub_agents[0] writer_cfg settings.sub_agents[1] research_card await discover_agent(research_cfg.url) print(f发现 {research_card[name]}: {research_card[description]}) writer_card await discover_agent(writer_cfg.url) print(f发现 {writer_card[name]}: {writer_card[description]}\n) # 第二步ResearchAgent 检索并摘要 print( 阶段一ResearchAgent 检索中...) summary await send_task( research_cfg.url, f请检索并总结以下主题的核心要点{topic} ) print(f摘要结果:\n{summary}\n) # 第三步WriterAgent 根据摘要写报告 print( 阶段二WriterAgent 撰写中...) report await send_task( writer_cfg.url, f请根据以下摘要撰写一段通顺的报告\n{summary} ) print(f最终报告:\n{report}) return report if __name__ __main__: asyncio.run(run_pipeline(A2A 协议如何实现多智能体协作))运行前先启动两个 Agent 服务端再跑主控# 终端 1 python research_agent_server.py # 终端 2WriterAgent端口 8001 python writer_agent_server.py # 终端 3 python orchestrator.py预期输出大致是这样两个 Agent 接力完成结果逐级回传任务主题: A2A 协议如何实现多智能体协作 发现 ResearchAgent: 负责检索和整理资料的智能体 发现 WriterAgent: 根据摘要撰写报告的智能体 阶段一ResearchAgent 检索中... 摘要结果: 1. A2A 通过 Agent Card 实现能力发现 2. 通过 Message 和 Task 实现任务分发 3. 通过 Artifact 回传最终交付物 阶段二WriterAgent 撰写中... 最终报告: A2A 协议让多智能体协作像人类团队一样自然。首先每个智能体通过 Agent Card 公开自己的能力其次主控智能体用 Message 分发任务 最后结果以 Artifact 形式回传形成完整闭环。看到这个输出说明 A2A 通信链路、TaoToken 模型调用、任务接力全部打通。你可以把topic换成任意主题两个 Agent 会自动完成检索和撰写。如果想验证模型调用是否走的是 TaoToken 通道可以在控制台看调用日志地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 每次请求都有记录。这里有个细节值得说主控 Agent 本身也可以是一个 A2A 服务端被更上层的调度器调用。这样层层嵌套就能搭出复杂的多智能体协作网络。验证通过后下一步就是排错因为真实环境里报错比成功更常见。5. 本篇常见错排查401、local proxy failed 与 reading choices协作链路跑通一次不代表稳定下面这几个报错我在调试时都真实遇到过按顺序排查基本能定位。第一个是401 Unauthorized。这个几乎都出在凭证层不是 A2A 通信层。先检查.env里的TAOTOKEN_API_KEY是否被正确加载Python 里可以用os.getenv打印确认。如果 Key 正确但仍 401检查 Base URL 是否写成了https://taotoken.net/api少写/api或写成别的路径都会认证失败。还有一种情况是 Agent 之间传递任务时接收方误把发送方的 Key 当成了自己的这时候要确认每个 Agent 都从本地环境变量读 Key而不是从 Message 里取。第二个是local proxy failed或连接被拒绝。这个报错通常出现在 A2A 客户端连不上服务端时。先确认服务端真的在监听curl http://localhost:8000/.well-known/agent-card.json能返回 JSON 才说明名片可访问。如果返回 404检查agent_card_path配置和实际注册路径是否一致。如果服务端在容器里localhost要换成容器网络里的服务名。端口冲突也会导致这个错用lsof -i :8000看端口是否被占。第三个是reading choices或NoneType has no attribute choices。这个报错说明模型调用返回了空结果代码却直接去取completion.choices[0]。常见原因是模型 ID 写错或者请求超时被中断。排查方法是在handle_message里加一层判断completion llm.chat.completions.create( modelmodel_id, messages[...], ) if not completion or not completion.choices: return Message(roleassistant, parts[TextPart(text模型调用返回为空请检查模型 ID 和网络)]) answer completion.choices[0].message.content第四个是 OAuth 相关的invalid_token或OAuth token expired。如果你给 Agent Card 配了oauth2认证但客户端没带 token 或 token 过期就会报这个。本文演示用的是apiKey方案相对简单。如果你确实要用 OAuth确保客户端在请求头里带Authorization: Bearer token并且 token 没过期。排查时先用curl手动带 token 请求一次确认服务端认证逻辑没问题再回到 A2A 客户端。为了让你对照更快我把常见报错和对应检查点整理成表报错信息可能原因检查点401 UnauthorizedKey 错误或 Base URL 不对环境变量、https://taotoken.net/apilocal proxy failed服务端未启动或端口冲突curl名片路径、lsof查端口reading choices模型返回为空模型 ID、超时、加空值判断OAuth token expiredtoken 过期或未携带请求头 Authorization、token 有效期Agent Card 404路径不一致agent_card_path与实际注册路径排错的核心思路是分层先确认模型调用层TaoToken 通道是否正常再确认 A2A 通信层是否正常最后确认编排逻辑是否正确。三层分开验证比一股脑看日志快得多。如果你在接入文档里找不到对应说明可以到 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 查接口细节。6. 把协作链路固化下来从能跑到好用的三个习惯跑通一次演示只是起点真正让多智能体协作在生产里稳定靠的是几个工程习惯。第一个习惯是凭证集中管理所有 Agent 共用 TaoToken 的统一 Key 和 Base URL改一处生效全局绝不把 Key 写进代码或 Message。第二个习惯是 Agent Card 先行先定义清楚每个 Agent 的能力边界和输入输出再写实现这样编排时不会出现“这个 Agent 到底能不能干这个”的扯皮。第三个习惯是分层排错模型调用层和 A2A 通信层分开验证报错时先定位在哪一层再深入。如果你打算把协作链路长期跑起来尤其是涉及编码类 Agent 或需要持续调用的场景可以了解 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合高频、长期的 Agent 调用。而日常验证模型输出是否正常用模型对话页面就够了地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。最后留一个可以直接上手的动作把本文的orchestrator.py里的topic换成你手头真实的任务比如“整理本周项目进展”或“检索某个技术方案的优缺点”让两个 Agent 接力跑一遍。跑通之后再尝试加第三个 Agent比如一个专门做事实核查的 ReviewerAgent插在 WriterAgent 之后。A2A 的好处就在这里加一个 Agent 只需要多注册一张名片、多配一个地址编排逻辑改几行不用重写通信层。 
长河君 · 2026/10/11 9:40:49 · 阅读 1.2k 【全网首发】Python+AI智能体通信：用A2A协议让AI像人类团队一样协作（附完整代码）
1. 从两个 Agent 互相“装死”说起Python 多智能体协作的真实卡点如果你正在用 Python 写 AI 智能体大概率遇到过这种场面一个 Agent 负责查资料另一个 Agent 负责写报告你把它们分别跑起来结果第一个把结果打印在终端里第二个在另一个进程里干等最后你只能手动复制粘贴。这不是智能体协作这是人工中转。A2AAgent2Agent协议要解决的就是这件事。它是一套开放标准让不同框架、不同进程、甚至不同机器上的 AI 智能体能够互相发现、互相发消息、互相回传任务结果。你可以把它理解成智能体世界的 HTTP以前每个 Agent 都要为对方写一套私有对接代码现在大家说同一种“语言”注册、发现、调用、回传都有统一格式。这篇文章面向的是已经在用 Python 做多智能体、但被通信层卡住的开发者。我会从零给出可复制的 Agent 注册、消息路由、协作编排配置演示两个智能体接力完成同一任务的完整验证步骤并说明如何通过 TaoToken 统一管理调用凭证避免每个 Agent 各配一套 Key 的混乱。全文代码可直接跑环境是 Python 3.10依赖a2a-sdk和httpx。先说清楚 A2A 和 MCP 的关系很多人会混。MCP 解决的是“智能体怎么调用工具和数据源”偏向 Agent 到工具A2A 解决的是“智能体之间怎么对话和分工”偏向 Agent 到 Agent。两者不冲突一个 Agent 可以同时用 MCP 拿工具、用 A2A 找同伴。下面这张表是我自己梳理的对照后面配置时会反复用到。维度MCPA2A核心目的智能体调用工具/数据智能体之间协作类比AI 界的 USB-CAI 界的互联网协议解决问题如何调用 API、查数据库如何发现对方、协商任务、传结果典型场景Agent 需要获取天气数据两个 Agent 协作完成旅行规划通信方向Agent → 工具Agent ↔ Agent理解了这层再看 A2A 的核心组件就顺了。Agent Card 是智能体的“数字名片”描述名称、能力、访问地址和认证方式Task 是一次完整协作单元带唯一 ID 和生命周期状态Message 是智能体之间的对话内容Artifact 是任务完成后产出的文档、图片等交付物。其他 Agent 通过/.well-known/agent-card.json这个固定路径就能拿到名片立刻知道对方能做什么、怎么联系。我试过在没有统一协议的情况下对接三个不同框架的 Agent光是消息格式转换就写了四百多行胶水代码还经常因为字段对不上而静默失败。A2A 把这些约定标准化之后通信层代码量能降一个数量级。接下来进入实操先把调用凭证这件事理顺否则后面每个 Agent 都要重复配 Key。2. TaoToken 前置用统一 Key 管理多 Agent 的模型调用凭证多智能体协作有一个容易被忽略的工程问题每个 Agent 在调用大模型时都需要凭证。如果两个 Agent 分别用不同的 Key、不同的 Base URL协作链路一旦出问题你根本不知道是通信层挂了还是某个 Agent 的模型调用失败了。更麻烦的是Agent 之间传递任务时接收方往往需要自己再调一次模型来理解任务这时候凭证从哪来我的做法是把模型调用统一收敛到 TaoToken 的 API 通道。TaoToken 提供统一的 Key 和 API 入口多个 Agent 共用同一套凭证配置调用日志也集中在一处排查问题时能快速区分“是 A2A 通信失败”还是“模型调用失败”。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要准备的东西不多一个 TaoToken 账号、一个 API Key、以及要用的模型 ID。Key 在控制台创建地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建后复制保存后面所有 Agent 共用它。模型 ID 可以在模型对话页面确认地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 选一个你常用的即可。这里要强调一个原则A2A 负责 Agent 之间的通信TaoToken 负责 Agent 到模型的调用两者分层清晰。不要把模型 Key 塞进 A2A 的 Message 里传来传去那样既不安全也难维护。正确做法是每个 Agent 在本地读取同一份环境变量通信层只传任务内容不传凭证。环境变量这样设置Linux/macOS 用 exportWindows 用 set写进.env文件更稳妥# .env 文件所有 Agent 共用 TAOTOKEN_API_KEYsk-你的TaoToken密钥 TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_ID你的模型ID然后在 Python 里统一读取封装成一个客户端工厂避免每个 Agent 重复写# llm_client.py import os from openai import OpenAI def build_llm_client(): 所有 Agent 共用同一个模型客户端配置 api_key os.getenv(TAOTOKEN_API_KEY) base_url os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) if not api_key: raise RuntimeError(缺少 TAOTOKEN_API_KEY请检查 .env 文件) return OpenAI(api_keyapi_key, base_urlbase_url) def get_model_id(): return os.getenv(TAOTOKEN_MODEL_ID, 你的模型ID)这样两个 Agent 各自 import 这个模块凭证只有一份改 Key 只改一处。如果你后面要上 Coding Plan 做长期编码类 Agent也可以在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 了解套餐但本文的协作演示用按量 Key 就够了。凭证理顺之后A2A 的通信层就可以专心处理任务分发和结果回传不用再操心模型调用。下面进入可复制配置环节我会给出 Agent 注册、消息路由和协作编排的完整代码。3. 可复制配置Agent 注册、消息路由与协作编排这一节是全文的核心我会给出三个可直接复制的配置片段Agent Card 的 JSON 定义、A2A 服务端的注册配置、以及协作编排的 settings 片段。路径和字段都按 A2A 规范来你改掉 URL 和名称就能用。先看 Agent Card。这是每个 Agent 对外暴露的“名片”放在服务端的/.well-known/agent-card.json路径下。下面是一个“资料检索 Agent”的名片注意skills里要写清楚它能做什么authentication里声明认证方式{ name: ResearchAgent, description: 负责检索和整理资料的智能体支持关键词搜索与摘要生成, url: http://localhost:8000, version: 1.0.0, capabilities: { streaming: true, pushNotifications: false }, skills: [ { id: search_and_summarize, name: 检索并摘要, description: 根据关键词检索资料并生成结构化摘要, examples: [检索 A2A 协议的核心概念, 整理多智能体协作的要点] } ], authentication: { schemes: [apiKey] } }第二个片段是 A2A 服务端的注册配置。我用 TOML 写放在项目根目录的a2a_config.toml服务启动时读取。这样 Agent 的地址、端口、认证方式都集中管理不用散落在代码里# a2a_config.toml [server] host 0.0.0.0 port 8000 agent_card_path /.well-known/agent-card.json [agent] name ResearchAgent version 1.0.0 [authentication] scheme apiKey header X-API-Key [llm] base_url https://taotoken.net/api model_id 你的模型ID # api_key 从环境变量 TAOTOKEN_API_KEY 读取不写进配置文件第三个片段是协作编排的 settings我用 Python 的 dataclass 定义放在orchestrator_settings.py。它描述主控 Agent 要协调哪些子 Agent、每个子 Agent 的地址和职责# orchestrator_settings.py from dataclasses import dataclass, field from typing import List dataclass class SubAgentConfig: name: str url: str role: str dataclass class OrchestratorSettings: main_agent_name: str OrchestratorAgent sub_agents: List[SubAgentConfig] field(default_factorylambda: [ SubAgentConfig( nameResearchAgent, urlhttp://localhost:8000, role检索并摘要资料 ), SubAgentConfig( nameWriterAgent, urlhttp://localhost:8001, role根据摘要撰写最终报告 ), ]) timeout_seconds: int 60 max_retries: int 2这三个片段合起来就构成了 A2A 协作的配置骨架Agent Card 定义能力TOML 定义服务端和模型通道settings 定义编排关系。注意 TOML 里的base_url指向 TaoToken 的 API 入口api_key不落盘从环境变量读这是安全底线。配置写好后服务端代码这样加载并注册# research_agent_server.py import asyncio import tomllib from a2a_sdk import A2AServer, AgentCard, AgentSkill, Message, TextPart from a2a_sdk.transport.http import HTTPServerTransport from llm_client import build_llm_client, get_model_id with open(a2a_config.toml, rb) as f: config tomllib.load(f) llm build_llm_client() model_id get_model_id() async def handle_message(message: Message) - Message: 收到任务后调用模型处理再回传结果 user_text for part in message.parts: if part.type text: user_text part.text break # 通过 TaoToken 统一通道调用模型 completion llm.chat.completions.create( modelmodel_id, messages[ {role: system, content: 你是资料检索助手输出结构化摘要。}, {role: user, content: user_text}, ], ) answer completion.choices[0].message.content return Message(roleassistant, parts[TextPart(textanswer)]) agent_card AgentCard( nameconfig[agent][name], description负责检索和整理资料的智能体, versionconfig[agent][version], urlfhttp://localhost:{config[server][port]}, capabilities{streaming: True, pushNotifications: False}, skills[ AgentSkill( idsearch_and_summarize, name检索并摘要, description根据关键词检索资料并生成结构化摘要, examples[检索 A2A 协议的核心概念], ) ], authentication{schemes: [apiKey]}, ) async def main(): transport HTTPServerTransport( hostconfig[server][host], portconfig[server][port], ) server A2AServer( agent_cardagent_card, message_handlerhandle_message, transporttransport, ) print(fResearchAgent 启动名片地址: http://localhost:{config[server][port]}/.well-known/agent-card.json) await server.start() if __name__ __main__: asyncio.run(main())这段代码的关键点有三个一是从 TOML 读配置二是模型调用走 TaoToken 统一通道三是handle_message只处理任务内容、不碰凭证。第二个 AgentWriterAgent把端口改成 8001、system prompt 改成“根据摘要撰写报告”即可其余结构完全一致。4. 验证请求两个智能体接力完成同一任务配置就绪后最激动人心的部分来了让 ResearchAgent 和 WriterAgent 接力完成一个任务。主控 Agent 先把“检索 A2A 协议要点”发给 ResearchAgent拿到摘要后再把摘要作为输入发给 WriterAgent让它写成一段可读的报告。整个过程通过 A2A 消息路由结果逐级回传。先写主控编排代码这是协作的“大脑”# orchestrator.py import asyncio import httpx from a2a_sdk import A2AClient, Message, TextPart from a2a_sdk.transport.http import HTTPClientTransport from orchestrator_settings import OrchestratorSettings settings OrchestratorSettings() async def discover_agent(base_url: str) - dict: 通过固定路径发现 Agent 名片 card_url f{base_url.rstrip(/)}/.well-known/agent-card.json async with httpx.AsyncClient(timeout10) as client: resp await client.get(card_url) resp.raise_for_status() return resp.json() async def send_task(base_url: str, text: str) - str: 向指定 Agent 发送任务并取回文本结果 transport HTTPClientTransport(base_urlbase_url) client A2AClient(transporttransport) try: msg Message(roleuser, parts[TextPart(texttext)]) resp await client.send_message(msg) for part in resp.parts: if part.type text: return part.text return finally: await client.close() async def run_pipeline(topic: str): print(f任务主题: {topic}\n) # 第一步发现两个子 Agent research_cfg settings.sub_agents[0] writer_cfg settings.sub_agents[1] research_card await discover_agent(research_cfg.url) print(f发现 {research_card[name]}: {research_card[description]}) writer_card await discover_agent(writer_cfg.url) print(f发现 {writer_card[name]}: {writer_card[description]}\n) # 第二步ResearchAgent 检索并摘要 print( 阶段一ResearchAgent 检索中...) summary await send_task( research_cfg.url, f请检索并总结以下主题的核心要点{topic} ) print(f摘要结果:\n{summary}\n) # 第三步WriterAgent 根据摘要写报告 print( 阶段二WriterAgent 撰写中...) report await send_task( writer_cfg.url, f请根据以下摘要撰写一段通顺的报告\n{summary} ) print(f最终报告:\n{report}) return report if __name__ __main__: asyncio.run(run_pipeline(A2A 协议如何实现多智能体协作))运行前先启动两个 Agent 服务端再跑主控# 终端 1 python research_agent_server.py # 终端 2WriterAgent端口 8001 python writer_agent_server.py # 终端 3 python orchestrator.py预期输出大致是这样两个 Agent 接力完成结果逐级回传任务主题: A2A 协议如何实现多智能体协作 发现 ResearchAgent: 负责检索和整理资料的智能体 发现 WriterAgent: 根据摘要撰写报告的智能体 阶段一ResearchAgent 检索中... 摘要结果: 1. A2A 通过 Agent Card 实现能力发现 2. 通过 Message 和 Task 实现任务分发 3. 通过 Artifact 回传最终交付物 阶段二WriterAgent 撰写中... 最终报告: A2A 协议让多智能体协作像人类团队一样自然。首先每个智能体通过 Agent Card 公开自己的能力其次主控智能体用 Message 分发任务 最后结果以 Artifact 形式回传形成完整闭环。看到这个输出说明 A2A 通信链路、TaoToken 模型调用、任务接力全部打通。你可以把topic换成任意主题两个 Agent 会自动完成检索和撰写。如果想验证模型调用是否走的是 TaoToken 通道可以在控制台看调用日志地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 每次请求都有记录。这里有个细节值得说主控 Agent 本身也可以是一个 A2A 服务端被更上层的调度器调用。这样层层嵌套就能搭出复杂的多智能体协作网络。验证通过后下一步就是排错因为真实环境里报错比成功更常见。5. 本篇常见错排查401、local proxy failed 与 reading choices协作链路跑通一次不代表稳定下面这几个报错我在调试时都真实遇到过按顺序排查基本能定位。第一个是401 Unauthorized。这个几乎都出在凭证层不是 A2A 通信层。先检查.env里的TAOTOKEN_API_KEY是否被正确加载Python 里可以用os.getenv打印确认。如果 Key 正确但仍 401检查 Base URL 是否写成了https://taotoken.net/api少写/api或写成别的路径都会认证失败。还有一种情况是 Agent 之间传递任务时接收方误把发送方的 Key 当成了自己的这时候要确认每个 Agent 都从本地环境变量读 Key而不是从 Message 里取。第二个是local proxy failed或连接被拒绝。这个报错通常出现在 A2A 客户端连不上服务端时。先确认服务端真的在监听curl http://localhost:8000/.well-known/agent-card.json能返回 JSON 才说明名片可访问。如果返回 404检查agent_card_path配置和实际注册路径是否一致。如果服务端在容器里localhost要换成容器网络里的服务名。端口冲突也会导致这个错用lsof -i :8000看端口是否被占。第三个是reading choices或NoneType has no attribute choices。这个报错说明模型调用返回了空结果代码却直接去取completion.choices[0]。常见原因是模型 ID 写错或者请求超时被中断。排查方法是在handle_message里加一层判断completion llm.chat.completions.create( modelmodel_id, messages[...], ) if not completion or not completion.choices: return Message(roleassistant, parts[TextPart(text模型调用返回为空请检查模型 ID 和网络)]) answer completion.choices[0].message.content第四个是 OAuth 相关的invalid_token或OAuth token expired。如果你给 Agent Card 配了oauth2认证但客户端没带 token 或 token 过期就会报这个。本文演示用的是apiKey方案相对简单。如果你确实要用 OAuth确保客户端在请求头里带Authorization: Bearer token并且 token 没过期。排查时先用curl手动带 token 请求一次确认服务端认证逻辑没问题再回到 A2A 客户端。为了让你对照更快我把常见报错和对应检查点整理成表报错信息可能原因检查点401 UnauthorizedKey 错误或 Base URL 不对环境变量、https://taotoken.net/apilocal proxy failed服务端未启动或端口冲突curl名片路径、lsof查端口reading choices模型返回为空模型 ID、超时、加空值判断OAuth token expiredtoken 过期或未携带请求头 Authorization、token 有效期Agent Card 404路径不一致agent_card_path与实际注册路径排错的核心思路是分层先确认模型调用层TaoToken 通道是否正常再确认 A2A 通信层是否正常最后确认编排逻辑是否正确。三层分开验证比一股脑看日志快得多。如果你在接入文档里找不到对应说明可以到 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 查接口细节。6. 把协作链路固化下来从能跑到好用的三个习惯跑通一次演示只是起点真正让多智能体协作在生产里稳定靠的是几个工程习惯。第一个习惯是凭证集中管理所有 Agent 共用 TaoToken 的统一 Key 和 Base URL改一处生效全局绝不把 Key 写进代码或 Message。第二个习惯是 Agent Card 先行先定义清楚每个 Agent 的能力边界和输入输出再写实现这样编排时不会出现“这个 Agent 到底能不能干这个”的扯皮。第三个习惯是分层排错模型调用层和 A2A 通信层分开验证报错时先定位在哪一层再深入。如果你打算把协作链路长期跑起来尤其是涉及编码类 Agent 或需要持续调用的场景可以了解 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合高频、长期的 Agent 调用。而日常验证模型输出是否正常用模型对话页面就够了地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。最后留一个可以直接上手的动作把本文的orchestrator.py里的topic换成你手头真实的任务比如“整理本周项目进展”或“检索某个技术方案的优缺点”让两个 Agent 接力跑一遍。跑通之后再尝试加第三个 Agent比如一个专门做事实核查的 ReviewerAgent插在 WriterAgent 之后。A2A 的好处就在这里加一个 Agent 只需要多注册一张名片、多配一个地址编排逻辑改几行不用重写通信层。