1. 从零理解 MCP 客户端与服务端到底在做什么

MCP（Model Context Protocol）说白了就是给大模型装一个“标准插座”。你写好的工具函数，只要按 MCP 的格式暴露出去，任何支持 MCP 的客户端（Claude Desktop、Cursor、你自己写的 Python 客户端）都能直接调用，不用再为每个模型单独写一套 Function Call 的胶水代码。它解决的问题是工具碎片化：以前接一个数据库要写一套 schema，接一个浏览器又要写一套，现在统一成 JSON-RPC 2.0 的通信格式，服务端只管声明能力，客户端只管发现和调用。

这套东西适合谁？如果你正在做 AI Agent、想把公司内部系统（CRM、工单、日志查询）安全地暴露给大模型，或者你只是想搞明白“客户端和服务端握手”到底发生了什么，那这篇就是给你写的。我会先带你把本地 stdio 模式跑通，看到完整的工具列表和调用日志，再把服务端的鉴权通道切到 TaoToken 统一 Key，让整条链路走同一个 API 入口。

先明确两个角色。MCP 服务端（Server）负责注册工具、资源、提示词，它不关心谁来调，只按协议响应。MCP 客户端（Client）负责启动或连接服务端、拉取工具列表、把用户问题交给大模型、解析模型返回的 tool_calls、再去调用服务端工具、把结果塞回对话。整个握手过程是：客户端 initialize → 服务端返回能力 → 客户端 list_tools → 拿到工具清单 → 进入对话循环。

我实测下来，最容易卡住的地方不是协议本身，而是环境变量和传输方式选错。stdio 模式适合本地单机调试，服务端作为子进程被客户端拉起；SSE 或 streamable-http 适合远程部署，但连接会自然断开，需要重连逻辑。下面从安装开始，一步步来。

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

在把服务端 endpoint 切到统一通道之前，你得先有一个能用的 Key 和 Base URL。TaoToken 在这里扮演的角色是统一的大模型 API 入口，客户端里 AsyncOpenAI 的 base_url 指向它，api_key 用你申请的那把，模型 ID 按文档填。这样你的 MCP 客户端在解析 tool_calls 时，走的就是同一条鉴权通道，不用在多个平台之间来回切换 Key。

具体操作：打开 https://taotoken.net/api-keys 创建一把 API Key，复制保存。然后确认你要用的模型 ID，这个在模型对话页面能看到。Base URL 固定用 https://taotoken.net/api，注意不要带多余的路径后缀，OpenAI SDK 会自己拼 /chat/completions。

这里有个细节很多人踩坑：环境变量名。我建议统一写成 TAOTOKEN_API_KEY 和 TAOTOKEN_BASE_URL，然后在代码里读，而不是硬编码。这样你本地调试和后面部署到服务器时，只改环境变量就行。配置文件我习惯放一个 config.json，结构如下，路径放在项目根目录：

{ "llm": { "api_key": "sk-你的TaoTokenKey", "base_url": "https://taotoken.net/api", "model": "你的模型ID" }, "mcp": { "server_command": "python", "server_args": ["server.py"], "transport": "stdio" } }

如果你用的是 Claude Code 这类工具，配置方式不太一样，它读的是 settings 文件。但核心三件套不变：Base URL、Key、Model ID。你可以先在模型对话页面确认这把 Key 能正常出结果，再去接 MCP，这样排障时能快速定位是 Key 的问题还是协议的问题。

注意：Key 不要提交到 Git，config.json 加进 .gitignore。生产环境用环境变量注入。

装依赖这一步别偷懒。官方 Python SDK 用pip install "mcp[cli]"，如果你用 uv 就uv add "mcp[cli]"。客户端还要装 openai 的异步版本，因为我们要用 AsyncOpenAI。装完之后python -c "import mcp; print(mcp.__version__)"验证一下，能打印版本号就说明环境没问题。

3. 可复制配置：服务端注册与客户端连接完整代码

先写服务端。核心是用 FastMCP 装饰器把普通 Python 函数变成 MCP 工具。下面这个 server.py 注册了两个工具、一个资源、一个提示词，启动方式用 stdio，方便本地调试：

