1. 为什么要把 MCP 和 A2A 拧在一起用
如果你最近在折腾 Agent,大概率会遇到一个很别扭的边界问题:Agent 能聊天、能互相转发任务,但一到“帮我查个网页 / 读个文件 / 调个数据库”就卡住了。原因不复杂——A2A 管的是 Agent 和 Agent 之间的协作,MCP 管的是 Agent 和工具之间的连接,这俩协议压根不是竞争关系,而是各管一层。
我把它类比成公司里的两套系统:MCP 像是给每个员工配的“工具接口”,插上就能用打印机、查资料库;A2A 像是员工之间的“内部沟通协议”,谁该把活转给谁、任务状态怎么同步,全靠它。你只上 MCP,Agent 能调工具但不会协作;只上 A2A,Agent 会互相喊话但手里没工具。双协议联合落地,才是 2026 年 Agent 工程化的完整形态。
这篇面向的是已经在本地跑过单 Agent、想进一步搭“工具层 + 协作层”并存的开发者。我会给你一套可直接复制的config.toml与settings.json骨架,用 TaoToken 统一 Key 接入模型调用,再给出一套双协议联调的验证动作和预期输出。目标很明确:一次跑通工具调用与多智能体协作,而不是停留在概念图。
适合谁:写过 Python、装过 Node、本地能跑起一个 HTTP 服务的 Agent 开发者。如果你还没搭过任何 Agent,建议先跑通一个最小 Echo Agent 再回来,这篇的配置密度会比较高。
2. 前置准备:TaoToken 统一 Key 与本地环境
双协议联调最烦的不是协议本身,而是模型调用入口散落各处——搜索 Agent 用一个 Key,总结 Agent 用另一个,翻译 Agent 再配一套,调试时根本分不清是哪层出的错。我的做法是统一走 TaoToken 的 API 入口,一个 Key 覆盖所有 Agent 的模型调用,排障时只看一个地方。
2.1 拿 Key 与确认接入地址
先到 TaoToken 控制台创建 API Key。入口在这里:
控制台(创建与管理 Key):https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite
API Keys 管理页:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite
创建完把 Key 存到环境变量,别硬编码进代码:
export TAOTOKEN_API_KEY="sk-你的key"TaoToken 的 API 基地址是https://taotoken.net/api,兼容 OpenAI 风格的/v1/chat/completions调用。也就是说,你原来用 OpenAI SDK 写的代码,只需要改base_url和api_key两个字段就能切过来,Agent 里的模型调用逻辑几乎不用动。
2.2 本地依赖清单
三个 Agent 加一个 MCP Server,本地需要这些:
| 组件 | 版本建议 | 用途 |
|---|---|---|
| Python | 3.11+ | Agent 运行时 |
| Node.js | 20 LTS | 跑 npx 形式的 MCP Server |
| httpx | 0.27+ | 异步 HTTP 调用 |
| uvicorn | 0.30+ | A2A Server 启动 |
| a2a-sdk | 最新 | A2A 协议实现 |
装依赖:
pip install httpx uvicorn "a2a-sdk" python-dotenv node -v # 确认 >= 202.3 目录结构
我习惯把工具层和协作层在目录上就分开,排障时一眼能定位:
mcp-a2a-workflow/ ├── config.toml # MCP 工具声明 ├── settings.json # A2A Agent 注册与模型配置 ├── searcher/ # 搜索 Agent(A2A Server + MCP Client) │ ├── server.py │ └── task_manager.py ├── summarizer/ # 总结 Agent(纯 A2A) │ ├── server.py │ └── task_manager.py ├── translator/ # 翻译 Agent(纯 A2A) │ ├── server.py │ └── task_manager.py └── client.py # 联调入口搜索 Agent 是唯一同时踩两个协议的节点:对外它是 A2A Server,对内它是 MCP Client。这个“双身份”节点就是双协议联合的关键,后面配置会重点讲。
3. 可复制配置:config.toml 与 settings.json 骨架
配置分两份:config.toml声明 MCP 工具,settings.json声明 A2A Agent 和模型入口。分开写的好处是——工具换了不动协作配置,Agent 换了不动工具配置。
3.1 config.toml:MCP 工具声明
# config.toml —— MCP 工具层配置 [mcp] # 工具调用超时(秒),搜索类工具建议给足 timeout = 30 # 失败重试次数 retry = 2 # 声明一个搜索工具(用文件系统 MCP Server 做可离线 demo) [[mcp.servers]] name = "file-tools" transport = "stdio" command = "npx" args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/search-results"] enabled = true # 声明一个真实搜索工具(有 Key 时启用) [[mcp.servers]] name = "web-search" transport = "stdio" command = "npx" args = ["-y", "@modelcontextprotocol/server-brave-search"] enabled = false [mcp.servers.env] BRAVE_API_KEY = "${BRAVE_API_KEY}"这里有个坑先提醒:transport = "stdio"表示 MCP Server 以子进程方式启动,通过标准输入输出通信。如果你把enabled全设成false,搜索 Agent 启动时会找不到任何工具,报错信息通常很隐晦,表现为“工具列表为空”。至少保留一个 enabled = true 的 Server。
3.2 settings.json:A2A Agent 注册与模型入口
{ "model": { "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "default_model": "gpt-4o-mini", "timeout": 60 }, "agents": { "translator": { "name": "Translator Agent", "url": "http://localhost:10002", "role": "entry" }, "summarizer": { "name": "Summarizer Agent", "url": "http://localhost:10003", "role": "middle" }, "searcher": { "name": "Searcher Agent", "url": "http://localhost:10004", "role": "tool-user", "mcp_config": "config.toml" } }, "workflow": { "entry": "translator", "chain": ["translator", "summarizer", "searcher"] } }model.base_url指向 TaoToken 的 API 地址,api_key_env只写环境变量名,不写明文 Key。三个 Agent 共用这一份模型配置,谁调用模型都走同一个入口,出问题时排查范围直接缩小一半。
workflow.chain把协作链路显式写出来:翻译 Agent 是入口,往下委托给总结 Agent,总结 Agent 再委托给搜索 Agent。这条链就是 A2A 协作层的骨架。
3.3 搜索 Agent 的双身份初始化
搜索 Agent 启动时要同时做两件事:加载 MCP 工具、注册为 A2A Server。核心初始化代码:
import tomllib import json import os from pathlib import Path def load_mcp_config(path: str = "config.toml") -> dict: with open(path, "rb") as f: cfg = tomllib.load(f) servers = [s for s in cfg["mcp"]["servers"] if s.get("enabled")] if not servers: raise RuntimeError("没有启用的 MCP Server,检查 config.toml 的 enabled 字段") return {"servers": servers, "timeout": cfg["mcp"]["timeout"]} def load_settings(path: str = "settings.json") -> dict: with open(path, "r", encoding="utf-8") as f: return json.load(f) def build_model_client(settings: dict): import httpx api_key = os.environ.get(settings["model"]["api_key_env"]) if not api_key: raise RuntimeError("环境变量里没有找到 API Key") return httpx.AsyncClient( base_url=settings["model"]["base_url"], headers={"Authorization": f"Bearer {api_key}"}, timeout=settings["model"]["timeout"], )tomllib是 Python 3.11 内置的,不用额外装。这段代码做了三件事:读 MCP 配置、读 Agent 注册表、构造统一的模型客户端。注意base_url后面不带/v1,SDK 会自己拼路径,这是很多人第一次接入时踩的坑。
4. 双协议联调:验证动作与预期输出
配置写完不算跑通,得用一次真实请求把工具层和协作层都串起来。验证思路是:从入口 Agent 发一个关键词,看它能不能一路委托到搜索 Agent,搜索 Agent 能不能通过 MCP 拿到工具结果,再原路返回。
4.1 启动三个 Agent
开三个终端,分别启动:
# 终端 1:搜索 Agent(A2A Server + MCP Client) cd searcher && python server.py --port 10004 # 终端 2:总结 Agent(纯 A2A) cd summarizer && python server.py --port 10003 # 终端 3:翻译 Agent(纯 A2A,入口) cd translator && python server.py --port 10002每个终端看到Uvicorn running on http://localhost:1000x才算启动成功。如果某个端口被占用,先lsof -i :10004查一下。
4.2 验证 MCP 工具层是否就绪
在发完整请求前,先单独确认搜索 Agent 的 MCP 工具加载成功。访问它的 Agent Card:
curl http://localhost:10004/.well-known/agent.json预期返回里能看到skills字段包含search-skill,tags里有mcp。如果返回 404,说明 A2A Server 没起来;如果 skills 为空,说明 MCP 工具没加载,回去检查config.toml。
4.3 发起一次完整联调
用客户端脚本从入口 Agent 发关键词:
# client.py import asyncio import httpx from a2a.client import A2ACardResolver, ClientConfig, create_client from a2a.helpers import new_text_message from a2a.types.a2a_pb2 import Role, SendMessageRequest async def main(): async with httpx.AsyncClient(timeout=120.0) as http: resolver = A2ACardResolver( httpx_client=http, base_url="http://localhost:10002", ) card = await resolver.get_agent_card() print(f"入口 Agent: {card.name}") client = await create_client( agent=card, client_config=ClientConfig(streaming=False), ) keyword = "MCP 与 A2A 双协议协作" msg = new_text_message(keyword, role=Role.ROLE_USER) print(f"发送: {keyword}") print("链路: 翻译 -> 总结 -> 搜索(MCP)") print("=" * 50) async for chunk in client.send_message(SendMessageRequest(message=msg)): if getattr(chunk, "artifacts", None): for part in chunk.artifacts[0].parts: print(part.text) await client.close() if __name__ == "__main__": asyncio.run(main())运行python client.py,预期输出结构如下:
入口 Agent: Translator Agent 发送: MCP 与 A2A 双协议协作 链路: 翻译 -> 总结 -> 搜索(MCP) ================================================== 【中文总结】 1. MCP 负责 Agent 与工具的连接,A2A 负责 Agent 之间的协作... 2. 搜索 Agent 同时作为 A2A Server 和 MCP Client... 3. 双协议联合可覆盖工具调用与多智能体协作两层... 【英文翻译】 1. MCP handles the connection between agents and tools...看到中文总结和英文翻译两段都出来,说明协作层(A2A)和工具层(MCP)都通了。如果只有翻译没有总结,问题在 A2A 委托;如果总结内容里出现“搜索调用失败”,问题在 MCP 工具层。
4.4 分层验证技巧
联调失败时别急着改代码,先分层定位。我常用的两个探针:
# 探针 1:直接问搜索 Agent,绕过协作层 curl -X POST http://localhost:10004 \ -H "Content-Type: application/json" \ -d '{"jsonrpc":"2.0","id":"t1","method":"message/send","params":{"message":{"role":"user","parts":[{"type":"text","text":"测试关键词"}]}}}'这个请求直接打到搜索 Agent,如果它能返回结果,说明 MCP 工具层没问题,故障在 A2A 协作层;如果它也失败,故障在 MCP 层。先隔离协议层,再定位具体节点,比从头读日志快得多。
5. 本篇常见错排查
双协议联调的报错往往不直观,因为错误可能来自任意一层。下面是我实际踩过或见过的高频问题。
5.1 MCP Server 启动失败:npx 找不到包
现象:搜索 Agent 启动时报spawn npx ENOENT或工具列表为空。
原因通常是 Node 没装或 npx 不在 PATH 里。先确认:
which npx npx -y @modelcontextprotocol/server-filesystem /tmp/search-results如果第二条命令能跑起来并停在等待输入状态,说明 MCP Server 本身没问题,是 Agent 启动子进程时的环境变量没继承。解决办法是在config.toml里显式补 PATH,或者用绝对路径调用 npx。
5.2 A2A 委托超时:Agent Card 拿不到
现象:总结 Agent 调用搜索 Agent 时报搜索调用失败: ...,日志里是连接超时。
先手动验证 Agent Card 可达:
curl -s http://localhost:10004/.well-known/agent.json | head -c 200如果这条通,但 Agent 之间调用失败,多半是端口写错了。三个 Agent 的端口很容易记混,建议在settings.json里统一管理,代码里从配置读,别硬编码。
5.3 模型调用 401:Key 没生效
现象:Agent 能收到请求,但总结或翻译内容为空,日志里是 401。
检查环境变量是否在当前终端生效:
echo $TAOTOKEN_API_KEY如果为空,说明启动 Agent 的终端没export。注意export只在当前终端会话有效,换终端要重新设。更稳的做法是用python-dotenv从.env文件加载,避免每次手动设。
5.4 工具调用返回空:MCP 配置 enabled 全为 false
现象:搜索 Agent 返回“未找到相关结果”,但没报错。
这是最隐蔽的一类。config.toml里如果所有 Server 的enabled都是false,加载逻辑会过滤掉全部工具,Agent 拿不到任何工具却不会抛异常。排查时先打印加载后的工具列表:
cfg = load_mcp_config() print("已加载 MCP Server:", [s["name"] for s in cfg["servers"]])如果打印出来是空列表,回去把至少一个 Server 的enabled改成true。
5.5 协作链路断在中间:chain 配置与实际不符
现象:翻译 Agent 有输出,但内容里没有总结部分。
检查settings.json的workflow.chain是否和实际启动的 Agent 一致。如果 chain 里写了summarizer但总结 Agent 没启动,翻译 Agent 委托时会超时。启动顺序建议从链路末端往前:先起搜索 Agent,再起总结 Agent,最后起翻译 Agent,这样每个节点启动时下游都已就绪。
6. 继续往下走:把双协议用进真实项目
跑通这套 demo 后,你手里其实已经有了一个可扩展的骨架。工具层想加新能力,就在config.toml里加一个 MCP Server;协作层想加新角色,就在settings.json里注册一个 Agent 并挂进 chain。两层解耦,改一层不影响另一层。
如果你接下来要把它用到真实项目,几个方向可以优先考虑:
把模型调用统一收口到 TaoToken。三个 Agent 共用一份model配置,换模型只改一个字段。需要验证不同模型在总结、翻译任务上的表现时,可以直接在模型对话页面对比:
模型对话体验:https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=chat&utm_campaign=rewrite
给搜索 Agent 换成真实 MCP Server。把config.toml里web-search的enabled改成true,配上搜索 API Key,就能从模拟结果切到真实网页搜索。接入细节和参数说明可以查文档:
接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
长期跑编码类 Agent 的话,考虑用 Coding Plan 管理额度。多 Agent 并发调用模型时,按量计费容易失控,包月方案更可控:
Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite
最后说个我自己的经验:双协议联调最耗时间的不是写代码,而是定位故障在哪一层。所以我在项目里养成了一个习惯——每个 Agent 启动时都打印一行“我加载了哪些 MCP 工具 / 我注册在哪个端口”,联调时一眼就能看出是工具层没起来还是协作层断了。这个习惯帮我省下的调试时间,比任何优化都值。