如何使用MCP开发一个客户端和服务端：TaoToken 统一 Key 接入与配置骨架

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(transportstdio)用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_keyapi_key, base_urlbase_url) self.model model self.messages [] async def connect(self, command, args): params StdioServerParameters(commandcommand, argsargs, 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( modelself.model, messagesself.messages, toolsself.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( modelself.model, messagesself.messages, toolsself.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, encodingutf-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/apikey 前缀和你在控制台看到的一致。如果还是 401去 API Keys 页面确认这个 Key 是否被禁用或额度耗尽。local proxy failed / connection refused这个报错一般出现在 sse 模式客户端连不上127.0.0.1:8256。先确认 server 是否真的在跑mcp dev server.py会打印监听端口。如果端口对但连不上检查是不是被本地防火墙拦了或者 server 启动时用了transportstdio却用 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 查 Key404 查路径再看 MCP 连接日志工具列表是否为空最后看 messages 结构tool 回填是否完整。大部分问题都出在这三层里的某一层。6. 把统一 Key 接入你的 MCP 工作流从验证到长期使用跑通一次请求之后接下来要考虑的是怎么把这套骨架用在你自己的项目里。我的做法是把settings.json作为唯一配置源客户端、服务端、CLI 工具都读它。这样换模型只改一个字段不用满仓库找 Key。如果你要长期跑编码类 Agent建议把模型通道固定成 Coding Plan 对应的配置避免每次手动切模型。日常调试用模型对话页面快速验证 Key 是否可用比在代码里反复跑客户端快得多。接入文档里有各语言 SDK 的完整示例遇到 SDK 版本差异时对照一下能省不少时间。最后给一个实用技巧在客户端启动时加一行日志把当前使用的 model 和 base_url 打出来。很多“模型答非所问”的问题其实是配置读错了文件或者环境变量覆盖了 settings.json。日志一打问题立刻现形。这套骨架我已经在本地文件和数据库查询两个场景里跑过工具数量加到十几个也没有出现协议层问题你可以放心往上叠。