企业官网建设流程全解析

1. 单Agent跑得挺好，为什么还要折腾Multi-Agent

如果你已经用单个Agent跑通过一些任务，大概率会遇到一个天花板：任务一复杂，它就开始顾此失彼。比如让它同时做「查资料 + 算数据 + 写报告」，它要么在查资料时忘了格式要求，要么在写报告时把中间算错的数据直接抄进去。这不是模型不行，而是单Agent的上下文里塞了太多互相干扰的目标。

Multi-Agent（多智能体）要解决的就是这件事：把一个大任务拆成几个专业角色，每个角色只关心自己那一摊，彼此通过一套约定好的消息格式交换结果。MCP协议在这里扮演的角色，就是这套「消息格式 + 通信机制」的落地规范。你可以把它理解成团队里的工单系统——谁发给谁、发什么类型、内容长什么样、怎么校验没被篡改，全都定死。

这篇文章面向的是已经写过单Agent、想往协作架构迁移的开发者。我会先讲清楚单Agent到Multi-Agent的架构跃迁路径，然后给出一份可复制的MCP配置文件骨架，接着用TaoToken的统一Key把多个Agent的模型调用接进来，最后给一套能跑通的通信链路验证动作。全程代码可复制，配置可改改就用。

需要提前说明：MCP在这里指的是多Agent之间的通信协议层，不是某个具体厂商的私有实现。我们关注的是消息结构、路由和校验这三件事，模型调用则统一走TaoToken的API，这样多个Agent不用各自维护一套Key。

2. 从单Agent到团队协作，架构上到底变了什么

2.1 单Agent的隐性瓶颈

单Agent的典型结构是：一个系统提示词 + 一堆工具 + 一个循环。任务简单时它很高效，但任务一复杂，问题就暴露了。

第一是上下文污染。搜索Agent返回的原始网页、分析Agent需要的中间数据、总结Agent要的结论，全挤在同一个上下文窗口里，模型很容易被无关信息带偏。第二是职责模糊。你没法给「搜索」和「分析」分别设定不同的温度、不同的工具集，因为它们本质上是同一个Agent。第三是容错差。搜索那一步失败了，整个链路就断了，没有别的角色能兜底。

我试过在一个单Agent里塞七八个工具，结果它经常在该调搜索的时候去调计算器，因为工具描述在长上下文里被稀释了。

2.2 Multi-Agent的跃迁路径

跃迁不是一步到位，建议分三步走。

第一步，角色拆分。把原来的单Agent按职责切成搜索、分析、总结三个角色，每个角色有独立的系统提示词和工具集。这一步不改通信方式，先让它们各自能独立跑通。

第二步，引入消息层。角色之间不再直接函数调用，而是通过统一的消息结构传递。这就是MCP协议要解决的问题：定义消息的字段、类型和校验方式。

第三步，加协调器。当角色多于三个、任务有依赖关系时，需要一个协调器来决定谁先跑、谁等谁、结果怎么合并。协调器本身也可以是一个Agent，但它只做调度，不做具体业务。

2.3 MCP协议在其中的位置

MCP协议不是替代Agent框架，而是补上「通信」这一层。它规定四件事：消息格式（JSON结构）、消息类型（任务分配、结果返回、错误上报）、路由规则（sender到recipient）、安全校验（签名防篡改）。

下面这张表把单Agent和Multi-Agent的关键差异列清楚，方便你判断自己该不该迁移。

3. TaoToken前置：一个Key管住所有Agent的模型调用

Multi-Agent落地时有个很现实的麻烦：三个Agent如果各自配一套模型Key，管理成本直接翻三倍，轮换、限额、审计都得做三遍。TaoToken的价值就在这里——它提供统一的API入口，多个Agent共用同一个Key，调用不同的模型。

3.1 获取Key与接入地址

先到TaoToken控制台创建API Key。地址是 https://taotoken.net/api-keys ，登录后新建一个Key，复制保存。注意这个Key只在创建时完整显示一次。

接入的基础地址是 https://taotoken.net/api ，所有Agent的模型请求都打到这个地址，通过model参数区分具体模型。这样搜索Agent可以用便宜快速的模型，分析Agent用推理强的模型，总结Agent用长文本模型，但Key只有一个。

3.2 环境变量配置

不要把Key硬编码进代码。用环境变量，本地开发和部署都统一。

# Linux / macOS export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" # Windows PowerShell $env:TAOTOKEN_API_KEY="sk-你的Key" $env:TAOTOKEN_BASE_URL="https://taotoken.net/api"

3.3 多Agent共用一个Key的调用封装

下面这段封装让每个Agent传入自己的model名，但共用同一个client。这样你换Key只需要改一个地方。

# core/llm_client.py import os from openai import OpenAI class LLMClient: """统一模型调用客户端，多Agent共用""" def __init__(self): self.client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) def chat(self, model: str, system: str, user: str) -> str: resp = self.client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system}, {"role": "user", "content": user}, ], temperature=0.3, ) return resp.choices[0].message.content llm = LLMClient()

