1. 从一次“Agent 各说各话”的翻车说起多 Agent 旅行规划助手听起来很美好一个 Agent 解析意图一个推荐目的地一个排行程一个算预算。但真动手写最先撞上的不是算法而是通信。我最早那版用函数直接互相调用结果意图 Agent 返回的字段名和行程 Agent 期望的对不上预算 Agent 拿到的是字符串却按数字算最后 Gradio 界面直接抛异常。问题根源在于Agent 之间没有统一的“信封格式”谁都能改字段谁都不负责校验。MCPModel Context Protocol解决的正是这件事。你可以把它理解成 Agent 世界的“快递面单标准”每个消息都有 sender、receiver、performative请求/通知、content 和 conversation_id所有 Agent 只认这个结构不关心对方内部怎么实现。再叠加 TaoToken 的统一 Key 通道Qwen 模型的调用也不用每个 Agent 各配一套密钥后端 FastAPI 和前端 Gradio 共用同一个入口。这篇要交付的东西很具体一份可复制的config.toml与settings.json骨架、Agent 注册与 MCP 工具调用配置、以及启动后怎么验证多 Agent 真的在协同而不是各跑各的。适合已经写过单 Agent、想往多 Agent 编排走一步的开发者。下面所有代码都可以直接落地不需要你再去猜字段。2. TaoToken 前置统一 Key 与 MCP 通道怎么接多 Agent 场景下最烦的是密钥散落。意图 Agent 要调 Qwen 做语义解析行程 Agent 要调 Qwen 生成每日安排如果每个 Agent 各自读环境变量配置会碎成一地。TaoToken 的做法是提供一个统一的 API 通道所有模型请求走同一个 base_url 和同一个 KeyAgent 侧只关心 prompt不关心鉴权。你需要先拿到 Key。进入控制台创建 API Key地址是https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite。创建后复制那串sk-开头的字符串后面写进config.toml。注意不要把它硬编码进 Git 仓库用环境变量或本地配置文件都行。模型侧我们选 Qwen 系列文本用qwen-turbo做意图解析和行程生成成本低、响应快。TaoToken 的 API 端点统一为https://taotoken.net/api兼容 OpenAI 风格的/v1/chat/completions所以你可以直接用openaiSDK 或requests调用不需要为 Qwen 单独装 DashScope SDK。这一点在多 Agent 里很关键所有 Agent 共用同一个 client 封装换模型只改一个字符串。如果你后面要做长期编码或 Agent 常驻任务可以了解 Coding Plan入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite。但本篇的旅行规划助手属于请求-响应式用按量 API Key 就够了。3. 可复制配置config.toml 与 settings.json 骨架先把配置文件落地。项目根目录建config.toml放模型通道和 MCP 总线参数# config.toml [taotoken] base_url https://taotoken.net/api api_key sk-你的Key写这里 default_model qwen-turbo timeout 30 max_retries 2 [mcp] bus_name travel_bus message_ttl 60 max_queue_size 256 [agents] # Agent 注册表name - 模块路径 intent agents.intent_agent:IntentAnalyzerAgent destination agents.destination_agent:DestinationAgent itinerary agents.itinerary_agent:ItineraryAgent budget agents.budget_agent:BudgetAgent central agents.central_agent:CentralTravelAgent [server] fastapi_host 0.0.0.0 fastapi_port 8000 gradio_port 7871再建settings.json放运行时可变参数和 MCP 工具声明。这个文件的作用是让 Agent 在启动时知道自己能调用哪些工具、参数 schema 是什么{ mcp_tools: [ { name: parse_intent, description: 解析用户旅行意图返回结构化字段, agent: intent, input_schema: { type: object, properties: { user_input: { type: string } }, required: [user_input] } }, { name: recommend_destination, description: 根据意图推荐目的地列表, agent: destination, input_schema: { type: object, properties: { intent: { type: object }, days: { type: integer } }, required: [intent] } }, { name: plan_itinerary, description: 生成每日行程安排, agent: itinerary, input_schema: { type: object, properties: { city: { type: string }, days: { type: integer }, interests: { type: array, items: { type: string } } }, required: [city, days] } }, { name: estimate_budget, description: 估算行程预算并检测冲突, agent: budget, input_schema: { type: object, properties: { itinerary: { type: object }, user_budget: { type: number } }, required: [itinerary] } } ], runtime: { default_days: 3, default_interests: [sightseeing, food], budget_currency: CNY } }这两个文件的分工要清楚config.toml管“连哪里、用什么模型”settings.json管“有哪些工具、参数长什么样”。Agent 启动时读settings.json注册工具运行时读config.toml建模型 client。这样你换模型只动 toml加工具只动 json互不干扰。4. Agent 注册与 MCP 工具调用配置配置有了接下来是代码侧怎么把 Agent 挂到 MCP 总线上。核心是一个MessageBus和一个Agent基类。MessageBus负责按 receiver 路由消息Agent基类负责收发和工具注册。先写总线与基类# mcp_protocol.py import uuid import queue import threading from dataclasses import dataclass, field from typing import Any, Dict, Callable, Optional dataclass class ACLMessage: sender: str receiver: str performative: str # request | inform content: Dict[str, Any] conversation_id: str field(default_factorylambda: str(uuid.uuid4())) class MessageBus: def __init__(self): self._queues: Dict[str, queue.Queue] {} self._lock threading.Lock() def register(self, name: str): with self._lock: if name not in self._queues: self._queues[name] queue.Queue(maxsize256) def send(self, msg: ACLMessage): with self._lock: q self._queues.get(msg.receiver) if q is None: raise KeyError(fAgent 未注册: {msg.receiver}) q.put(msg) def receive(self, name: str, timeout: float 1.0) - Optional[ACLMessage]: q self._queues.get(name) if q is None: return None try: return q.get(timeouttimeout) except queue.Empty: return None class Agent: def __init__(self, name: str, bus: MessageBus): self.name name self.bus bus self.bus.register(name) self._tools: Dict[str, Callable] {} def register_tool(self, tool_name: str, fn: Callable): self._tools[tool_name] fn def call_tool(self, tool_name: str, **kwargs) - Any: if tool_name not in self._tools: raise KeyError(f{self.name} 未注册工具: {tool_name}) return self._tools[tool_name](**kwargs) def send(self, receiver: str, performative: str, content: dict, conversation_id: str None): msg ACLMessage( senderself.name, receiverreceiver, performativeperformative, contentcontent, conversation_idconversation_id or str(uuid.uuid4()), ) self.bus.send(msg) return msg.conversation_id def run_once(self, timeout: float 1.0): msg self.bus.receive(self.name, timeouttimeout) if msg is not None: self.handle(msg) return msg def handle(self, msg: ACLMessage): raise NotImplementedError然后是模型 client 封装所有 Agent 共用# llm_client.py import requests from typing import Optional class TaoTokenClient: def __init__(self, base_url: str, api_key: str, model: str qwen-turbo, timeout: int 30): self.base_url base_url.rstrip(/) self.api_key api_key self.model model self.timeout timeout def chat(self, prompt: str, system: Optional[str] None, temperature: float 0.6) - str: messages [] if system: messages.append({role: system, content: system}) messages.append({role: user, content: prompt}) resp requests.post( f{self.base_url}/v1/chat/completions, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, json{ model: self.model, messages: messages, temperature: temperature, }, timeoutself.timeout, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content].strip()接着是意图 Agent 的注册示例它把parse_intent工具挂到自己身上# agents/intent_agent.py import json from mcp_protocol import Agent, ACLMessage from llm_client import TaoTokenClient class IntentAnalyzerAgent(Agent): def __init__(self, name, bus, llm: TaoTokenClient): super().__init__(name, bus) self.llm llm self.register_tool(parse_intent, self.parse_intent) def parse_intent(self, user_input: str) - dict: system ( 你是旅行意图解析器。只输出 JSON字段 destination_type, days, budget, companions, interests。 未提及的字段填 unknown。 ) raw self.llm.chat(f用户需求{user_input}, systemsystem) start, end raw.find({), raw.rfind(}) 1 if start -1 or end 0: return {destination_type: unknown, days: 3, budget: unknown, companions: unknown, interests: [sightseeing]} try: return json.loads(raw[start:end]) except json.JSONDecodeError: return {destination_type: unknown, days: 3, budget: unknown, companions: unknown, interests: [sightseeing]} def handle(self, msg: ACLMessage): if msg.performative request: intent self.call_tool(parse_intent, user_inputmsg.content[user_input]) self.send( receivermsg.sender, performativeinform, content{intent: intent}, conversation_idmsg.conversation_id, )目的地 Agent 类似注册recommend_destination内部可以先用规则打分再让 Qwen 补充理由。行程 Agent 注册plan_itinerary把 Qwen 返回的文本按天切分。预算 Agent 注册estimate_budget做数字校验和冲突检测。中央协调 Agent 不注册工具它只负责按顺序发 request、收 inform、拼最终结果。启动装配代码# main.py import toml import json from mcp_protocol import MessageBus from llm_client import TaoTokenClient from agents.intent_agent import IntentAnalyzerAgent from agents.destination_agent import DestinationAgent from agents.itinerary_agent import ItineraryAgent from agents.budget_agent import BudgetAgent from agents.central_agent import CentralTravelAgent cfg toml.load(config.toml) settings json.load(open(settings.json, encodingutf-8)) bus MessageBus() llm TaoTokenClient( base_urlcfg[taotoken][base_url], api_keycfg[taotoken][api_key], modelcfg[taotoken][default_model], timeoutcfg[taotoken][timeout], ) agents { intent: IntentAnalyzerAgent(intent, bus, llm), destination: DestinationAgent(destination, bus, llm), itinerary: ItineraryAgent(itinerary, bus, llm), budget: BudgetAgent(budget, bus, llm), central: CentralTravelAgent(central, bus), }到这里Agent 注册和工具调用就串起来了。每个 Agent 只认 MCP 消息工具通过settings.json声明模型通过 TaoToken 统一通道调用。5. 验证请求启动后怎么确认多 Agent 真的在协同配置写完不代表跑通。你需要一个可观测的验证动作。最直接的方式是给 FastAPI 加一个/plan端点然后在中央 Agent 里打印每一步的消息流转。FastAPI 端点# api.py from fastapi import FastAPI from pydantic import BaseModel from main import bus, agents app FastAPI(titleMCP 多 Agent 旅行规划) class PlanRequest(BaseModel): message: str user_id: str anonymous app.post(/plan) def plan(req: PlanRequest): central agents[central] cid central.start_plan(req.message) result central.wait_result(cid, timeout60) return {success: True, conversation_id: cid, data: result}中央 Agent 的start_plan和wait_result核心逻辑# agents/central_agent.py import time from mcp_protocol import Agent, ACLMessage class CentralTravelAgent(Agent): def __init__(self, name, bus): super().__init__(name, bus) self.sessions {} def start_plan(self, user_input: str) - str: cid self.send( receiverintent, performativerequest, content{user_input: user_input}, ) self.sessions[cid] {user_input: user_input, steps: [], result: None} return cid def handle(self, msg: ACLMessage): cid msg.conversation_id if cid not in self.sessions: return sess self.sessions[cid] sess[steps].append(msg.sender) if msg.sender intent: sess[intent] msg.content[intent] self.send(destination, request, {intent: sess[intent], days: sess[intent].get(days, 3)}, cid) elif msg.sender destination: sess[destinations] msg.content[destinations] city sess[destinations][0][city] sess[city] city self.send(itinerary, request, {city: city, days: sess[intent].get(days, 3), interests: sess[intent].get(interests, [sightseeing])}, cid) elif msg.sender itinerary: sess[itinerary] msg.content[itinerary] self.send(budget, request, {itinerary: sess[itinerary], user_budget: sess[intent].get(budget, unknown)}, cid) elif msg.sender budget: sess[budget] msg.content[budget] sess[result] { user_input: sess[user_input], intent: sess[intent], destinations: sess[destinations], chosen_city: sess[city], itinerary: sess[itinerary], budget: sess[budget], agent_chain: sess[steps], } def wait_result(self, cid: str, timeout: float 60): start time.time() while time.time() - start timeout: if self.sessions[cid][result] is not None: return self.sessions[cid][result] self.run_once(timeout0.5) return {error: timeout, steps: self.sessions[cid][steps]}启动服务uvicorn api:app --host 0.0.0.0 --port 8000发一个真实请求curl -X POST http://localhost:8000/plan \ -H Content-Type: application/json \ -d {message:带父母7月去海边度假预算15000元玩5天,user_id:test_01}成功时你会看到agent_chain字段是[intent, destination, itinerary, budget]这说明四个 Agent 按顺序被触发消息通过 MCP 总线流转而不是某个函数一把梭。如果agent_chain只有[intent]说明目的地 Agent 没收到消息去查settings.json里recommend_destination的 agent 名是否和注册名一致。Gradio 前端只需要调这个/plan端点把itinerary和budget渲染成文本。前端不参与 Agent 通信它只是触发器和展示层。这样后端 FastAPI 负责编排前端 Gradio 负责交互职责清晰。6. 本篇常见错排查错误一KeyError: Agent 未注册。原因是MessageBus.register没被调用或者 Agent 初始化顺序不对。检查main.py里是否所有 Agent 都在bus创建之后实例化。Agent.__init__里会自动调register但如果你手动 new 了 Agent 却没传 bus就会漏注册。错误二Qwen 返回的不是 JSON意图解析直接崩。模型有时会加“好的以下是解析结果”这种前缀。代码里用raw.find({)和raw.rfind(})截取是必要的但更稳的做法是在 system prompt 里强调“只输出 JSON不要任何解释”。如果还是不稳定把temperature降到 0.2。错误三conversation_id对不上中央 Agent 收不到后续消息。每次send如果不传conversation_id会生成新的 UUID导致会话断裂。正确做法是中央 Agent 在start_plan里拿到cid后后续所有send都显式传这个cid。上面代码里已经这么做了你抄的时候别漏。错误四TaoToken 请求 401。先确认config.toml里的api_key是完整的sk-字符串没有多余空格。再确认base_url是https://taotoken.net/api不要在后面多加/v1因为 client 里已经拼了/v1/chat/completions。如果还不行去控制台重新生成一个 Key 试试。错误五Gradio 端口冲突。默认 7871 如果被占用改config.toml里的gradio_port。FastAPI 的 8000 同理。两个服务分开跑不要试图塞进同一个进程。错误六预算 Agent 拿到的是字符串却做加法。Qwen 返回的预算可能是15000元这种带单位的字符串。在estimate_budget里先用正则提取数字再做计算。别直接int()会抛ValueError。7. 下一步把验证动作变成日常跑通之后建议你把agent_chain打印到日志里每次请求都看一眼。多 Agent 系统最怕的是“看起来返回了结果其实中间某个 Agent 被跳过了”。有了这条链谁没参与一目了然。如果你要接更多 Agent比如天气 Agent、签证 Agent只需要在settings.json里加一条工具声明在main.py里注册新 Agent中央 Agent 的handle里加一个分支。MCP 总线和 TaoToken 通道都不用动。这就是统一协议的价值扩展的是 Agent不是通信层。模型对话调试可以用https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite快速验证 Qwen 的返回格式。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有完整的端点说明和参数列表。API Key 管理还是那个地址https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite。
基于MCP的多Agent旅行规划助手完整实现：TaoToken统一Key接入FastAPI与Gradio
1. 从一次“Agent 各说各话”的翻车说起多 Agent 旅行规划助手听起来很美好一个 Agent 解析意图一个推荐目的地一个排行程一个算预算。但真动手写最先撞上的不是算法而是通信。我最早那版用函数直接互相调用结果意图 Agent 返回的字段名和行程 Agent 期望的对不上预算 Agent 拿到的是字符串却按数字算最后 Gradio 界面直接抛异常。问题根源在于Agent 之间没有统一的“信封格式”谁都能改字段谁都不负责校验。MCPModel Context Protocol解决的正是这件事。你可以把它理解成 Agent 世界的“快递面单标准”每个消息都有 sender、receiver、performative请求/通知、content 和 conversation_id所有 Agent 只认这个结构不关心对方内部怎么实现。再叠加 TaoToken 的统一 Key 通道Qwen 模型的调用也不用每个 Agent 各配一套密钥后端 FastAPI 和前端 Gradio 共用同一个入口。这篇要交付的东西很具体一份可复制的config.toml与settings.json骨架、Agent 注册与 MCP 工具调用配置、以及启动后怎么验证多 Agent 真的在协同而不是各跑各的。适合已经写过单 Agent、想往多 Agent 编排走一步的开发者。下面所有代码都可以直接落地不需要你再去猜字段。2. TaoToken 前置统一 Key 与 MCP 通道怎么接多 Agent 场景下最烦的是密钥散落。意图 Agent 要调 Qwen 做语义解析行程 Agent 要调 Qwen 生成每日安排如果每个 Agent 各自读环境变量配置会碎成一地。TaoToken 的做法是提供一个统一的 API 通道所有模型请求走同一个 base_url 和同一个 KeyAgent 侧只关心 prompt不关心鉴权。你需要先拿到 Key。进入控制台创建 API Key地址是https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite。创建后复制那串sk-开头的字符串后面写进config.toml。注意不要把它硬编码进 Git 仓库用环境变量或本地配置文件都行。模型侧我们选 Qwen 系列文本用qwen-turbo做意图解析和行程生成成本低、响应快。TaoToken 的 API 端点统一为https://taotoken.net/api兼容 OpenAI 风格的/v1/chat/completions所以你可以直接用openaiSDK 或requests调用不需要为 Qwen 单独装 DashScope SDK。这一点在多 Agent 里很关键所有 Agent 共用同一个 client 封装换模型只改一个字符串。如果你后面要做长期编码或 Agent 常驻任务可以了解 Coding Plan入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite。但本篇的旅行规划助手属于请求-响应式用按量 API Key 就够了。3. 可复制配置config.toml 与 settings.json 骨架先把配置文件落地。项目根目录建config.toml放模型通道和 MCP 总线参数# config.toml [taotoken] base_url https://taotoken.net/api api_key sk-你的Key写这里 default_model qwen-turbo timeout 30 max_retries 2 [mcp] bus_name travel_bus message_ttl 60 max_queue_size 256 [agents] # Agent 注册表name - 模块路径 intent agents.intent_agent:IntentAnalyzerAgent destination agents.destination_agent:DestinationAgent itinerary agents.itinerary_agent:ItineraryAgent budget agents.budget_agent:BudgetAgent central agents.central_agent:CentralTravelAgent [server] fastapi_host 0.0.0.0 fastapi_port 8000 gradio_port 7871再建settings.json放运行时可变参数和 MCP 工具声明。这个文件的作用是让 Agent 在启动时知道自己能调用哪些工具、参数 schema 是什么{ mcp_tools: [ { name: parse_intent, description: 解析用户旅行意图返回结构化字段, agent: intent, input_schema: { type: object, properties: { user_input: { type: string } }, required: [user_input] } }, { name: recommend_destination, description: 根据意图推荐目的地列表, agent: destination, input_schema: { type: object, properties: { intent: { type: object }, days: { type: integer } }, required: [intent] } }, { name: plan_itinerary, description: 生成每日行程安排, agent: itinerary, input_schema: { type: object, properties: { city: { type: string }, days: { type: integer }, interests: { type: array, items: { type: string } } }, required: [city, days] } }, { name: estimate_budget, description: 估算行程预算并检测冲突, agent: budget, input_schema: { type: object, properties: { itinerary: { type: object }, user_budget: { type: number } }, required: [itinerary] } } ], runtime: { default_days: 3, default_interests: [sightseeing, food], budget_currency: CNY } }这两个文件的分工要清楚config.toml管“连哪里、用什么模型”settings.json管“有哪些工具、参数长什么样”。Agent 启动时读settings.json注册工具运行时读config.toml建模型 client。这样你换模型只动 toml加工具只动 json互不干扰。4. Agent 注册与 MCP 工具调用配置配置有了接下来是代码侧怎么把 Agent 挂到 MCP 总线上。核心是一个MessageBus和一个Agent基类。MessageBus负责按 receiver 路由消息Agent基类负责收发和工具注册。先写总线与基类# mcp_protocol.py import uuid import queue import threading from dataclasses import dataclass, field from typing import Any, Dict, Callable, Optional dataclass class ACLMessage: sender: str receiver: str performative: str # request | inform content: Dict[str, Any] conversation_id: str field(default_factorylambda: str(uuid.uuid4())) class MessageBus: def __init__(self): self._queues: Dict[str, queue.Queue] {} self._lock threading.Lock() def register(self, name: str): with self._lock: if name not in self._queues: self._queues[name] queue.Queue(maxsize256) def send(self, msg: ACLMessage): with self._lock: q self._queues.get(msg.receiver) if q is None: raise KeyError(fAgent 未注册: {msg.receiver}) q.put(msg) def receive(self, name: str, timeout: float 1.0) - Optional[ACLMessage]: q self._queues.get(name) if q is None: return None try: return q.get(timeouttimeout) except queue.Empty: return None class Agent: def __init__(self, name: str, bus: MessageBus): self.name name self.bus bus self.bus.register(name) self._tools: Dict[str, Callable] {} def register_tool(self, tool_name: str, fn: Callable): self._tools[tool_name] fn def call_tool(self, tool_name: str, **kwargs) - Any: if tool_name not in self._tools: raise KeyError(f{self.name} 未注册工具: {tool_name}) return self._tools[tool_name](**kwargs) def send(self, receiver: str, performative: str, content: dict, conversation_id: str None): msg ACLMessage( senderself.name, receiverreceiver, performativeperformative, contentcontent, conversation_idconversation_id or str(uuid.uuid4()), ) self.bus.send(msg) return msg.conversation_id def run_once(self, timeout: float 1.0): msg self.bus.receive(self.name, timeouttimeout) if msg is not None: self.handle(msg) return msg def handle(self, msg: ACLMessage): raise NotImplementedError然后是模型 client 封装所有 Agent 共用# llm_client.py import requests from typing import Optional class TaoTokenClient: def __init__(self, base_url: str, api_key: str, model: str qwen-turbo, timeout: int 30): self.base_url base_url.rstrip(/) self.api_key api_key self.model model self.timeout timeout def chat(self, prompt: str, system: Optional[str] None, temperature: float 0.6) - str: messages [] if system: messages.append({role: system, content: system}) messages.append({role: user, content: prompt}) resp requests.post( f{self.base_url}/v1/chat/completions, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, json{ model: self.model, messages: messages, temperature: temperature, }, timeoutself.timeout, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content].strip()接着是意图 Agent 的注册示例它把parse_intent工具挂到自己身上# agents/intent_agent.py import json from mcp_protocol import Agent, ACLMessage from llm_client import TaoTokenClient class IntentAnalyzerAgent(Agent): def __init__(self, name, bus, llm: TaoTokenClient): super().__init__(name, bus) self.llm llm self.register_tool(parse_intent, self.parse_intent) def parse_intent(self, user_input: str) - dict: system ( 你是旅行意图解析器。只输出 JSON字段 destination_type, days, budget, companions, interests。 未提及的字段填 unknown。 ) raw self.llm.chat(f用户需求{user_input}, systemsystem) start, end raw.find({), raw.rfind(}) 1 if start -1 or end 0: return {destination_type: unknown, days: 3, budget: unknown, companions: unknown, interests: [sightseeing]} try: return json.loads(raw[start:end]) except json.JSONDecodeError: return {destination_type: unknown, days: 3, budget: unknown, companions: unknown, interests: [sightseeing]} def handle(self, msg: ACLMessage): if msg.performative request: intent self.call_tool(parse_intent, user_inputmsg.content[user_input]) self.send( receivermsg.sender, performativeinform, content{intent: intent}, conversation_idmsg.conversation_id, )目的地 Agent 类似注册recommend_destination内部可以先用规则打分再让 Qwen 补充理由。行程 Agent 注册plan_itinerary把 Qwen 返回的文本按天切分。预算 Agent 注册estimate_budget做数字校验和冲突检测。中央协调 Agent 不注册工具它只负责按顺序发 request、收 inform、拼最终结果。启动装配代码# main.py import toml import json from mcp_protocol import MessageBus from llm_client import TaoTokenClient from agents.intent_agent import IntentAnalyzerAgent from agents.destination_agent import DestinationAgent from agents.itinerary_agent import ItineraryAgent from agents.budget_agent import BudgetAgent from agents.central_agent import CentralTravelAgent cfg toml.load(config.toml) settings json.load(open(settings.json, encodingutf-8)) bus MessageBus() llm TaoTokenClient( base_urlcfg[taotoken][base_url], api_keycfg[taotoken][api_key], modelcfg[taotoken][default_model], timeoutcfg[taotoken][timeout], ) agents { intent: IntentAnalyzerAgent(intent, bus, llm), destination: DestinationAgent(destination, bus, llm), itinerary: ItineraryAgent(itinerary, bus, llm), budget: BudgetAgent(budget, bus, llm), central: CentralTravelAgent(central, bus), }到这里Agent 注册和工具调用就串起来了。每个 Agent 只认 MCP 消息工具通过settings.json声明模型通过 TaoToken 统一通道调用。5. 验证请求启动后怎么确认多 Agent 真的在协同配置写完不代表跑通。你需要一个可观测的验证动作。最直接的方式是给 FastAPI 加一个/plan端点然后在中央 Agent 里打印每一步的消息流转。FastAPI 端点# api.py from fastapi import FastAPI from pydantic import BaseModel from main import bus, agents app FastAPI(titleMCP 多 Agent 旅行规划) class PlanRequest(BaseModel): message: str user_id: str anonymous app.post(/plan) def plan(req: PlanRequest): central agents[central] cid central.start_plan(req.message) result central.wait_result(cid, timeout60) return {success: True, conversation_id: cid, data: result}中央 Agent 的start_plan和wait_result核心逻辑# agents/central_agent.py import time from mcp_protocol import Agent, ACLMessage class CentralTravelAgent(Agent): def __init__(self, name, bus): super().__init__(name, bus) self.sessions {} def start_plan(self, user_input: str) - str: cid self.send( receiverintent, performativerequest, content{user_input: user_input}, ) self.sessions[cid] {user_input: user_input, steps: [], result: None} return cid def handle(self, msg: ACLMessage): cid msg.conversation_id if cid not in self.sessions: return sess self.sessions[cid] sess[steps].append(msg.sender) if msg.sender intent: sess[intent] msg.content[intent] self.send(destination, request, {intent: sess[intent], days: sess[intent].get(days, 3)}, cid) elif msg.sender destination: sess[destinations] msg.content[destinations] city sess[destinations][0][city] sess[city] city self.send(itinerary, request, {city: city, days: sess[intent].get(days, 3), interests: sess[intent].get(interests, [sightseeing])}, cid) elif msg.sender itinerary: sess[itinerary] msg.content[itinerary] self.send(budget, request, {itinerary: sess[itinerary], user_budget: sess[intent].get(budget, unknown)}, cid) elif msg.sender budget: sess[budget] msg.content[budget] sess[result] { user_input: sess[user_input], intent: sess[intent], destinations: sess[destinations], chosen_city: sess[city], itinerary: sess[itinerary], budget: sess[budget], agent_chain: sess[steps], } def wait_result(self, cid: str, timeout: float 60): start time.time() while time.time() - start timeout: if self.sessions[cid][result] is not None: return self.sessions[cid][result] self.run_once(timeout0.5) return {error: timeout, steps: self.sessions[cid][steps]}启动服务uvicorn api:app --host 0.0.0.0 --port 8000发一个真实请求curl -X POST http://localhost:8000/plan \ -H Content-Type: application/json \ -d {message:带父母7月去海边度假预算15000元玩5天,user_id:test_01}成功时你会看到agent_chain字段是[intent, destination, itinerary, budget]这说明四个 Agent 按顺序被触发消息通过 MCP 总线流转而不是某个函数一把梭。如果agent_chain只有[intent]说明目的地 Agent 没收到消息去查settings.json里recommend_destination的 agent 名是否和注册名一致。Gradio 前端只需要调这个/plan端点把itinerary和budget渲染成文本。前端不参与 Agent 通信它只是触发器和展示层。这样后端 FastAPI 负责编排前端 Gradio 负责交互职责清晰。6. 本篇常见错排查错误一KeyError: Agent 未注册。原因是MessageBus.register没被调用或者 Agent 初始化顺序不对。检查main.py里是否所有 Agent 都在bus创建之后实例化。Agent.__init__里会自动调register但如果你手动 new 了 Agent 却没传 bus就会漏注册。错误二Qwen 返回的不是 JSON意图解析直接崩。模型有时会加“好的以下是解析结果”这种前缀。代码里用raw.find({)和raw.rfind(})截取是必要的但更稳的做法是在 system prompt 里强调“只输出 JSON不要任何解释”。如果还是不稳定把temperature降到 0.2。错误三conversation_id对不上中央 Agent 收不到后续消息。每次send如果不传conversation_id会生成新的 UUID导致会话断裂。正确做法是中央 Agent 在start_plan里拿到cid后后续所有send都显式传这个cid。上面代码里已经这么做了你抄的时候别漏。错误四TaoToken 请求 401。先确认config.toml里的api_key是完整的sk-字符串没有多余空格。再确认base_url是https://taotoken.net/api不要在后面多加/v1因为 client 里已经拼了/v1/chat/completions。如果还不行去控制台重新生成一个 Key 试试。错误五Gradio 端口冲突。默认 7871 如果被占用改config.toml里的gradio_port。FastAPI 的 8000 同理。两个服务分开跑不要试图塞进同一个进程。错误六预算 Agent 拿到的是字符串却做加法。Qwen 返回的预算可能是15000元这种带单位的字符串。在estimate_budget里先用正则提取数字再做计算。别直接int()会抛ValueError。7. 下一步把验证动作变成日常跑通之后建议你把agent_chain打印到日志里每次请求都看一眼。多 Agent 系统最怕的是“看起来返回了结果其实中间某个 Agent 被跳过了”。有了这条链谁没参与一目了然。如果你要接更多 Agent比如天气 Agent、签证 Agent只需要在settings.json里加一条工具声明在main.py里注册新 Agent中央 Agent 的handle里加一个分支。MCP 总线和 TaoToken 通道都不用动。这就是统一协议的价值扩展的是 Agent不是通信层。模型对话调试可以用https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite快速验证 Qwen 的返回格式。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有完整的端点说明和参数列表。API Key 管理还是那个地址https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite。