1. 为什么我要手写一个 MCP clientMCP 全称 Model Context Protocol简单说就是一套让大模型能伸手去调用外部工具的协议。你可以把它理解成 USB-C 接口大模型是电脑MCP server 是各种外设只要都遵守这个接口标准插上就能用。市面上 Cherry Studio、Claude Desktop、Cline 这些工具已经能直接挂载别人写好的 MCP server但它们的操作入口都在 UI 上点几下就完事作为程序员总觉得少了点什么——我想让 MCP 调用变成代码里的一行函数嵌进自己的脚本、定时任务或者后端服务里。这篇就聚焦一件事用 Python 手写一个最小可用的 MCP client让它能同时连上远程 SSE 服务和本地 stdio 服务再通过 TaoToken 的统一 Key 把大模型这一环的鉴权也收拢到一处。适合已经跑通过 MCP server、想进一步用程序调用的同学也适合刚接触 MCP、想看清大模型到底怎么决定调哪个工具的读者。整条链路我会给出可复制的骨架代码、config.toml 配置示例以及一次完整的连接与工具调用验证步骤跟着敲一遍就能跑通。2. TaoToken 前置把大模型鉴权收拢成一个 Key写 MCP client 时最容易乱的地方不是协议本身而是大模型从哪来、Key 怎么管。如果每个 server 或每个模型都配一套 Key代码里到处是环境变量换模型就得改一堆地方。我的做法是让大模型请求统一走 TaoToken 的 API 通道官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。它对外暴露的是 OpenAI 兼容风格的接口所以 LangChain 里那些 ChatOpenAI 之类的封装基本不用改只把 base_url 指过去、Key 换成 TaoToken 的就行。具体操作上先去控制台创建一个 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 的管理页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。拿到 Key 之后不要硬编码进代码写进环境变量或者 config.toml后面 client 骨架里我会用配置读取的方式引用。这样做的直接好处是MCP server 那边完全不用关心大模型是谁client 这边只认一个 Key模型想从 Qwen 换成别的改一行配置即可。注意TaoToken 在这里承担的是大模型 API 的统一入口角色MCP server 的连接方式SSE / stdio跟它无关两者是并行的两条链路别混在一起理解。3. 可复制配置config.toml 与 client 骨架代码先看配置文件。我把大模型和 MCP server 的信息都放进 config.tomlclient 启动时读一次后面加 server 只改这个文件。# config.toml [llm] base_url https://taotoken.net/api api_key sk-你的TaoTokenKey model gpt-4o-mini [mcp_servers.weather] transport sse url http://127.0.0.1:8000/sse [mcp_servers.math] transport stdio command python args [mcp_local_server.py]这里 weather 是远程 SSE 服务math 是本地 stdio 服务跟上篇 server 实现对应。接下来是 client 骨架核心思路是用 langchain_mcp_adapters 的 MultiServerMCPClient 一次性把多个 server 的 tools 拉出来再把这些 tools 交给大模型做 tool calling。# mcp_client.py import asyncio import tomllib from typing import Any from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_mcp_adapters.client import MultiServerMCPClient def load_config(path: str config.toml) - dict: with open(path, rb) as f: return tomllib.load(f) def build_llm(cfg: dict) - ChatOpenAI: llm_cfg cfg[llm] return ChatOpenAI( base_urlllm_cfg[base_url], api_keyllm_cfg[api_key], modelllm_cfg[model], temperature0, ) def build_mcp_config(cfg: dict) - dict: servers {} for name, item in cfg[mcp_servers].items(): servers[name] item return servers async def execute_workflow(user_input: str, cfg: dict) - Any: llm build_llm(cfg) mcp_config build_mcp_config(cfg) client MultiServerMCPClient(mcp_config) all_tools await client.get_tools() print(已加载工具:, [t.name for t in all_tools]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个智能助手尽可能调用工具回答用户问题), MessagesPlaceholder(variable_namechat_history, optionalTrue), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad, optionalTrue), ]) agent create_tool_calling_agent(llm, all_tools, prompt) executor AgentExecutor(agentagent, toolsall_tools) response await executor.ainvoke({input: user_input}) return response[output] if __name__ __main__: config load_config() print(asyncio.run(execute_workflow(北京天气怎样, config))) print(asyncio.run(execute_workflow(帮我计算 20 和 12 的乘积, config)))跟原版相比我把模型换成了 ChatOpenAI 指向 TaoToken 的 base_url这样 Key 就统一了配置从代码里抽到 config.toml加 server 不用动 Python。运行前记得先把 weather 那个 SSE server 起起来本地 math server 的脚本放在同目录。4. 验证请求一次完整的连接与工具调用配置和代码就位后按顺序验证。第一步确认 server 在跑python mcp_local_server.py python weather_server.py 第二步跑 clientpython mcp_client.py正常的话你会先看到工具列表被打印出来类似已加载工具: [get_weather, calculate]这说明 MultiServerMCPClient 已经把两个 server 的 tools 都拉到了。接着大模型会根据用户输入决定调哪个工具问北京天气怎样时它会选中 get_weather 并带上城市参数问20 和 12 的乘积时会选中 calculate。最终输出类似北京今天晴气温 18 到 26 摄氏度。 240如果工具没被调用、直接返回一段自然语言通常是 prompt 里的 system 提示不够明确或者模型本身 tool calling 能力弱。我实测下来把 temperature 设成 0、system 里写清尽可能调用工具命中率会明显提升。想单独验证模型这一环是否通可以打开模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 发一句带工具意图的话看返回结构里有没有 tool_calls 字段。5. 本篇常见错排查报错一Connection refused连不上 SSE。九成是 server 没起或者端口不对。先 curl 一下http://127.0.0.1:8000/sse能返回事件流才说明 server 正常。stdio 那边如果报 command not found检查 config.toml 里的 command 是不是当前环境的 python虚拟环境里要用绝对路径。报错二401 / invalid api key。这是大模型那一环的问题跟 MCP 无关。检查 config.toml 里 api_key 是不是 TaoToken 控制台创建的 Keybase_url 有没有写成https://taotoken.net/api注意结尾不要多加/v1之类具体以接入文档为准。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有各语言示例。报错三tools 列表为空。说明 client 连上了但没拿到工具。常见原因是 server 的 tools 注册函数没被触发或者 SSE 路径写错有的 server 是/sse有的是/mcp。打印all_tools长度先确认再回去看 server 日志。报错四模型不调工具只聊天。换一个 tool calling 支持更好的模型或者在 prompt 里加 few-shot 示例。LangChain 的 create_tool_calling_agent 依赖模型返回结构化 tool_calls纯文本模型是走不通的。报错五异步嵌套报错 asyncio.run() cannot be called from a running event loop。如果你在 Jupyter 里跑把asyncio.run换成await execute_workflow(...)或者用 nest_asyncio。脚本方式运行不会有这个问题。6. 把 MCP 调用变成你代码里的一个函数跑通之后你会发现MCP client 的本质就是把连 server、拉 tools、交给模型决策、执行工具这四步封装成一个 async 函数。之后你想做定时天气播报、想让 Agent 自动查数据库、想把工具调用嵌进 FastAPI 后端都只是在这个骨架上加东西。如果后面要长期跑编码类或 Agent 类任务可以考虑用 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 把额度固定下来避免临时 Key 额度不够打断链路。Claude Code 相关的接入方式在 https://taotoken.net/claudecode?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 也有说明思路和这里一致统一入口配置集中代码只认一个 base_url。
