1. 从零理解 MCP 客户端与服务端到底在解决什么问题
MCP 全称 Model Context Protocol,你可以把它理解成“AI 世界的 USB-C 接口”。以前我们让大模型调用外部能力,要么写死 Function Call 的 JSON Schema,要么直接拼 REST API,每换一个模型、每换一个工具就要重写一遍胶水代码。MCP 做的事情,是把“模型 ↔ 工具”的交互抽象成一套标准协议:客户端负责和模型对话、把工具列表喂给模型,服务端负责真正执行工具并把结果按 JSON-RPC 2.0 的格式回传。
这套东西适合谁?如果你正在做本地 AI 工具、想让 Cursor / Claude Desktop / 自研 Agent 调用本地文件、数据库、内部系统,又不想为每个模型单独适配,那 MCP 就是当前最省心的路径。它和 Function Call 最大的区别在于解耦:Function Call 是模型原生能力,工具描述直接塞进请求;MCP 是独立协议层,工具由服务端声明,客户端动态发现,跨平台通用。
我这次要交付的是一条能跑通的端到端链路:一个 Python 写的 MCP Server(暴露 add / multiply 两个工具),一个 MCP Client(用 OpenAI 兼容的 SDK 去调模型),中间所有模型请求统一走 TaoToken 的 Key 和 Base URL。这样你本地工具只需要维护一份 Key,就能切换不同模型,不用在每个客户端里重复配置。
整条链路的关键点有三个:第一,MCP Server 用 stdio 或 sse 传输,客户端要能连上;第二,客户端调模型时用的是 OpenAI 兼容接口,所以 Base URL 必须指向统一通道;第三,工具调用的结果要正确回填到 messages 里,否则模型会一直重复调用同一个工具。下面按“先搭服务端、再搭客户端、最后验证”的顺序展开,每一步都给可复制的代码和配置。
2. TaoToken 统一 Key 的前置准备与 settings.json 配置骨架
在写客户端之前,先把“模型通道”这件事定下来。MCP Client 本身不产生模型能力,它只是把工具列表和用户问题打包发给某个 OpenAI 兼容端点。如果你每个项目都去申请不同厂商的 Key,配置会散落在十几个文件里。用 TaoToken 的好处是:一个 Key、一个 Base URL,就能覆盖 Claude、GPT、国产模型等,客户端代码完全不用改。
你需要先拿到两样东西:API Key 和 Base URL。Key 在控制台的 API Keys 页面创建,Base URL 固定为https://taotoken.net/api(注意这个地址不带任何查询参数,直接作为 OpenAI SDK 的 base_url 使用)。模型 ID 则根据你要用的模型填,比如claude-sonnet-4-5、gpt-4o这类,具体以文档里的模型列表为准。
为了让配置可复用,我建议把模型通道信息单独抽成一个settings.json,客户端启动时读取。这样以后换模型只改一个文件:
{ "llm": { "api_key": "sk-你的TaoTokenKey", "base_url": "https://taotoken.net/api", "model": "claude-sonnet-4-5" }, "mcp": { "transport": "stdio", "command": "python", "args": ["server.py"] } }如果你更习惯 TOML(比如配合 Codex 或某些 CLI 工具),等价写法是:
[llm] api_key = "sk-你的TaoTokenKey" base_url = "https://taotoken.net/api" model = "claude-sonnet-4-5" [mcp] transport = "stdio" command = "python" args = ["server.py"]这里有个容易踩的坑:Base URL 末尾不要自己加/v1。OpenAI SDK 内部会拼接/chat/completions,如果你写成https://taotoken.net/api/v1,最终请求路径会变成/api/v1/chat/completions,部分通道会返回 404。统一用https://taotoken.net/api即可。
另外,Key 不要硬编码进 Git 仓库。本地开发可以用环境变量兜底:
import os api_key = os.getenv("TAOTOKEN_API_KEY", config["llm"]["api_key"])这样在 CI 或服务器上只需要注入环境变量,配置文件里留空也不会泄露。前置准备做完,接下来就是真正写服务端和客户端。
3. 可复制的 MCP 服务端与客户端配置骨架(server.py / client.py)
先写服务端。用官方mcp包里的 FastMCP,安装命令:
pip install "mcp[cli]" openaiserver.py内容如下,暴露两个工具和一个资源:
from mcp.server.fastmcp import FastMCP mcp = FastMCP("demo-server") @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" if __name__ == "__main__": mcp.run(transport="stdio")用mcp dev server.py可以在浏览器里看到工具列表,确认add_2_numbers和multiply_2_numbers都被正确注册。这一步过了,说明服务端没问题。
客户端client.py的核心逻辑是:启动 stdio 子进程连上 server,拿到 tools 列表,转成 OpenAI 的 function 格式,然后进入对话循环。关键片段:
import asyncio import json from contextlib import AsyncExitStack from mcp import ClientSession, StdioServerParameters, stdio_client from openai import AsyncOpenAI class MCPClient: def __init__(self, api_key, base_url, model): self.exit_stack = AsyncExitStack() self.client = AsyncOpenAI(api_key=api_key, base_url=base_url) self.model = model self.messages = [] async def connect(self, command, args): params = StdioServerParameters(command=command, args=args, env={}) transport = await self.exit_stack.enter_async_context(stdio_client(params)) self.stdio, self.write = transport self.session = await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() resp = await self.session.list_tools() self.available_tools = [ { "type": "function", "function": { "name": t.name, "description": t.description, "parameters": t.inputSchema, }, } for t in resp.tools ] print("已连接工具:", [t["function"]["name"] for t in self.available_tools]) async def ask(self, query: str) -> str: self.messages.append({"role": "user", "content": query}) resp = await self.client.chat.completions.create( model=self.model, messages=self.messages, tools=self.available_tools, ) msg = resp.choices[0].message while msg.tool_calls: for call in msg.tool_calls: name = call.function.name args = json.loads(call.function.arguments) result = await self.session.call_tool(name, args) text = result.content[0].text print(f"调用工具 {name}({args}) -> {text}") self.messages.extend([ {"role": "assistant", "content": None, "tool_calls": [call]}, {"role": "tool", "content": text, "tool_call_id": call.id}, ]) resp = await self.client.chat.completions.create( model=self.model, messages=self.messages, tools=self.available_tools, ) msg = resp.choices[0].message self.messages.append({"role": "assistant", "content": msg.content}) return msg.content async def close(self): await self.exit_stack.aclose()启动入口:
async def main(): with open("settings.json", "r", encoding="utf-8") as f: cfg = json.load(f) c = MCPClient(cfg["llm"]["api_key"], cfg["llm"]["base_url"], cfg["llm"]["model"]) try: await c.connect(cfg["mcp"]["command"], cfg["mcp"]["args"]) while True: q = input("\nQuery: ").strip() if q.lower() in ("quit", "exit"): break print("AI:", await c.ask(q)) finally: await c.close() if __name__ == "__main__": asyncio.run(main())注意call_tool返回的result.content[0].text是字符串,如果你的工具返回结构化数据,记得在服务端json.dumps一下,否则模型拿到的可能是 Python 的 repr 格式,解析会出问题。这套骨架同时支持 stdio 和 sse,只要把stdio_client换成sse_client(server_url)即可,其余逻辑不变。
4. 用统一 Key 完成一次请求验证:从 Query 到工具回填的完整链路
配置写好后,跑一次真实请求来验证。先启动客户端:
python client.py如果连接成功,你会看到类似输出:
已连接工具: ['add_2_numbers', 'multiply_2_numbers'] Query: 帮我算一下 12 加 30 等于多少 调用工具 add_2_numbers({'a': 12, 'b': 30}) -> 42 AI: 12 加 30 等于 42。这条链路里发生了四件事:第一,客户端把用户问题和工具列表一起发给 TaoToken 的/api/chat/completions;第二,模型返回一个tool_calls,里面是add_2_numbers和参数{"a":12,"b":30};第三,客户端通过 MCP 协议调用本地 server 的add_2_numbers,拿到结果42;第四,客户端把工具结果作为role: tool的消息回填,再次请求模型,模型生成最终自然语言回答。
再测一个多步场景,验证模型是否会连续调用工具:
Query: 先算 7 乘 8,再把结果加 6 调用工具 multiply_2_numbers({'a': 7, 'b': 8}) -> 56 调用工具 add_2_numbers({'a': 56, 'b': 6}) -> 62 AI: 7 乘 8 等于 56,再加 6 等于 62。这里能跑通说明while msg.tool_calls循环是正确的。如果你只调用一次工具就退出,模型会拿不到最终答案,因为它还在等工具结果。验证时重点看两个地方:一是tool_call_id是否和 assistant 消息里的 id 一致,二是messages里是否同时有 assistant 的 tool_calls 和 tool 的结果。这两个对了,链路就稳了。
如果你用的是 sse 传输,验证方式一样,只是连接时换成:
from mcp.client.sse import sse_client transport = await self.exit_stack.enter_async_context( sse_client("http://127.0.0.1:8256/sse") )sse 有个已知问题:连接大约两分钟后会自然断开,所以生产环境建议在每次call_tool前重连一次,或者直接迁移到官方新的streamable-http传输。验证阶段先用 stdio 跑通,再换 sse 排查网络问题,会省很多时间。
5. 本篇常见报错排查:401、local proxy failed、reading choices 与 OAuth
跑 MCP 客户端时,报错基本集中在模型通道和协议连接两块。下面按真实遇到的频率排一下。
401 Unauthorized:最常见。原因通常是 Key 没填对、Key 前后有空格、或者 Base URL 写成了带/v1的地址。排查方法是在客户端初始化后打印一次:
print("base_url =", self.client.base_url) print("key prefix =", self.client.api_key[:8])确认 base_url 是https://taotoken.net/api,key 前缀和你在控制台看到的一致。如果还是 401,去 API Keys 页面确认这个 Key 是否被禁用或额度耗尽。
local proxy failed / connection refused:这个报错一般出现在 sse 模式,客户端连不上127.0.0.1:8256。先确认 server 是否真的在跑,mcp dev server.py会打印监听端口。如果端口对但连不上,检查是不是被本地防火墙拦了,或者 server 启动时用了transport="stdio"却用 sse 去连。stdio 和 sse 不能混用,这是两套传输。
reading choices 相关报错:典型信息是'NoneType' object has no attribute 'choices'或reading 'choices'。这通常不是模型的问题,而是你的请求体里messages格式不对。比如把tool_calls消息的content写成了空字符串而不是None,或者tool_call_id缺失。检查回填逻辑:
self.messages.extend([ {"role": "assistant", "content": None, "tool_calls": [call]}, {"role": "tool", "content": text, "tool_call_id": call.id}, ])content必须是None,不能是"",否则部分通道会直接返回空响应。
OAuth / 认证跳转类报错:如果你在 Claude Code 或某些 CLI 里配置 MCP,可能会遇到要求 OAuth 登录的提示。这类工具通常需要你在settings.json或auth.json里显式写全三件套:Base URL、API Key、Model ID。以 Codex 的auth.json为例:
{ "base_url": "https://taotoken.net/api", "api_key": "sk-你的TaoTokenKey", "model": "claude-sonnet-4-5" }三个字段缺一个都可能触发 OAuth 流程或认证失败。Cline 的 MCP 配置同理,在cline_mcp_settings.json里把command、args、env写清楚,env 里带上TAOTOKEN_API_KEY。CC Switch 这类切换工具也是同样的三件套逻辑,不要只填 Key 不填 Base URL。
排查顺序建议:先看 HTTP 状态码(401 查 Key,404 查路径),再看 MCP 连接日志(工具列表是否为空),最后看 messages 结构(tool 回填是否完整)。大部分问题都出在这三层里的某一层。
6. 把统一 Key 接入你的 MCP 工作流:从验证到长期使用
跑通一次请求之后,接下来要考虑的是怎么把这套骨架用在你自己的项目里。我的做法是把settings.json作为唯一配置源,客户端、服务端、CLI 工具都读它。这样换模型只改一个字段,不用满仓库找 Key。
如果你要长期跑编码类 Agent,建议把模型通道固定成 Coding Plan 对应的配置,避免每次手动切模型。日常调试用模型对话页面快速验证 Key 是否可用,比在代码里反复跑客户端快得多。接入文档里有各语言 SDK 的完整示例,遇到 SDK 版本差异时对照一下能省不少时间。
最后给一个实用技巧:在客户端启动时加一行日志,把当前使用的 model 和 base_url 打出来。很多“模型答非所问”的问题,其实是配置读错了文件,或者环境变量覆盖了 settings.json。日志一打,问题立刻现形。这套骨架我已经在本地文件和数据库查询两个场景里跑过,工具数量加到十几个也没有出现协议层问题,你可以放心往上叠。