from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.prompts import base mcp = FastMCP(name="demo") @mcp.tool() def add_2_numbers(a: int, b: int) -> int: """两个数字相加""" return a + b @mcp.tool() def multiply_2_numbers(a: int, b: int) -> int: """两个数字相乘""" return a * b @mcp.resource("config://app") def get_config() -> str: """静态配置数据""" return "App configuration here" @mcp.prompt() def debug_error(error: str) -> list[base.Message]: return [ base.UserMessage("I'm seeing this error:"), base.UserMessage(error), base.AssistantMessage("I'll help debug that. What have you tried so far?"), ] if __name__ == "__main__": mcp.run(transport="stdio")

解释一下三个装饰器。@mcp.tool()注册的是可执行动作，模型能调它产生副作用或计算；@mcp.resource()注册的是只读数据，类似 GET 端点，用来给模型补充上下文；@mcp.prompt()注册的是可复用提示模板，参数化之后能标准化交互。stdio 模式下，服务端不监听端口，而是等客户端通过标准输入输出跟它通信。

客户端这边，我用 AsyncOpenAI 接 TaoToken 通道，同时用 stdio_client 拉起上面的服务端。关键代码：

import asyncio import json from contextlib import AsyncExitStack from typing import Optional from mcp import ClientSession, StdioServerParameters, stdio_client from openai import AsyncOpenAI class StdioMCPClient: def __init__(self, api_key, base_url, model): self.session: Optional[ClientSession] = None self.exit_stack = AsyncExitStack() self.client = AsyncOpenAI(api_key=api_key, base_url=base_url) self.model = model self.message = [] async def connect_to_stdio_server(self, mcp_name, command, args, env={}): server_params = StdioServerParameters(command=command, args=args, env=env) stdio_transport = await self.exit_stack.enter_async_context( stdio_client(server_params) ) self.stdio, self.write = stdio_transport self.session = await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() response = await self.session.list_tools() tools = response.tools print(f"成功链接到 {mcp_name} 服务，对应 tools：", [t.name for t in tools]) self.available_tools = [ { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema, }, } for tool in response.tools ] async def process_query(self, query: str): self.message.append({"role": "user", "content": query}) response = await self.client.chat.completions.create( model=self.model, messages=self.message, tools=self.available_tools, ) final_text = [] assistant_message = response.choices[0].message while assistant_message.tool_calls: for tool_call in assistant_message.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) result = await self.session.call_tool(tool_name, tool_args) print(f"calling tools {tool_name}, with args {tool_args}") print("Result:", result.content[0].text) self.message.extend([ {"role": "assistant", "content": None, "tool_calls": [tool_call]}, {"role": "tool", "content": result.content[0].text, "tool_call_id": tool_call.id}, ]) response = await self.client.chat.completions.create( model=self.model, messages=self.message, tools=self.available_tools, max_tokens=8048, ) assistant_message = response.choices[0].message final_text.append(assistant_message.content) return "\n".join(final_text) async def chat_loop(self): self.message = [] while True: query = input("\nQuery: ").strip() if query.lower() == "quit": break if not query: continue response = await self.process_query(query) print("\nAI:", response) async def cleanup(self): await self.exit_stack.aclose() async def main(): with open("config.json", "r") as f: config = json.load(f) client = StdioMCPClient( config["llm"]["api_key"], config["llm"]["base_url"], config["llm"]["model"], ) try: await client.connect_to_stdio_server("testserver", "python", ["server.py"], {}) await client.chat_loop() finally: await client.cleanup() if __name__ == "__main__": asyncio.run(main())

这段代码里，connect_to_stdio_server完成握手和工具发现，process_query是核心循环：先让模型决定调哪个工具，拿到 tool_calls 后去 session.call_tool 执行，把结果以 role=tool 塞回消息历史，再让模型生成最终回答。注意 tool_call_id 必须对上，否则模型会报错。

如果你要远程部署，把 stdio_client 换成 sse_client，服务端启动改成mcp.run(transport="sse")，并指定 host 和 port。SSE 连接大概两分钟会自然断开，所以客户端要加一个 reconnect 函数，在每次 call_tool 前重新建立 session。这个坑我在生产环境踩过，不加重连的话，长对话跑到一半就断了。

4. 验证请求：一次完整调用日志与成功结果

配置写完后，跑起来看日志。先启动客户端，它会自动拉起 server.py 子进程。终端输出应该是：