注意：base_url末尾不要带斜杠，否则部分SDK会拼出双斜杠导致404。这是接入时最常见的低级错误。

4. 可复制的MCP配置文件骨架

MCP协议落地最怕的是「每个Agent自己定义消息格式」，最后互相看不懂。所以第一步是把协议配置抽成一个独立文件，所有Agent都读它。

4.1 mcp_config.yaml

# config/mcp_config.yaml protocol: version: "1.0" secret_key: "${MCP_SECRET_KEY}" # 从环境变量注入，不要写死 sign_algorithm: "HMAC-SHA256" message: required_fields: - sender_id - recipient_id - message_type - content - timestamp - version optional_fields: - signature - trace_id max_content_bytes: 65536 message_types: task_assign: "任务分配" task_result: "任务结果" error_report: "错误上报" heartbeat: "心跳" agents: - id: "agent_search" role: "搜索" model: "gpt-4o-mini" tools: ["web_search"] - id: "agent_analysis" role: "分析" model: "gpt-4o" tools: ["calculator"] - id: "agent_summary" role: "总结" model: "gpt-4o" tools: [] routing: default_timeout_ms: 30000 max_retry: 2 retry_backoff_ms: 500

这份配置定义了协议版本、消息必填字段、消息类型枚举、Agent清单和路由策略。改Agent只需要改agents段，不用动代码。

4.2 协议实现：消息创建与校验

# core/mcp_protocol.py import json import hmac import hashlib import time from typing import Dict, Any, Optional class MCPProtocol: def __init__(self, secret_key: str): self.secret_key = secret_key def create_message(self, sender_id: str, recipient_id: str, message_type: str, content: Dict[str, Any]) -> str: message = { "sender_id": sender_id, "recipient_id": recipient_id, "message_type": message_type, "content": content, "timestamp": int(time.time() * 1000), "version": "1.0", } message["signature"] = self._sign(message) return json.dumps(message, ensure_ascii=False) def parse_message(self, raw: str) -> Optional[Dict[str, Any]]: try: msg = json.loads(raw) except json.JSONDecodeError: return None if not self._verify(msg): return None return msg def _sign(self, message: Dict[str, Any]) -> str: payload = {k: v for k, v in message.items() if k != "signature"} raw = json.dumps(payload, sort_keys=True, ensure_ascii=False) return hmac.new( self.secret_key.encode(), raw.encode(), hashlib.sha256 ).hexdigest() def _verify(self, message: Dict[str, Any]) -> bool: signature = message.get("signature") if not signature: return False return hmac.compare_digest(signature, self._sign(message))

这里用hmac.compare_digest而不是==，是为了避免时序攻击。虽然内网通信风险低，但养成习惯没坏处。

4.3 协调器：任务分发与结果合并

# core/orchestrator.py import asyncio from core.mcp_protocol import MCPProtocol from core.llm_client import llm class Orchestrator: def __init__(self, protocol: MCPProtocol, agents: dict): self.protocol = protocol self.agents = agents # {agent_id: {"role":..., "model":...}} async def run_pipeline(self, task: str) -> dict: # 1. 搜索 search_out = await self._call("agent_search", task) # 2. 分析（带上搜索结果的摘要） analysis_out = await self._call("agent_analysis", search_out) # 3. 总结 summary_out = await self._call("agent_summary", analysis_out) return { "search": search_out, "analysis": analysis_out, "summary": summary_out, } async def _call(self, agent_id: str, payload: str) -> str: cfg = self.agents[agent_id] msg = self.protocol.create_message( sender_id="orchestrator", recipient_id=agent_id, message_type="task_assign", content={"task": payload}, ) parsed = self.protocol.parse_message(msg) if parsed is None: raise ValueError(f"消息校验失败: {agent_id}") # 实际调用模型 return await asyncio.to_thread( llm.chat, model=cfg["model"], system=f"你是{cfg['role']}Agent，只做{cfg['role']}相关的事。", user=parsed["content"]["task"], )

协调器只负责按顺序调用和传递，不掺和具体业务逻辑。这样以后加一个「审核Agent」，只需要在pipeline里插一步。

5. 验证请求：确认通信链路真的通了

配置写完不代表能跑。Multi-Agent最容易出问题的地方就是「消息发出去了但对面没收到」或者「收到了但校验失败」。所以要有明确的验证动作。

5.1 单条消息往返验证

先不接模型，只验证协议层。跑下面这段，确认消息能创建、能解析、签名能通过。

