ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

MCP协议+多Agent协作:从零搭建企业级AI Agent系统的实战指南(TaoToken统一Key接入版)

MCP协议+多Agent协作:从零搭建企业级AI Agent系统的实战指南(TaoToken统一Key接入版) 1. 为什么单个 Agent 跑不通企业级任务如果你已经写过一个能查数据库、能读文件的 AI Agent大概率会遇到这样的瓶颈让它同时做代码审查、写测试、查文档、发通知它就开始胡言乱语工具调用顺序混乱上下文越滚越大最后直接超窗。这不是模型不行而是单 Agent 架构本身撑不起复杂业务链路。MCPModel Context Protocol解决的就是这个问题。你可以把它理解成 AI Agent 世界的 USB 接口以前每接一个工具都要写一套私有适配层现在只要工具端实现 MCP Server任何支持 MCP 的 Client 都能即插即用。而多 Agent 协作则是把 Planner、Coder、Reviewer、Executor 这些角色拆成独立的 MCP Server由一个 Orchestrator 统一调度。这套架构落地时最容易被忽略的一环是模型接入层。每个 Agent 都要调 LLM如果每个 Agent 各自维护一套 Key、各自处理限流和计费运维成本会指数级上升。这篇就按「MCP 打通协作 TaoToken 统一 Key 接入」的路线从零搭一个最小可用的企业级多 Agent 系统交付可直接复制的settings.json与config.toml骨架、多 Agent 注册路由示例以及连通性验证和报错排查动作。适合谁看已经写过单 Agent Demo、想升级到多 Agent 协作的开发者正在做企业内部 AI 工具链、需要统一模型接入通道的工程团队以及被各家模型 Key 管理搞到头大的运维同学。2. TaoToken 前置统一 Key 与 API 通道在动手写 Agent 之前先把模型接入层固定下来。多 Agent 系统里每个 Agent 都要调模型如果 Planner 用一家、Coder 用另一家、Reviewer 又换一家Key 管理、额度监控、故障切换会变成噩梦。TaoToken 的作用就是把这些收敛成一个统一入口。它的核心价值有三点一是统一 Key所有 Agent 共用一套凭证不用在每个 Agent 里散落配置二是统一 API 通道兼容主流模型调用格式Agent 侧代码不用为每家模型写适配三是统一计费与限流视图方便你在多 Agent 并发时观察谁在消耗额度。你需要先拿到两样东西API Key 和接入地址。Key 在控制台的 API Keys 页面创建地址用https://taotoken.net/api注意 API 调用不加 UTM 参数。控制台入口在这里控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite API Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite创建 Key 的时候建议按 Agent 角色分几个 Key比如planner-key、coder-key、reviewer-key。虽然 TaoToken 支持一个 Key 走全部但分 Key 的好处是你能在控制台按角色看消耗某个 Agent 跑飞了也能单独吊销不影响其他 Agent。这一步花五分钟后面排障能省几小时。拿到 Key 之后先别急着写 Agent用一条 curl 验证通道是否通curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: claude-sonnet-4-20250514, messages: [{role: user, content: ping}], max_tokens: 16 }返回里有choices字段就说明通道正常。这一步不通后面所有 Agent 都白搭所以务必先过。如果返回 401检查 Key 是否复制完整返回 404检查地址是不是写成了带路径的完整 URL返回 429说明触发了限流去控制台看额度。3. 可复制配置settings.json 与 config.toml 骨架多 Agent 系统的配置分两层一层是 MCP Client 侧的 Server 注册用settings.json一层是模型接入与 Agent 角色定义用config.toml。分开的原因是这样职责清晰——settings.json管「有哪些 Agent 可用」config.toml管「每个 Agent 怎么调模型」。3.1 settings.jsonMCP Server 注册骨架这个文件放在你的 Orchestrator 项目根目录作用是告诉 MCP Client 去哪里拉起各个 Agent Server。每个 Agent 本质是一个独立的 MCP Server 进程通过 stdio 通信。{ mcpServers: { planner-agent: { command: python, args: [agents/planner_server.py], env: { TAOTOKEN_API_KEY: ${TAOTOKEN_PLANNER_KEY}, TAOTOKEN_BASE_URL: https://taotoken.net/api } }, coder-agent: { command: python, args: [agents/coder_server.py], env: { TAOTOKEN_API_KEY: ${TAOTOKEN_CODER_KEY}, TAOTOKEN_BASE_URL: https://taotoken.net/api } }, reviewer-agent: { command: python, args: [agents/reviewer_server.py], env: { TAOTOKEN_API_KEY: ${TAOTOKEN_REVIEWER_KEY}, TAOTOKEN_BASE_URL: https://taotoken.net/api } }, executor-agent: { command: python, args: [agents/executor_server.py], env: { TAOTOKEN_API_KEY: ${TAOTOKEN_EXECUTOR_KEY}, TAOTOKEN_BASE_URL: https://taotoken.net/api } } } }注意env里用的是环境变量引用而不是硬编码 Key。这样settings.json可以进版本库Key 通过.env或 CI 注入避免泄露。每个 Agent 用独立 Key方便按角色监控。3.2 config.toml模型接入与 Agent 角色定义config.toml管的是每个 Agent 用哪个模型、超时多少、重试几次、上下文窗口多大。这是多 Agent 系统里最值得花时间调的部分因为不同角色对模型能力的要求完全不同。[gateway] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY default_timeout 60 max_retries 3 retry_backoff 1.5 [agents.planner] model claude-sonnet-4-20250514 temperature 0.2 max_tokens 4096 context_window 32000 system_prompt 你是任务规划专家负责把复杂需求拆解成可执行的子任务列表。 [agents.coder] model claude-sonnet-4-20250514 temperature 0.1 max_tokens 8192 context_window 64000 system_prompt 你是资深工程师根据任务描述生成可运行代码只输出代码和必要注释。 [agents.reviewer] model claude-sonnet-4-20250514 temperature 0.0 max_tokens 4096 context_window 32000 system_prompt 你是代码审查专家找出逻辑错误、边界问题和安全隐患按严重程度排序。 [agents.executor] model claude-sonnet-4-20250514 temperature 0.0 max_tokens 2048 context_window 16000 system_prompt 你是执行器根据指令调用工具并返回结构化结果。 [routing] # 任务类型到 Agent 的映射 code_review [planner, coder, reviewer] data_query [planner, executor] full_pipeline [planner, coder, reviewer, executor]temperature的设置很关键Planner 需要一点创造性所以给 0.2Coder 要稳定给 0.1Reviewer 和 Executor 要确定性给 0.0。context_window按角色分配Coder 处理大文件所以给最大Executor 只做工具调用给最小。这些参数不是拍脑袋是实测下来能明显降低「Agent 跑偏」概率的配置。3.3 Agent 基类把每个角色封装成 MCP Server有了配置接下来写 Agent 基类。核心思路是每个 Agent 继承同一个基类基类负责从config.toml读自己的配置、初始化模型客户端、注册 MCP 工具。import os import tomllib import httpx from mcp.server import Server from mcp.types import Tool, TextContent class BaseAgent: def __init__(self, role: str, config_path: str config.toml): with open(config_path, rb) as f: self.config tomllib.load(f) self.role role self.agent_cfg self.config[agents][role] self.gateway self.config[gateway] self.server Server(f{role}-agent) self.client httpx.AsyncClient( base_urlself.gateway[base_url], headers{Authorization: fBearer {os.environ[self.gateway[api_key_env]]}}, timeoutself.gateway[default_timeout] ) self._register() def _register(self): self.server.list_tools() async def list_tools(): return self.get_tools() self.server.call_tool() async def call_tool(name, arguments): return await self.execute(name, arguments) async def call_llm(self, messages: list) - str: payload { model: self.agent_cfg[model], messages: messages, temperature: self.agent_cfg[temperature], max_tokens: self.agent_cfg[max_tokens], } resp await self.client.post(/v1/chat/completions, jsonpayload) resp.raise_for_status() return resp.json()[choices][0][message][content] def get_tools(self) - list: raise NotImplementedError async def execute(self, name: str, arguments: dict): raise NotImplementedError这个基类把「读配置、建客户端、注册 MCP 工具、调模型」四件事收敛到一处。子类只需要实现get_tools和execute代码量能压到几十行。注意call_llm里统一走base_url所有 Agent 的模型请求都经过 TaoToken 通道Key 从环境变量取不落盘。4. 多 Agent 注册与路由跑通最小可用系统配置和基类就位后写一个具体的 Agent 和 Orchestrator把链路串起来。4.1 Coder Agent 实现from base_agent import BaseAgent from mcp.types import Tool, TextContent class CoderAgent(BaseAgent): def __init__(self): super().__init__(coder) def get_tools(self): return [ Tool( namegenerate_code, description根据任务描述生成代码。输入 requirement 为需求文本language 为目标语言返回完整可运行代码。, inputSchema{ type: object, properties: { requirement: {type: string, description: 代码需求描述}, language: {type: string, default: python} }, required: [requirement] } ) ] async def execute(self, name, arguments): if name generate_code: req arguments[requirement] lang arguments.get(language, python) messages [ {role: system, content: self.agent_cfg[system_prompt]}, {role: user, content: f用{lang}实现{req}} ] code await self.call_llm(messages) return [TextContent(typetext, textcode)] raise ValueError(f未知工具: {name}) if __name__ __main__: import asyncio from mcp.server.stdio import stdio_server agent CoderAgent() async def main(): async with stdio_server() as (r, w): await agent.server.run(r, w, agent.server.create_initialization_options()) asyncio.run(main())工具描述写得具体LLM 才知道什么时候该调它。generate_code的描述里明确说了输入输出这比写「生成代码」四个字强太多。4.2 Orchestrator 路由Orchestrator 负责按config.toml里的routing配置把任务分发给对应 Agent 链。import asyncio import tomllib from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class Orchestrator: def __init__(self, config_pathconfig.toml): with open(config_path, rb) as f: self.config tomllib.load(f) self.routing self.config[routing] async def run_pipeline(self, task_type: str, payload: dict): chain self.routing.get(task_type) if not chain: raise ValueError(f未注册的任务类型: {task_type}) context payload for role in chain: context await self._invoke(role, context) return context async def _invoke(self, role: str, payload: dict): params StdioServerParameters( commandpython, args[fagents/{role}_server.py] ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() tool_name tools.tools[0].name result await session.call_tool(tool_name, payload) return {text: result.content[0].text} if __name__ __main__: orch Orchestrator() result asyncio.run(orch.run_pipeline( code_review, {requirement: 实现一个带重试的 HTTP 客户端, language: python} )) print(result[text])run_pipeline按路由链顺序执行每个 Agent 的输出作为下一个的输入。这就是最小可用的多 Agent 协作Planner 拆任务、Coder 写代码、Reviewer 审查链路清晰每步可观测。5. 连通性验证与常见报错排查系统搭完先做三步验证再上真实任务。第一步验证 TaoToken 通道。用第 2 节的 curl 命令返回choices即通。这一步不通后面全是白费。第二步验证单个 Agent 能拉起。直接跑python agents/coder_server.py如果进程能起来并等待 stdio 输入说明 MCP Server 注册没问题。报ModuleNotFoundError就检查依赖报KeyError: TAOTOKEN_API_KEY就检查环境变量是否注入。第三步验证 Orchestrator 能串链路。跑第 4.2 节的__main__看是否返回审查结果。常见报错按这个顺序排查报错原因动作401 UnauthorizedKey 无效或未注入检查环境变量名与config.toml里api_key_env是否一致404 Not Foundbase_url 写错确认是https://taotoken.net/api不要带多余路径429 Too Many Requests触发限流去控制台看额度或给 Agent 加重试退避Connection refusedMCP Server 未启动检查settings.json里args路径是否正确asyncio.TimeoutError模型响应超时调大default_timeout或给call_llm包asyncio.wait_for上下文超窗消息历史过长给每个 Agent 加滑动窗口只保留最近 N 轮超时处理建议在call_llm里显式包一层import asyncio async def call_llm_safe(self, messages, timeout60): try: return await asyncio.wait_for(self.call_llm(messages), timeouttimeout) except asyncio.TimeoutError: return [模型调用超时请稍后重试]上下文窗口溢出是多 Agent 系统最常见的坑。每个 Agent 都维护自己的消息历史几轮下来就爆。解决方案是给基类加一个trim_history方法只保留最近 N 轮对话和关键长文本摘要。这个动作不做系统跑不了几个任务就会挂。如果排障过程中需要单独验证某个模型是否可用可以直接用模型对话页面测一条模型对话https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite接入相关的完整参数说明和错误码对照看接入文档接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite6. 长期编码与 Agent 场景的接入建议如果你只是跑通 Demo上面的配置够了。但如果要把这套系统长期用于日常编码、CI 流水线或企业内部 Agent 平台有几个点值得提前规划。第一Key 分层。Planner 和 Reviewer 调用频率低但要求质量高Coder 调用频率高且 token 消耗大Executor 调用最频繁但每次很短。按角色分 Key 之后你能在控制台清楚看到哪类 Agent 在吃额度也方便给不同角色设不同限流策略。第二模型分级。不是所有 Agent 都需要最强模型。Planner 和 Reviewer 用强模型保证质量Executor 这种只做工具调用的角色可以用更轻的模型成本能降一大截。config.toml里每个 Agent 独立配model就是为这个准备的。第三Coding Plan 适合长期编码场景。如果你的多 Agent 系统主要跑代码生成、审查、重构这类任务且调用量大可以了解下 Coding Plan 的额度方案比按量付费更适合稳定负载Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite第四Claude Code 接入。如果你团队用 Claude Code 做日常开发它本身也支持通过统一通道接入配置方式和本文的 Agent 基类思路一致把 base_url 和 Key 指到 TaoToken 即可Claude Code 接入https://taotoken.net/claude-code?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite最后说个实测经验多 Agent 系统最容易出问题的地方不是通信而是任务拆解。Planner 拆得粗Coder 就写不出东西拆得细链路又太长容易超时。建议先用code_review这种三 Agent 链路跑两周观察每步的输入输出质量再逐步加 Agent。一上来就上五六个 Agent排障会让你怀疑人生。系统跑通之后把settings.json和config.toml提交到版本库Key 走环境变量注入每个 Agent 的日志单独落文件。这样下次某个 Agent 抽风你能五分钟定位到是配置问题、Key 问题还是模型问题。这套骨架不复杂但能省下大量重复搭建的时间。
返回列表