企业官网建设流程全解析

1. 为什么你的 Agent 每加一个工具就要改代码

先说一个我踩过的坑。去年做客服 Agent 的时候，工具从 3 个涨到 17 个，每加一个工具就要动三处代码：写 handler、改 TOOLS 字典、调 prompt 里的工具描述。最要命的是上线之后发现某个工具参数写错了，得重新发版。这种耦合度在生产环境里就是灾难。

MCP（Model Context Protocol）要解决的就是这件事。它把工具从 Agent 代码里彻底剥离出来，变成一个独立的 Server 进程，Agent 启动时通过 JSON-RPC 2.0 协议去问 Server「你有哪些工具」，拿到工具列表后自动注册到自己的工具系统里。整个过程 Agent 主逻辑一行不改。

你可以把 MCP 理解成 Agent 的「应用商店」：每个 MCP Server 就是一个 App，里面装着若干工具；Agent 是手机，开机时扫描已安装的 App，把里面的功能挂到桌面上。想加新功能？装个新 App 就行，不用刷机。

这套协议适合谁？三类人：一是自建 Agent 框架、工具数量超过 5 个的开发者；二是想让多个 Agent 共享同一套工具的后端团队；三是想把内部系统（数据库、工单、监控）包装成标准工具对外暴露的平台方。如果你还在用硬编码的 function calling 列表，工具数一超过 10 个就会开始难受。

这一篇我会从零实现一个可运行的 MCP Server + Client，包含 JSON-RPC 通信层、工具注册、自动发现、热插拔验证脚本，最后用 TaoToken 统一 Key 打通多工具调用的连通性验证。代码可以直接复制跑。

2. TaoToken 前置准备：统一 Key 与 API 通道

在写 MCP 之前，得先把模型调用这条链路理顺。MCP 解决的是「工具有哪些、怎么调」，但工具调用的决策还是模型做的——模型得能看懂工具描述、能返回结构化的 tool_call。所以你需要一个稳定的模型 API 通道。

我用 TaoToken 做统一入口，原因是它把多个模型的 Key 收敛成一个，Agent 侧只配一个 Base URL 和一个 Key，切换模型不用改代码。对 MCP 场景特别友好：工具描述会随 Server 变化，模型得频繁重新理解工具集，用统一通道省去多 Key 管理的麻烦。

先拿 Key。打开 https://taotoken.net/api-keys ，登录后创建一个 API Key，复制出来形如sk-xxxxxxxx。这个 Key 后面会写进环境变量，不要硬编码到代码里。

然后确认你的调用地址。TaoToken 的 API 入口是https://taotoken.net/api，兼容 OpenAI 的/v1/chat/completions格式。也就是说你原来用 openai SDK 的代码，只改base_url和api_key两个字段就能跑。

模型 ID 这块要注意：MCP 场景下模型必须支持 function calling / tool use，否则工具描述传过去它也不认。选模型时优先挑带 tool 能力的，具体可用列表在 https://taotoken.net/models 查。我实测下来，工具数量在 20 个以内时，主流模型的工具选择准确率都够用。

环境变量这样配：

export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL="你的模型ID"

如果你用 Claude Code 或者 Cline 这类客户端，配置方式略有不同，但核心三件套不变：Base URL、API Key、Model ID。这三样凑齐，模型通道就通了。MCP 的 Server 和 Client 都跑在本地，模型调用走 TaoToken，整条链路就完整了。

有一点提醒：MCP Server 本身不调模型，它只负责执行工具。模型调用发生在 Agent 主循环里。所以 TaoToken 的 Key 是配在 Agent 侧的，不是配在 MCP Server 里的。这个边界要分清，不然后面调试会绕晕。

3. 可复制的 MCP Server 配置与工具注册

现在进入正题。MCP 的通信层是 JSON-RPC 2.0，标准方法有四个：initialize（握手）、tools/list（列工具）、tools/call（调工具）、ping（心跳）。我先把协议层封装出来。

import json, uuid from typing import Callable, Optional class JSONRPC: VERSION = "2.0" @staticmethod def request(method: str, params: dict = None, id: str = None) -> dict: return { "jsonrpc": JSONRPC.VERSION, "method": method, "params": params or {}, "id": id or str(uuid.uuid4())[:8], } @staticmethod def response(id: str, result: dict) -> dict: return {"jsonrpc": JSONRPC.VERSION, "result": result, "id": id} @staticmethod def error(id: str, code: int, message: str, data: dict = None) -> dict: return { "jsonrpc": JSONRPC.VERSION, "error": {"code": code, "message": message, "data": data or {}}, "id": id, } class MCPMethods: INITIALIZE = "initialize" TOOLS_LIST = "tools/list" TOOLS_CALL = "tools/call" PING = "ping"