成功链接到 testserver 服务，对应 tools： ['add_2_numbers', 'multiply_2_numbers'] Query: 帮我算一下 12 加 30 等于多少 calling tools add_2_numbers, with args {'a': 12, 'b': 30} Result: 42 AI: 12 加 30 等于 42。

再试一个乘法，验证多工具链路：

Query: 先算 7 乘 8，再把结果加 6 calling tools multiply_2_numbers, with args {'a': 7, 'b': 8} Result: 56 calling tools add_2_numbers, with args {'a': 56, 'b': 6} Result: 62 AI: 7 乘 8 等于 56，再加 6 等于 62。

这里能看到模型连续调了两个工具，中间结果通过消息历史传递，这就是 MCP 多工具调用的完整链路。整个过程里，模型请求走的是 TaoToken 的 Base URL，鉴权用同一把 Key，工具执行在本地 stdio 子进程里完成，两边解耦得很干净。

如果你想确认请求确实到了统一通道，可以在客户端初始化时打印 base_url，或者抓一下 AsyncOpenAI 的请求日志。正常情况下，模型返回的 tool_calls 结构里 function.name 和 function.arguments 都是标准 JSON，解析不会出问题。实测下来，从握手到第一次工具调用成功，整个链路在本地不到两秒。

提示：验证阶段先用简单算术工具，确认链路通了再换成真实业务工具，比如查数据库、调内部 API。这样排障范围小。

5. 本篇常见错误排查清单

报错一：401 Unauthorized。这是 Key 或 Base URL 的问题。先检查 config.json 里的 api_key 是不是复制完整，有没有多余空格。再确认 base_url 是 https://taotoken.net/api，不要写成带 /v1 的路径。如果还报 401，去 API Keys 页面重新生成一把，排除 Key 被禁用的情况。

报错二：local proxy failed 或连接超时。这种一般是网络出口问题，不是 MCP 协议本身。检查你的运行环境能不能正常访问外部 API，本地防火墙有没有拦。stdio 模式下服务端是子进程，不涉及端口，所以这个报错基本都出在客户端请求模型那一步。

报错三：reading 'choices' 相关错误。典型是模型返回结构不符合预期，或者 max_tokens 设太小导致返回被截断。检查 model ID 是否填对，TaoToken 通道支持的模型名要和文档一致。另外把 max_tokens 调到 8048 以上，避免长回答被切。

报错四：OAuth 或鉴权头缺失。如果你用的是 Claude Code 或 Cline 这类工具，它们读的是自己的 settings 文件，不是 config.json。这时候要写全三件套：Base URL、Key、Model ID，缺一个都会握手失败。Claude Code 的配置在 settings.json，Cline 的 MCP 配置在它自己的 mcp 设置里，格式不同但字段含义一样。

报错五：工具列表为空。客户端连上了但 list_tools 返回空，说明服务端装饰器没生效。检查@mcp.tool()有没有写在函数上面，函数有没有被正确导入。还有一种情况是 stdio 子进程启动失败，比如 python 路径不对，把 command 改成绝对路径试试。

报错六：SSE 连接两分钟后断开。这是 SSE 传输的已知行为，不是 bug。解决办法是在客户端加 reconnect 逻辑，每次 call_tool 前重新 enter_async_context 建立 session。或者直接升级到 streamable-http 传输，官方新版本已经支持，稳定性更好。

排障时建议按顺序来：先确认 Key 能单独调通模型，再确认服务端能单独 list_tools，最后合起来跑。这样能把问题定位到具体环节，不用瞎猜。

6. 把链路固定下来：从调试到长期使用的建议

跑通之后，我建议你把配置和代码分开管理。config.json 只放环境相关的 Key 和 URL，代码里读配置，这样换环境只改一个文件。工具函数单独放一个 tools 目录，每个业务模块一个文件，server.py 只负责注册，保持可维护性。

如果你打算长期跑 Agent 类任务，可以考虑用 Coding Plan 这类方案来管理调用配额和模型切换，避免每次手动改 Key。接入文档里有完整的参数说明，遇到协议层面的问题可以先查那里。模型对话页面适合快速验证某个模型 ID 能不能用，省得在代码里反复试。

最后说一个实用技巧：在 process_query 里加一个最大循环次数限制，比如 while 循环最多跑 5 轮工具调用，防止模型陷入无限调用。生产环境里这个保护很有必要，我见过模型反复调同一个工具出不来的情况。加上计数器和超时，链路就稳了。