# tests/test_mcp_roundtrip.py import os from core.mcp_protocol import MCPProtocol def test_roundtrip(): proto = MCPProtocol(secret_key=os.environ["MCP_SECRET_KEY"]) raw = proto.create_message( sender_id="agent_search", recipient_id="agent_analysis", message_type="task_result", content={"result": "搜索到3条相关记录"}, ) parsed = proto.parse_message(raw) assert parsed is not None, "消息解析失败" assert parsed["sender_id"] == "agent_search" assert parsed["content"]["result"].startswith("搜索到") print("往返验证通过") if __name__ == "__main__": test_roundtrip()

运行python tests/test_mcp_roundtrip.py，看到「往返验证通过」说明协议层没问题。

5.2 篡改检测验证

再验证一下签名是否真的起作用。手动改一个字段，解析应该返回None。

# tests/test_mcp_tamper.py import json, os from core.mcp_protocol import MCPProtocol proto = MCPProtocol(secret_key=os.environ["MCP_SECRET_KEY"]) raw = proto.create_message("a", "b", "task_assign", {"task": "原始任务"}) msg = json.loads(raw) msg["content"]["task"] = "被篡改的任务" # 改内容但不改签名 tampered = json.dumps(msg, ensure_ascii=False) assert proto.parse_message(tampered) is None, "篡改未被检测到" print("篡改检测通过")

5.3 端到端链路验证

协议层通过后，跑完整pipeline。下面这段会真实调用TaoToken的API，确认三个Agent都能拿到模型返回。

# tests/test_pipeline.py import asyncio, os from core.mcp_protocol import MCPProtocol from core.orchestrator import Orchestrator AGENTS = { "agent_search": {"role": "搜索", "model": "gpt-4o-mini"}, "agent_analysis": {"role": "分析", "model": "gpt-4o"}, "agent_summary": {"role": "总结", "model": "gpt-4o"}, } async def main(): proto = MCPProtocol(secret_key=os.environ["MCP_SECRET_KEY"]) orch = Orchestrator(proto, AGENTS) result = await orch.run_pipeline("用三句话说明MCP协议的作用") for k, v in result.items(): print(f"[{k}] {v[:80]}...") if __name__ == "__main__": asyncio.run(main())

成功的话你会看到三段输出，分别来自搜索、分析、总结三个Agent。如果某一段报401，说明Key没配好；如果报消息校验失败，回去检查MCP_SECRET_KEY是否一致。

6. 本篇常见错排查

6.1 401 Unauthorized

最常见的原因是环境变量没生效。检查echo $TAOTOKEN_API_KEY是否有值。另一个原因是Key复制时带了空格，或者把sk-前缀漏了。还有一种情况是base_url写成了https://taotoken.net/api/（带尾斜杠），部分SDK会拼成//chat/completions导致路径错误。

6.2 消息校验一直失败

先确认创建消息和解析消息用的是同一个secret_key。如果Key从环境变量读，检查两个进程的环境变量是否一致。其次检查json.dumps是否用了sort_keys=True，签名和验签的序列化方式必须完全一致，否则哈希对不上。

6.3 Agent之间死循环

如果A等B的结果、B又等A的结果，pipeline会卡住。解决办法是在协调器里给每个_call加超时，超时后走降级逻辑（比如返回空结果并记录错误）。配置里的default_timeout_ms就是干这个的，但要在代码里真正用上。

# 在 _call 里加超时 try: return await asyncio.wait_for( asyncio.to_thread(llm.chat, ...), timeout=30, ) except asyncio.TimeoutError: return f"[{agent_id}] 超时，已降级"

6.4 模型返回被截断

Multi-Agent里每个Agent的输出会作为下一个Agent的输入，如果第一个Agent返回太长，会挤占后面的上下文。建议在消息content里加一个summary字段，只传摘要不传全文。或者在协调器里做一次截断，比如只取前2000字符。

6.5 并发调用触发限流

三个Agent如果并发调用同一个Key，可能触发速率限制。两个办法：一是串行调用（本文pipeline就是串行），二是加一个简单的令牌桶。串行对大多数场景够用，延迟换稳定。

7. 下一步：把协作体系跑起来

到这里，你已经有了协议配置、消息实现、协调器和验证脚本。接下来最实际的动作是：把mcp_config.yaml里的Agent清单改成你自己的角色，把AGENTS字典里的model换成你实际要用的模型，然后跑一遍test_pipeline.py。

如果你要长期跑编码类或Agent类任务，建议用Coding Plan来管理调用额度，地址是 https://taotoken.net/coding-plan 。它适合那种需要持续、稳定调用多个模型的场景，比按次计费更可控。

模型对话的调试入口在 https://taotoken.net/chat ，当你怀疑是模型本身的问题而不是协议问题时，可以先去那里单独测一下同一个prompt。接入文档在 https://taotoken.net/doc ，里面有各语言SDK的完整示例，遇到参数不确定时翻一下比猜快。

最后提醒一句：Multi-Agent的复杂度主要不在模型，而在通信和协调。先把两个Agent的往返跑通，再加第三个。一次加五个角色，调试成本会指数上升。