ARTICLE DETAIL

资讯详情

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

MCP 协议开发指南:用 Python 与 BaseTool 构建 MCPServer 的完整实践

MCP 协议开发指南:用 Python 与 BaseTool 构建 MCPServer 的完整实践 1. 从零理解 MCP 协议Python 开发者为什么需要 MCPServerMCP 协议开发指南里最容易被忽略的一点是MCP 不是某个具体框架的名字而是一套让模型和外部工具对话的约定。你可以把它想成 USB-C 接口——模型是电脑工具是外设只要双方都遵守接口形状插上就能用。Python 在这个场景里扮演的是外设制造机的角色而 BaseTool 就是那个标准模具。我第一次接触 MCP 的时候脑子里全是问号为什么不能直接写个函数让模型调用后来踩过坑才明白直接调函数的问题在于参数格式、错误返回、并发控制全得自己管模型每次拿到的描述还不一样。MCP 把这些抽象成协议层工具只需要声明我叫什么、需要什么参数、返回什么剩下的交给 MCPServer 处理。这套东西适合谁如果你正在做 AI Agent、想让模型读取本地文件或调用内部 API又不想每次都手写胶水代码那 MCP 就是为你准备的。Python 生态里 BaseTool 提供了统一的工具基类你继承它、实现 execute 方法注册到 MCPServer 就能跑。整个过程不需要理解底层传输细节stdio 和 SSE 两种模式开箱即用。本文会带你从项目结构开始一步步搭出一个能运行的 MCPServer包含依赖清单、工具注册代码、本地启动命令和调用验证。所有代码都可以直接复制跑通之后你会对 MCP 协议开发有一个完整的体感。核心检索词先记住MCP 协议、Python、BaseTool、MCPServer这四个词贯穿全文。2. 环境准备与 TaoToken 接入前置BaseTool 工具基类依赖清单在写第一行工具代码之前得先把运行环境和模型接入这两件事理清楚。很多人卡在服务跑起来了但模型调不通问题往往出在接入配置上。我试过用 TaoToken 作为模型接入层它的好处是 Base URL 和 Key 的管理比较集中配合 MCP 的 stdio 模式调试起来很顺。先说 Python 环境。建议用 3.10 以上版本因为 BaseTool 里用到了async def和类型注解的新特性。虚拟环境用 venv 就行不需要 conda 那么重。依赖清单我整理成 requirements.txt你可以直接复制mcp1.0.0 pydantic2.0 httpx0.27.0 anyio4.0.0 python-dotenv1.0.0这里 mcp 是协议实现库pydantic 负责参数校验httpx 用于 SSE 模式下的网络请求anyio 提供异步运行时支持。装完之后用pip list确认一下版本避免 pydantic 1.x 和 2.x 混用导致的校验报错。接下来是 TaoToken 的接入前置。你需要准备三样东西Base URL、API Key、Model ID。Base URL 填https://taotoken.net/api注意这里不加任何多余路径。API Key 在控制台生成建议单独建一个项目级的 Key方便后续轮换。Model ID 根据你用的模型填比如 claude 系列或 gpt 系列具体以文档为准。把这三样写进.env文件不要硬编码到代码里TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的key TAOTOKEN_MODEL_ID你的模型ID然后在代码里用 python-dotenv 读取。这样做的好处是MCP 服务在 stdio 模式下启动时环境变量会随进程传递工具内部调用模型时直接读环境变量即可不需要额外传参。如果你还没生成 Key可以去 API Keys 页面操作生成后记得复制保存页面刷新后就看不到了。这一步做完你的 MCPServer 就有了大脑和手脚BaseTool 定义手脚的形状TaoToken 提供大脑的推理能力。两者通过 MCP 协议里的工具注册机制连接起来。下一节开始写真正的配置和代码。3. 可复制配置MCPServer 项目结构与工具注册代码项目结构我建议这样组织清晰且方便扩展mcp-demo/ ├── app/ │ ├── __init__.py │ ├── tool/ │ │ ├── __init__.py │ │ └── base.py │ ├── tools/ │ │ ├── __init__.py │ │ └── file_reader.py │ └── mcp/ │ ├── __init__.py │ └── server.py ├── .env ├── requirements.txt └── README.md先写 BaseTool 基类放在app/tool/base.py。这个类定义了所有工具必须实现的接口from abc import ABC, abstractmethod from typing import Any, Dict class BaseTool(ABC): name: str description: str abstractmethod def to_param(self) - Dict[str, Any]: 返回 JSON Schema 格式的工具描述 raise NotImplementedError abstractmethod async def execute(self, **kwargs) - Any: 执行工具逻辑返回结果 raise NotImplementedError def validate_params(self, params: Dict[str, Any]) - bool: schema self.to_param()[function][parameters] required schema.get(required, []) return all(k in params for k in required)注意to_param返回的结构必须符合 MCP 协议约定外层有 name、description、functionfunction 里有 description 和 parameters。parameters 用 JSON Schema 描述type 固定为 objectproperties 列出每个参数的类型和说明required 列出必填项。接着写一个具体工具 FileReaderTool放在app/tools/file_reader.pyfrom typing import Any, Dict from app.tool.base import BaseTool class FileReaderTool(BaseTool): name file_reader description 读取指定路径的文件内容 def to_param(self) - Dict[str, Any]: return { name: self.name, description: self.description, function: { description: 读取文件内容支持指定编码, parameters: { type: object, properties: { file_path: { type: string, description: 要读取的文件路径 }, encoding: { type: string, description: 文件编码默认 utf-8 } }, required: [file_path] } } } async def execute(self, file_path: str, encoding: str utf-8) - str: try: with open(file_path, r, encodingencoding) as f: return f.read() except Exception as e: return fError reading file: {str(e)}然后是 MCPServer 的注册逻辑放在app/mcp/server.pyimport asyncio from mcp.server import Server from mcp.server.stdio import stdio_server from app.tools.file_reader import FileReaderTool class MCPServer: def __init__(self): self.server Server(mcp-demo) self.tools {} def register_tool(self, tool): self.tools[tool.name] tool self.server.list_tools()(self._list_tools) self.server.call_tool()(self._call_tool) async def _list_tools(self): return [t.to_param() for t in self.tools.values()] async def _call_tool(self, name: str, arguments: dict): tool self.tools.get(name) if not tool: return {error: ftool {name} not found} if not tool.validate_params(arguments): return {error: missing required params} return await tool.execute(**arguments) async def run(self): async with stdio_server() as (read, write): await self.server.run(read, write, self.server.create_initialization_options()) if __name__ __main__: server MCPServer() server.register_tool(FileReaderTool()) asyncio.run(server.run())这段代码里register_tool把工具实例存进字典同时绑定 list_tools 和 call_tool 两个回调。stdio_server 负责标准输入输出的读写create_initialization_options生成握手参数。启动命令就是python -m app.mcp.server。如果你用的是 Claude Code 或 Cline 这类客户端配置片段长这样放在 settings.json 或对应的 MCP 配置里{ mcpServers: { mcp-demo: { command: python, args: [-m, app.mcp.server], env: { TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-你的key, TAOTOKEN_MODEL_ID: 你的模型ID } } } }三件套 Base URL、Key、Model ID 都在 env 里客户端启动子进程时会自动注入。这样工具内部如果要调用模型直接读环境变量就行。4. 验证请求与成功结果本地启动 MCPServer 并调用 file_reader配置写完之后最关键的一步是验证。很多人代码写对了但跑不起来问题出在启动方式或调用格式上。这一节给你完整的验证流程。先确认依赖装好在项目根目录执行pip install -r requirements.txt然后启动服务。注意 stdio 模式下服务不会打印启动成功之类的日志因为它等着标准输入。你可以先用一个简单的测试脚本模拟客户端调用import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters( commandpython, args[-m, app.mcp.server], ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具:, [t.name for t in tools.tools]) result await session.call_tool( file_reader, {file_path: README.md} ) print(读取结果:, result.content) if __name__ __main__: asyncio.run(main())把这段存成test_client.py运行python test_client.py。如果一切正常你会看到类似输出可用工具: [file_reader] 读取结果: [TextContent(typetext, text# MCP Demo\n...)]看到可用工具里有 file_reader说明工具注册成功看到读取结果里有文件内容说明 execute 方法被正确调用。这两个信号缺一不可。如果你想用 TaoToken 的模型对话能力来驱动这个工具可以在客户端侧把模型接进来。模型对话入口可以帮你快速验证模型是否能正确识别工具描述并生成调用参数。实测下来模型看到to_param返回的 JSON Schema 后能准确生成{file_path: config.txt}这样的参数不需要额外提示词工程。再补充一个 SSE 模式的验证方式。如果你想让服务以 HTTP 方式暴露把 run 方法改成 SSE 传输from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route sse SseServerTransport(/messages) async def handle_sse(request): async with sse.connect_sse(request.scope, request.receive, request._send) as streams: await server.server.run(streams[0], streams[1], server.server.create_initialization_options()) app Starlette(routes[Route(/sse, endpointhandle_sse)])然后用uvicorn app.mcp.server:app --port 8000启动客户端连http://localhost:8000/sse即可。stdio 适合本地调试SSE 适合远程部署按需选择。验证通过后你的第一个 MCPServer 就算跑通了。接下来是排错环节这些错误我基本都遇到过。5. 本篇常见错误排查401、local proxy failed 与 reading choices 报错即使代码一字不差环境差异也会导致各种报错。这一节按真实错误信息来对照排查覆盖 401、local proxy failed、reading choices、OAuth 这几类高频问题。401 Unauthorized这个最直接Key 不对或没传。检查.env里的TAOTOKEN_API_KEY是否以sk-开头是否有多余空格。如果你在 MCP 配置的 env 里写 Key注意 JSON 里不能有换行。还有一种情况是 Key 过期了去控制台重新生成一个。401 不会因为代码逻辑改变而消失一定是凭证问题。local proxy failed这个报错通常出现在客户端启动 MCP 子进程时环境变量没正确传递。比如你在.env里配了 Base URL但 MCP 配置的 env 块里没写子进程读不到就报这个。解决办法是把三件套 Base URL、Key、Model ID 都写进 MCP 配置的 env 里不要依赖父进程的环境变量继承。另外检查command路径如果用python找不到换成绝对路径如/usr/bin/python3。reading choices 报错完整信息一般是Error reading choices: ...出现在模型返回格式不符合预期时。MCP 工具调用要求模型返回结构化的 tool_calls如果模型返回的是纯文本解析就会失败。检查你的 Model ID 是否支持 function calling不支持的话换一个。另外确认to_param返回的 parameters 是合法 JSON Schemarequired 数组里的字段名和 properties 里的键名要完全一致大小写都不能错。OAuth 相关报错如果你在客户端配置里启用了 OAuth 流程但没配回调地址会卡在授权环节。MCP 的 stdio 模式不需要 OAuthSSE 模式如果服务端要求鉴权才需要。排查时先确认你的传输模式stdio 下出现 OAuth 报错说明配置串了。把 MCP 配置里的 auth 相关字段删掉只保留 command、args、env 三件套。工具注册失败表现是list_tools返回空列表。检查工具类是否正确继承 BaseToolto_param是否返回了 name 字段name 是否重复。如果两个工具同名后注册的会覆盖先注册的但 list 里只显示一个。参数校验不通过validate_params返回 False说明必填参数没传。对照to_param里的 required 数组确认调用时参数名拼写一致。JSON Schema 里写的是file_path调用时传filePath就会失败。执行超时工具里如果有阻塞操作比如读大文件或网络请求会卡住整个 stdio 通道。给 execute 加超时控制import asyncio async def execute(self, file_path: str, encoding: str utf-8) - str: try: return await asyncio.wait_for(self._read(file_path, encoding), timeout10) except asyncio.TimeoutError: return Error: execution timeout排错的核心思路是分层先确认进程能启动再确认工具能注册再确认参数能校验最后确认执行能返回。每一层都有对应的错误信息对照上面的清单基本能定位。6. 语义一致 CTA把 MCPServer 接入长期编码工作流跑通第一个 MCPServer 只是起点。真正让 MCP 协议开发产生价值的地方是把它接入日常的编码和 Agent 工作流。当你有了稳定的工具注册机制就可以把文件读写、命令执行、API 调用都封装成 BaseTool 子类让模型按需调用。如果你打算长期做编码类 AgentCoding Plan 提供了更集中的额度管理和模型调度适合把 MCP 工具链跑在生产环境。接入方式和本文的 stdio 配置一致把 Base URL 和 Key 换成 Coding Plan 对应的即可。对于需要频繁验证模型行为的场景模型对话入口可以快速测试工具描述是否被正确理解。你可以在里面粘贴to_param的返回看模型生成的调用参数是否符合预期这比反复改代码重启服务高效得多。所有接入细节和参数说明都在接入文档里遇到配置问题先查文档再排查。API Key 的管理在 API Keys 页面建议按项目分 Key方便追踪调用来源。最后留一个实用技巧把 MCP 服务的启动命令写进 Makefilemake mcp一键启动make test跑验证脚本。工具多了之后用register_tool批量注册别一个个手写。BaseTool 的抽象价值就在于新增工具只需要关注 execute 里的业务逻辑协议层的事交给 MCPServer。
返回列表