如何使用MCP开发一个客户端和服务端：从零搭建到TaoToken统一Key接入的完整实践

1. 从零理解 MCP 客户端与服务端到底在做什么MCPModel 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 不要提交到 Gitconfig.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(namedemo) 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(Im seeing this error:), base.UserMessage(error), base.AssistantMessage(Ill help debug that. What have you tried so far?), ] if __name__ __main__: mcp.run(transportstdio)解释一下三个装饰器。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_keyapi_key, base_urlbase_url) self.model model self.message [] async def connect_to_stdio_server(self, mcp_name, command, args, env{}): server_params StdioServerParameters(commandcommand, argsargs, envenv) 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( modelself.model, messagesself.message, toolsself.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(fcalling 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( modelself.model, messagesself.message, toolsself.available_tools, max_tokens8048, ) 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 执行把结果以 roletool 塞回消息历史再让模型生成最终回答。注意 tool_call_id 必须对上否则模型会报错。如果你要远程部署把 stdio_client 换成 sse_client服务端启动改成mcp.run(transportsse)并指定 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.jsonCline 的 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 轮工具调用防止模型陷入无限调用。生产环境里这个保护很有必要我见过模型反复调同一个工具出不来的情况。加上计数器和超时链路就稳了。