1. 为什么我要自己写一个 MCP-ServerMCP-Server 是什么一句话它是把本地函数、脚本、数据库查询包装成 AI 能直接调用的“工具插座”。你写好一个 Python 文件Claude Desktop、Cursor、LangChain Agent 都能通过同一套协议调用它不用为每个客户端重写一遍适配层。适合谁适合手里已经有一堆 Python 脚本、想让 AI Agent 真正动手干活的人而不是只会在聊天框里生成文本。我最初的需求很土让 Claude 帮我查本地 SQLite 里的订单表再顺手算一下环比。结果发现 Claude Desktop 只能读文件不能执行我的业务逻辑。于是我去翻 MCP 协议发现它本质上是 JSON-RPC over stdioServer 端暴露 tools/resources/prompts 三类能力Client 端负责发现和调用。听起来简单但第一次跑通花了整整一个下午坑集中在三处stdout 被 print 污染、Schema 类型不兼容、异步工具里混了同步阻塞。这篇就按“最小可用闭环”来写先给一个能跑的 server.py 骨架再讲工具注册和参数校验然后分别接 Claude Desktop 和 LangChain最后用一次真实调用链路验证。全程 Python命令可复制报错可对照。你不需要先理解全部协议细节跟着把闭环跑通再回头补理论会快很多。需要说明的是MCP 的传输层有 stdio 和 HTTP SSE 两种。本地工具集成优先 stdio因为零网络配置、延迟低远程服务才考虑 SSE。本文主线是 stdio因为 Claude Desktop 和大多数本地 Agent 都走这条路。下面所有代码都在 Python 3.11 mcp SDK 上实测过。2. 前置准备TaoToken 接入与 Python 环境在写 Server 之前先把模型调用侧准备好。因为后面验证工具调用链路时需要一个能稳定调用 Claude 或兼容模型的入口。我用的方式是 TaoToken 的 API 网关它把模型调用统一成 OpenAI 兼容格式Base URL 填https://taotoken.net/apiKey 在控制台生成。这样 LangChain 那边不用改代码直接换 base_url 就能跑。第一步拿 Key。打开 https://taotoken.net/api-keys 登录后创建一个 API Key复制保存。注意这个 Key 只在创建时完整显示一次丢了就重新建。控制台地址是 https://taotoken.net/console 里面能看到用量和余额。第二步确认模型 ID。不同场景用的模型不一样验证工具调用链路用对话模型就够长期跑 Agent 建议用 Coding Plan 里的模型额度更划算。模型列表在文档里能查到接入文档地址是 https://taotoken.net/doc 。如果你用 Claude Code 做润色或代码生成可以看 https://taotoken.net/claude-code-anthropic 这个页面里面有专门的接入参数。第三步装 Python 依赖。我推荐用 uv比 pip 快很多# 创建项目目录 mkdir mcp-demo cd mcp-demo # 用 uv 初始化没有 uv 就先 pip install uv uv init # 安装 MCP SDK 和 HTTP 客户端 uv add mcp httpx pydantic # 如果不用 uv用 pip 也行 pip install mcp httpx pydantic装完后验证一下版本python -c import mcp; print(mcp.__version__)能打印出版本号就说明 SDK 装好了。这里有个细节mcp SDK 的版本迭代比较快不同小版本的 API 可能有差异。我实测用的是 1.x 系列如果你装到的是 0.xserver.tool()的写法可能不生效建议先升级到最新稳定版。环境变量方面把 TaoToken 的 Key 和 Base URL 写进.env或直接 export后面 LangChain 部分要用export TAOTOKEN_API_KEY你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api到这里前置就齐了一个能跑的 Python 环境、一个可用的模型调用入口、一个明确的模型 ID。接下来进入正题写 Server 骨架。3. 可复制的 MCP-Server 骨架与工具注册这一节是全文核心我给一个完整的server.py包含三个工具加法、读本地文件、查 SQLite。你可以直接复制运行再按自己的需求替换工具实现。先看完整代码再逐段解释。# server.py import asyncio import json import logging import os import sqlite3 import sys from datetime import datetime from typing import Optional import httpx from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import TextContent from pydantic import BaseModel, Field # 关键日志必须写 stderr不能写 stdout logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, streamsys.stderr, ) logger logging.getLogger(mcp-demo) server Server(demo-server) # ---------- 工具 1简单加法 ---------- server.tool() def add(a: int, b: int) - int: 两数相加用于验证工具调用链路 return a b # ---------- 工具 2读文件带路径安全检查 ---------- class ReadFileInput(BaseModel): path: str Field(description要读取的文件绝对路径) max_bytes: int Field(default4096, ge1, le65536, description最大读取字节数) server.tool() async def read_file(input: ReadFileInput) - list[TextContent]: 读取本地文件内容限制在允许目录内 allowed os.environ.get(ALLOWED_DIRS, /tmp).split(:) real os.path.realpath(input.path) if not any(real.startswith(d) for d in allowed): return [TextContent(typetext, textf拒绝访问{input.path} 不在允许目录内)] try: with open(real, r, encodingutf-8) as f: content f.read(input.max_bytes) return [TextContent(typetext, textcontent)] except FileNotFoundError: return [TextContent(typetext, textf文件不存在{input.path})] except Exception as e: return [TextContent(typetext, textf读取失败{type(e).__name__}: {e})] # ---------- 工具 3查 SQLite ---------- class QueryInput(BaseModel): db_path: str Field(descriptionSQLite 数据库文件路径) sql: str Field(description只读 SELECT 语句) limit: int Field(default50, ge1, le500) server.tool() async def query_sqlite(input: QueryInput) - list[TextContent]: 执行只读 SQL 查询返回 JSON 结果 if not input.sql.strip().lower().startswith(select): return [TextContent(typetext, text只允许 SELECT 查询)] try: conn sqlite3.connect(input.db_path) conn.row_factory sqlite3.Row cur conn.execute(input.sql) rows [dict(r) for r in cur.fetchmany(input.limit)] conn.close() return [TextContent(typetext, textjson.dumps(rows, ensure_asciiFalse, defaultstr))] except Exception as e: return [TextContent(typetext, textf查询失败{type(e).__name__}: {e})] # ---------- 资源服务器信息 ---------- server.resource(config://server/info) async def server_info() - str: return json.dumps({ name: demo-server, version: 1.0.0, time: datetime.now().isoformat(), }, ensure_asciiFalse, indent2) # ---------- 启动 ---------- async def main(): logger.info(MCP Server 启动中...) async with stdio_server() as (read_stream, write_stream): logger.info(已就绪等待 Client 连接) await server.run( read_stream, write_stream, server.create_initialization_options(), ) if __name__ __main__: asyncio.run(main())几个关键点必须说清楚。第一logging.basicConfig的streamsys.stderr不能省。MCP 用 stdin/stdout 传 JSON-RPC 消息任何写到 stdout 的 print 都会把协议帧冲乱Client 端会报解析错误。我踩过的坑就是调试时随手 print结果 Claude Desktop 直接连不上。第二工具参数用 Pydantic 模型比裸类型注解更可控。Field(description...)会生成 JSON Schema 里的描述模型看到描述才知道这个参数是干嘛的。ge/le做范围校验避免模型传个负数进来。如果你只用def add(a: int, b: int)SDK 也能自动生成 Schema但描述是空的模型容易猜错。第三异步工具里不要用同步阻塞调用。sqlite3本身是同步的小查询没问题但如果换成requests.get就会卡住事件循环。要发 HTTP 请求就用httpx.AsyncClient。这个区别在单次调用时看不出来一旦 Agent 并发调多个工具就会暴露。第四路径安全检查别省。read_file里用os.path.realpath解析真实路径再判断是否在允许目录内能挡住../../etc/passwd这类遍历。生产环境还要考虑符号链接这里给的是最小可用版本。写完保存为server.py先本地跑一下确认不报错python server.py如果卡住不动、没有输出说明它在等 stdin 输入这是正常的。按 CtrlC 退出即可。接下来把它注册到 Claude Desktop。4. 接入 Claude Desktop 与 LangChain 并验证调用Claude Desktop 的配置文件位置macOS 是~/Library/Application Support/Claude/claude_desktop_config.jsonWindows 是%APPDATA%\Claude\claude_desktop_config.json。没有就新建一个写入{ mcpServers: { demo-server: { command: python, args: [/绝对路径/mcp-demo/server.py], env: { ALLOWED_DIRS: /tmp:/Users/你的用户名/data, TAOTOKEN_API_KEY: 你的Key } } } }注意args里必须是绝对路径相对路径在 Claude Desktop 启动时的工作目录下会找不到文件。env里把允许目录和 Key 传进去Server 里用os.environ.get读。改完配置重启 Claude Desktop在输入框旁边能看到工具图标点开应该列出add、read_file、query_sqlite三个工具。验证调用链路在 Claude 里输入“用 add 工具算一下 17 加 25”。正常情况它会请求调用工具返回 42。如果没反应先看 Claude 的日志macOS 在~/Library/Logs/Claude/mcp*.log。常见问题是 Python 路径不对command里写python可能指向了系统自带的旧版本改成which python的绝对路径更稳。LangChain 侧接入。LangChain 有langchain-mcp-adapters这个包能把 MCP Server 的工具转成 LangChain Toolpip install langchain-mcp-adapters langchain-openai# agent_demo.py import asyncio import os from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate async def main(): client MultiServerMCPClient({ demo: { command: python, args: [/绝对路径/mcp-demo/server.py], transport: stdio, } }) tools await client.get_tools() print(可用工具:, [t.name for t in tools]) llm ChatOpenAI( modelclaude-3-5-sonnet, # 换成你实际可用的模型 ID base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY], ) prompt ChatPromptTemplate.from_messages([ (system, 你可以调用工具完成任务。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result await executor.ainvoke({input: 用 add 工具算 17 加 25}) print(结果:, result[output]) asyncio.run(main())跑这个脚本如果verboseTrue打开你能看到完整的调用链模型先输出 tool_callsLangChain 执行add(17, 25)把结果 42 回填给模型模型再生成最终回答。这就是最小可用闭环。模型 ID 按你 TaoToken 控制台里实际可用的填别照抄。5. 常见报错排查对照这一节按真实报错来。第一个401 Unauthorized。LangChain 侧报这个八成是api_key没读到或者 Key 复制时带了空格。检查os.environ[TAOTOKEN_API_KEY]是否为空以及 base_url 是不是https://taotoken.net/api末尾不要多加/v1网关会自己处理路径。第二个local proxy failed或连接被拒。这个通常出现在 Claude Desktop 启动 Server 时command指向的 Python 不存在或者args路径写错。把command改成which python的输出绝对路径args用ls确认文件存在。还有一种情况是虚拟环境没激活Claude Desktop 用的是系统 Python装依赖时记得在同一个解释器里装。第三个reading choices field报错。这是 OpenAI 兼容接口返回体里没有choices字段常见原因是 base_url 写错请求打到了非兼容端点。确认 base_url 是https://taotoken.net/api模型 ID 拼写正确。如果模型 ID 不存在网关可能返回错误结构LangChain 解析时就报这个。第四个OAuth 相关报错。如果你用的是 Claude Code 或某些需要 OAuth 的客户端报OAuth token expired之类去 https://taotoken.net/console 重新生成 Key或者检查客户端里的认证配置。Claude Code 的接入参数在 https://taotoken.net/claude-code-anthropic 有说明Base URL、Key、Model ID 三件套要填全。第五个工具调用没触发。模型回复了文本但没调工具通常是工具描述太模糊。把Field(description...)写具体比如“要读取的文件绝对路径”比“路径”好。另外确认模型本身支持 function calling部分小模型不支持工具调用换模型再试。第六个stdio 通信乱码或解析失败。回到第 3 节说的检查代码里有没有print写到 stdout。所有调试输出走logger或print(..., filesys.stderr)。这个坑我踩过两次第二次是因为某个第三方库内部 print 了版本信息只能把它静音。排查顺序建议先看 Server 能不能单独启动再用 MCP Inspector 连一下最后才查客户端配置。MCP Inspector 的用法是npx anthropic-ai/mcp-inspector python server.py它能列出工具并手动调用比在 Claude 里试快得多。6. 把闭环跑通之后可以做什么最小闭环跑通后下一步是把真实业务工具塞进去。我的做法是每个工具一个文件用server.py统一注册避免单文件膨胀。工具实现里只做参数校验和调用业务逻辑抽到单独的模块方便单测。这样 Server 层很薄改起来不心疼。模型侧的选择上验证阶段用对话模型就够长期跑 Agent 建议用 Coding Plan额度更耐用地址是 https://taotoken.net/coding-plan 。如果你要接 Claude Code 做代码类任务接入文档在 https://taotoken.net/doc 里面有完整的配置示例。API Key 管理在 https://taotoken.net/api-keys 控制台在 https://taotoken.net/console 。最后一个实用技巧给每个工具加超时。MCP 协议本身不强制超时但 Agent 调用时如果某个工具卡住整个链路会挂起。在工具内部用asyncio.wait_for包一层超时就返回错误文本模型能感知到并换策略。这个细节在 demo 里没加但上生产前一定要补。