这段是协议骨架，没什么花活。id用 uuid 前 8 位，保证请求响应能对上。生产环境如果走 HTTP，建议用完整 uuid 避免碰撞。

接下来是 Server 本体。核心是一个tools字典，key 是工具名，value 存定义和 handler。register_tool负责往字典里塞，handle_request负责路由。

class MCPServer: def __init__(self, name: str, version: str = "1.0.0"): self.name = name self.version = version self.tools: dict[str, dict] = {} self._handlers = { MCPMethods.INITIALIZE: self._handle_initialize, MCPMethods.TOOLS_LIST: self._handle_tools_list, MCPMethods.TOOLS_CALL: self._handle_tools_call, MCPMethods.PING: self._handle_ping, } def register_tool(self, name: str, description: str, parameters: list[dict], handler: Callable): self.tools[name] = { "definition": { "name": name, "description": description, "inputSchema": { "type": "object", "properties": { p["name"]: { "type": p.get("type", "string"), "description": p.get("description", ""), } for p in parameters }, "required": [p["name"] for p in parameters if p.get("required", True)], }, }, "handler": handler, } def handle_request(self, raw: dict) -> dict: method = raw.get("method", "") req_id = raw.get("id", "") handler = self._handlers.get(method) if not handler: return JSONRPC.error(req_id, -32601, f"Method not found: {method}") try: return JSONRPC.response(req_id, handler(raw.get("params", {}))) except Exception as e: return JSONRPC.error(req_id, -32000, str(e)) def _handle_initialize(self, params: dict) -> dict: return { "protocolVersion": "2024-11-05", "serverInfo": {"name": self.name, "version": self.version}, "capabilities": {"tools": {}}, } def _handle_tools_list(self, params: dict) -> dict: return {"tools": [t["definition"] for t in self.tools.values()]} def _handle_tools_call(self, params: dict) -> dict: tool = self.tools.get(params.get("name", "")) if not tool: raise ValueError(f"Tool not found: {params.get('name')}") result = tool["handler"](**params.get("arguments", {})) return {"content": [{"type": "text", "text": str(result)}], "isError": False} def _handle_ping(self, params: dict) -> dict: return {"status": "ok"}

注意inputSchema的结构，这是给模型看的工具描述，格式必须和 OpenAI function calling 的 parameters 对齐，否则模型理解不了参数。required字段从参数列表里自动推导，标了required: False的就不进必填项。

现在注册两个 Server 做演示。一个是搜索服务，一个是通知服务。

search_server = MCPServer("search-service") search_server.register_tool( name="web_search", description="搜索互联网获取信息，用于查询最新新闻、事实、人物。", parameters=[ {"name": "query", "type": "string", "description": "搜索关键词", "required": True}, {"name": "max_results", "type": "number", "description": "结果数，默认5", "required": False}, ], handler=lambda query, max_results=5: f"搜索[{query}]返回{max_results}条结果", ) notify_server = MCPServer("notification-service") notify_server.register_tool( name="send_email", description="发送邮件，用于通知、报告分发。", parameters=[ {"name": "to", "type": "string", "description": "收件人邮箱", "required": True}, {"name": "subject", "type": "string", "description": "邮件主题", "required": True}, {"name": "body", "type": "string", "description": "邮件正文", "required": True}, ], handler=lambda to, subject, body: f"邮件已发送至{to}，主题：{subject}", )

到这里 Server 侧就完成了。两个 Server 各自独立，互不感知，这就是「工具服务化」——每个 Server 是一个沙箱，一个挂了不影响另一个。

4. MCP Client 自动发现与热插拔验证

Client 是 Agent 侧的东西，职责是连 Server、拉工具列表、把远程工具包装成本地可调用的函数。关键在_register_to_agent这一步，它把发现的工具动态注入 Agent 的工具注册表。

