ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

快速上手:用 Python SDK 实现你的第一个 MCP Client 并接入 TaoToken

快速上手:用 Python SDK 实现你的第一个 MCP Client 并接入 TaoToken 1. 从零跑通 MCP Client为什么 stdio 是最短路径MCP Client 是连接 LLM 与 MCP Server 的中间层它负责把用户的问题、工具清单一起交给模型模型决定调用哪个工具后Client 再去 Server 执行并把结果回传。如果你刚接触 MCP想先看到一次真实的工具调用结果stdio 传输是最省事的方式Client 直接以子进程方式拉起 Server通过标准输入输出通信不需要额外端口、不需要网络配置一个 Python 文件就能跑完整个链路。这篇面向的是想从零跑通 MCP Client 的 Python 开发者聚焦 stdio 传输下连接 LLM 的完整链路。我会给出可复制的 Python SDK 客户端骨架、TaoToken 统一 Key/API 通道的 config 配置片段以及一次本地 stdio 调用验证动作。你跟着敲完大概十分钟内就能看到第一个 MCP 工具调用结果。先说清楚 stdio 和 SSE 的区别避免选错方向。stdio 是一对一关系一个 Server 进程只服务启动它的那个 Client适合本地开发、单机集成、快速验证。SSE 是 Server 独立进程运行多个 Client 可以随时连上断开适合部署成服务。本文只走 stdio因为它的心智负担最小出问题也最好排查。整个链路是这样的Client 启动 Server 子进程初始化会话拉取工具列表把工具描述和用户问题一起发给 LLMLLM 返回 tool_callsClient 执行工具把结果塞回消息历史再让 LLM 生成最终回答。理解这条链路后面所有代码都是它的展开。2. TaoToken 前置统一 Key 与 API 通道准备在写代码之前先把 LLM 的访问通道准备好。MCP Client 本身不产生模型能力它需要一个 OpenAI 兼容的接口来发 function calling 请求。TaoToken 提供统一的 Key 和 API 通道你只需要一个 base_url 和一个 api_key就能用标准 OpenAI SDK 的写法调用模型不用为每个模型单独改代码。具体操作路径打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后进入控制台在 API Keys 页面创建一个 Key。这个 Key 就是后面 config 里的 api_keybase_url 统一填 https://taotoken.net/api。注意 API 地址不带任何查询参数保持干净。创建 Key 的入口在控制台里建议单独建一个用于本地开发的 Key方便后续轮换。拿到 Key 后不要硬编码进代码用环境变量注入这样提交到 Git 也不会泄露。下面这段就是你要放进 config 的核心片段后面所有代码都引用它。# config.py import os TAOTOKEN_CONFIG { base_url: https://taotoken.net/api, api_key: os.getenv(TAOTOKEN_API_KEY), model: claude-sonnet-4-20250514, # 按控制台可用模型替换 }如果你更习惯用命令行验证通道是否通可以先跑一个最小请求确认 Key 和 base_url 没问题再去写 MCP 那部分。这样排障时能快速定位是通道问题还是代码问题。export TAOTOKEN_API_KEY你的Key curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d {model:claude-sonnet-4-20250514,messages:[{role:user,content:ping}]}返回里有 choices 字段就说明通道正常。这一步花不了一分钟但能省掉后面大量「到底是哪错了」的纠结。3. 可复制配置MCP Client 骨架与 stdio 连接现在进入正题。先建项目用 uv 管理依赖装 mcp 和 openai 两个包就够了。如果你还没装 uv用 pip 装也行命令等价。uv init mcp-client-demo cd mcp-client-demo uv add mcp openai接着写一个最简单的 Server 用于验证它只提供一个工具返回一句固定文本。把它存成demo_server.py后面 Client 会以子进程方式拉起它。# demo_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(demo-server) mcp.tool() def get_greeting(name: str) - str: 根据名字返回一句问候 return f你好{name}这是来自 MCP 工具的响应。 if __name__ __main__: mcp.run(transportstdio)然后是 Client 骨架。核心是StdioServerParameters配置子进程启动命令stdio_client建立双向流ClientSession管理会话。这三层分工明确参数决定怎么拉起 Server流负责读写会话负责协议层交互。# client.py import asyncio import json import sys from contextlib import AsyncExitStack from typing import Optional from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import AsyncOpenAI from config import TAOTOKEN_CONFIG class MCPClient: def __init__(self): self.session: Optional[ClientSession] None self.exit_stack AsyncExitStack() self.client AsyncOpenAI( base_urlTAOTOKEN_CONFIG[base_url], api_keyTAOTOKEN_CONFIG[api_key], ) async def connect_to_server(self, server_script_path: str): server_params StdioServerParameters( commandpython, args[server_script_path], envNone, ) 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(\n已连接 Server可用工具:, [t.name for t in tools]) async def cleanup(self): await self.exit_stack.aclose()这段代码里有两个容易踩的点。第一command用python还是python3取决于你的环境Windows 上通常是pythonmacOS/Linux 可能是python3报FileNotFoundError时先查这里。第二args里传的是 Server 脚本路径用sys.argv[1]从命令行拿别写死。连接建立后list_tools()返回的是工具元数据包含 name、description、inputSchema。这三个字段后面要原样转成 OpenAI 的 function 格式description 写得好不好直接决定模型选不选得对工具。4. 验证请求一次 stdio 调用看到工具结果骨架搭好后补上处理查询的逻辑。process_query是整条链路的核心把工具描述转成 function calling 格式发给 LLM拿到 tool_calls 后执行再把结果回传循环直到模型不再请求工具。async def process_query(self, query: str) - str: messages [{role: user, content: query}] response await self.session.list_tools() available_tools [ { type: function, function: { name: tool.name, description: tool.description, parameters: tool.inputSchema, }, } for tool in response.tools ] response await self.client.chat.completions.create( modelTAOTOKEN_CONFIG[model], messagesmessages, toolsavailable_tools, ) final_text [] message response.choices[0].message if message.content: final_text.append(message.content) while message.tool_calls: for tool_call in 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) final_text.append(f[调用工具 {tool_name}参数 {tool_args}]) messages.append({ role: assistant, tool_calls: [{ id: tool_call.id, type: function, function: { name: tool_name, arguments: json.dumps(tool_args), }, }], }) messages.append({ role: tool, tool_call_id: tool_call.id, content: str(result.content), }) response await self.client.chat.completions.create( modelTAOTOKEN_CONFIG[model], messagesmessages, toolsavailable_tools, ) message response.choices[0].message if message.content: final_text.append(message.content) return \n.join(final_text)再补一个交互循环和入口就能跑了。chat_loop负责读输入、调process_query、打印结果输入 quit 退出。async def chat_loop(self): print(\nMCP Client 已启动输入问题或 quit 退出。) while True: try: query input(\nQuery: ).strip() if query.lower() quit: break response await self.process_query(query) print(\n response) except Exception as e: print(f\n出错: {e}) async def main(): if len(sys.argv) 2: print(用法: uv run client.py server_script_path) sys.exit(1) client MCPClient() try: await client.connect_to_server(sys.argv[1]) await client.chat_loop() finally: await client.cleanup() if __name__ __main__: asyncio.run(main())运行命令把 Server 脚本路径传进去uv run client.py demo_server.py启动后你会先看到工具列表然后输入一句「帮我向 Alice 问好」。预期输出里会出现[调用工具 get_greeting参数 {name: Alice}]紧接着是模型基于工具结果生成的回答。看到这一行说明 stdio 链路、TaoToken 通道、function calling 三部分全部打通。如果你想让模型自己决定调用哪个工具可以再加一个工具比如返回当前时间的get_time然后问「现在几点」观察模型是否选对了工具。这一步能帮你确认 description 和 inputSchema 的写法是否清晰。5. 本篇常见错排查跑不通的时候按下面这几类对照基本能覆盖九成问题。第一类是子进程启动失败报FileNotFoundError或No such file or directory。原因通常是command写成了当前环境不存在的解释器或者 Server 脚本路径传错。先手动执行python demo_server.py确认脚本本身能跑再检查 Client 里的command和args。第二类是连接建立后卡住没有任何输出。多半是 Server 没有用 stdio 传输启动比如写成了mcp.run()默认走别的传输方式。确认 Server 入口是mcp.run(transportstdio)stdio 模式下 Server 不应该往 stdout 打日志否则会污染协议流。日志请走 stderr。第三类是 LLM 请求报 401 或 403。检查TAOTOKEN_API_KEY环境变量是否真的注入到了当前 shellecho $TAOTOKEN_API_KEY看一眼。base_url 必须是https://taotoken.net/api多一个斜杠或少一段路径都会导致 404。第四类是模型不调用工具直接自己编答案。这通常是工具 description 太模糊或者 inputSchema 的 required 字段没写对。把 description 写成一句明确的功能说明参数类型和必填项对齐模型的选择准确率会明显上升。第五类是tool_calls处理时报 JSON 解析错误。tool_call.function.arguments是字符串需要json.loads但模型偶尔会返回空字符串或非标准 JSON。加一层 try 兜底解析失败时把原始字符串传回去别让整个循环崩掉。第六类是消息历史拼错导致 400。role: tool的消息必须带tool_call_id且要和前面 assistant 消息里的id对应。顺序也不能乱assistant 的 tool_calls 消息必须在 tool 结果消息之前。提示stdio 模式下调试信息一律走 stderrprint到 stdout 会破坏 MCP 协议帧表现为 Client 端解析异常或直接挂起。6. 继续深入从验证到长期编码跑通第一个工具调用后你手里就有了一套可复用的骨架。接下来可以往两个方向走一是把 Server 换成真实业务工具比如查数据库、读文件、调内部 API二是把 Client 接到编辑器或 Agent 框架里让它成为长期运行的编码助手。如果你打算把 MCP Client 用在日常编码、Agent 编排这类长期场景建议了解一下 Coding Plan它更适合持续性的模型调用需求配置入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。只是想快速验证模型对话效果可以直接用模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 试几句。接入过程中如果遇到 Key 或通道相关的问题去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 检查配置协议细节和参数说明看接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里能看到调用记录排障时对着日志看比猜快得多。最后留一个实用习惯把 Server 脚本路径、模型名、base_url 都抽到 config 里Client 代码保持不动。这样你换模型、换工具、换环境时只改一处配置验证链路的那套代码可以一直复用。
返回列表