class MCPClient: def __init__(self, agent): self.agent = agent self.servers: dict[str, MCPServer] = {} self.discovered_tools: dict[str, dict] = {} def connect_server(self, server: MCPServer): self.servers[server.name] = server self._send_and_receive(server, MCPMethods.INITIALIZE, {}) resp = self._send_and_receive(server, MCPMethods.TOOLS_LIST, {}) for tool_def in resp.get("result", {}).get("tools", []): self.discovered_tools[tool_def["name"]] = { "server": server.name, "definition": tool_def, } print(f"[DISCOVER] {tool_def['name']} from {server.name}") self._register_to_agent() def _send_and_receive(self, server: MCPServer, method: str, params: dict) -> dict: return server.handle_request(JSONRPC.request(method, params)) def call_tool(self, tool_name: str, arguments: dict) -> str: info = self.discovered_tools.get(tool_name) if not info: return f"Error: Tool '{tool_name}' not found" server = self.servers[info["server"]] resp = self._send_and_receive(server, MCPMethods.TOOLS_CALL, { "name": tool_name, "arguments": arguments, }) if "error" in resp: return f"MCP Error: {resp['error']['message']}" content = resp.get("result", {}).get("content", []) return content[0].get("text", str(content)) if content else str(resp) def _register_to_agent(self): for name, info in self.discovered_tools.items(): schema = info["definition"].get("inputSchema", {}) props = schema.get("properties", {}) def make_wrapper(tool_name): def wrapper(**kwargs): return self.call_tool(tool_name, kwargs) return wrapper self.agent.register_tool( name=name, description=info["definition"]["description"], parameters=[ {"name": pn, "type": pi.get("type", "string"), "description": pi.get("description", ""), "required": pn in schema.get("required", [])} for pn, pi in props.items() ], handler=make_wrapper(name), )

make_wrapper这里有个闭包陷阱要注意：如果直接写lambda **kw: self.call_tool(name, kw)，循环里name会被最后一次迭代覆盖。用工厂函数把tool_name固定住才对。这个坑我在第一版实现时踩过，所有工具都调到了最后一个。

现在跑热插拔验证。核心思路是：先连一个 Server，看工具列表；再连第二个 Server，看工具列表是否自动增长；最后调一个远程工具，确认能通。

class DummyAgent: def __init__(self): self.tools = {} def register_tool(self, name, description, parameters, handler): self.tools[name] = {"description": description, "handler": handler} agent = DummyAgent() mcp = MCPClient(agent) mcp.connect_server(search_server) print("连接 search 后工具:", list(agent.tools.keys())) mcp.connect_server(notify_server) print("连接 notify 后工具:", list(agent.tools.keys())) print("调用 web_search:", agent.tools["web_search"]["handler"](query="MCP协议", max_results=3)) print("调用 send_email:", agent.tools["send_email"]["handler"]( to="boss@example.com", subject="日报", body="今日完成MCP实现"))

预期输出：

[DISCOVER] web_search from search-service 连接 search 后工具: ['web_search'] [DISCOVER] send_email from notification-service 连接 notify 后工具: ['web_search', 'send_email'] 调用 web_search: 搜索[MCP协议]返回3条结果 调用 send_email: 邮件已发送至boss@example.com，主题：日报

看到工具列表从 1 个变成 2 个，且调用都返回正常结果，热插拔就验证通过了。整个过程 Agent 的DummyAgent类没有任何改动，工具是运行时注入的。

如果你要接真实模型，把agent.tools里的工具定义转成 OpenAI tools 格式，连同用户消息一起发给 TaoToken 的/v1/chat/completions，模型返回tool_calls后，你根据function.name找到对应的 handler 执行，把结果作为tool角色消息回传。这就是完整的 Agent 工具调用循环。

5. 常见报错排查：401、local proxy failed、reading choices

MCP 落地时踩的坑集中在两类：模型通道问题和协议层问题。我按真实报错逐个说。

401 Unauthorized。这个基本是 Key 的问题。先确认环境变量有没有生效：echo $TAOTOKEN_API_KEY，如果为空说明 export 没在当前 shell 生效。再确认 Key 有没有多余空格，复制时容易带上换行。还有一种情况是 Base URL 写成了https://taotoken.net/api/v1，多加了/v1，正确写法是https://taotoken.net/api，SDK 会自己拼/v1/chat/completions。如果用的是 Claude Code 或 Cline，检查配置文件里的ANTHROPIC_BASE_URL或OPENAI_BASE_URL是否指向正确地址。

local proxy failed / connection refused。这个报错通常出现在客户端配置了本地代理端口但代理没起来。MCP 场景下，如果你用 stdio 方式启动 Server，Client 是通过子进程管道通信的，不走网络，不会出现这个错。出现这个错说明你在 HTTP 模式下配了http://127.0.0.1:xxxx但服务没监听。检查 Server 是否真的启动了，端口是否被占用。另外有些客户端默认会读系统代理环境变量，如果HTTP_PROXY指向一个不存在的地址，也会报这个。清掉代理环境变量再试。

reading 'choices' of undefined。这是 OpenAI SDK 的经典报错，意思是响应体里没有choices字段。原因通常是：一、模型 ID 写错了，服务端返回了错误 JSON 而不是标准响应；二、Base URL 配错，请求打到了非兼容端点；三、请求体格式不对，比如messages为空。排查方法是在代码里打印原始响应：

import openai client = openai.OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) try: resp = client.chat.completions.create( model=os.environ["TAOTOKEN_MODEL"], messages=[{"role": "user", "content": "ping"}], ) print(resp.choices[0].message.content) except Exception as e: print("原始错误:", e)

如果这里就报错，说明模型通道没通，跟 MCP 无关。先把这个最小请求跑通，再往上叠 MCP。

OAuth / token expired。有些客户端（比如 Claude Code）走的是 OAuth 流程，配置里如果混用了 OAuth token 和 API Key 会冲突。用 TaoToken 的 Key 时，确保配置项是 API Key 字段而不是 OAuth 字段。CC Switch 这类工具切换配置时，注意 Base URL、Key、Model ID 三件套要一起换，只换其中一个会导致鉴权失败。

工具调用返回 tool not found。这是 MCP 协议层的错，不是模型通道的错。检查discovered_tools里有没有这个工具名，大小写是否一致。MCP 的工具名是大小写敏感的，web_search和Web_Search是两个工具。另外如果 Server 重启了但 Client 没重连，discovered_tools里还是旧列表，调新工具就会找不到。生产环境建议加个心跳检测，Server 挂了自动重连并刷新工具列表。

模型不返回 tool_calls。工具描述传过去了，但模型就是不用工具，直接编答案。这通常是工具描述写得太模糊。description要写清楚「什么时候用这个工具」，而不是「这个工具是什么」。比如web_search的描述写成「搜索互联网获取信息，用于查询最新新闻、事实、人物」，比「搜索工具」强十倍。参数描述同理，query要写「搜索关键词」而不是「输入」。

6. 把 MCP 接进你的 Agent 主循环

前面验证的是工具发现和调用，现在说怎么接进真实的 Agent 循环。核心是把discovered_tools转成模型能理解的 tools 格式，然后在模型返回tool_calls时分发执行。

def build_tools_payload(mcp_client: MCPClient) -> list[dict]: payload = [] for name, info in mcp_client.discovered_tools.items(): schema = info["definition"]["inputSchema"] payload.append({ "type": "function", "function": { "name": name, "description": info["definition"]["description"], "parameters": schema, }, }) return payload def run_agent_turn(mcp_client, messages, client, model): tools = build_tools_payload(mcp_client) resp = client.chat.completions.create( model=model, messages=messages, tools=tools, tool_choice="auto", ) msg = resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for call in msg.tool_calls: args = json.loads(call.function.arguments) result = mcp_client.call_tool(call.function.name, args) messages.append({ "role": "tool", "tool_call_id": call.id, "content": result, }) return run_agent_turn(mcp_client, messages, client, model)

这段是 Agent 的核心循环：模型决定调哪个工具，Client 执行，结果回传，模型继续推理，直到不再调工具为止。build_tools_payload每次从discovered_tools现算，所以新连的 Server 工具会自动出现在下一轮请求里，这就是热插拔在 Agent 层面的体现。

接 TaoToken 的完整初始化：

import os, openai client = openai.OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) model = os.environ["TAOTOKEN_MODEL"] mcp = MCPClient(agent) mcp.connect_server(search_server) mcp.connect_server(notify_server) messages = [{"role": "user", "content": "搜索MCP协议最新进展，把摘要发到 dev@example.com"}] print(run_agent_turn(mcp, messages, client, model))

跑通之后你会看到模型先调web_search，拿到结果后再调send_email，两步工具调用自动串联。整个过程 Agent 代码没变，工具是运行时注入的。

想验证热插拔，在connect_server之后再注册一个新工具到 Server，然后重新connect_server一次，discovered_tools会刷新，下一轮请求模型就能看到新工具。生产环境可以把这个刷新做成定时任务或者 Server 推送通知。

如果你要长期跑 Agent 任务，建议用 Coding Plan 这类按量方案，工具调用频繁时成本可控。模型对话入口可以用来快速验证工具描述写得对不对，接入文档里有完整的参数说明。工具生态这块，MCP 的价值不在协议本身，而在于它让工具变成了可分发、可组合的资产——你写的 Server 别人能直接用，别人写的你也能挂载。这才是「应用商店」的真正含义。