building-python-mcp-servers

构建 Python MCP 服务

MCP 服务会向大模型客户端（Claude Desktop、Claude Code 等）提供工具。 大模型是调用方，因此故障模式不同于普通库：错误必须可由机器读取， 每个输入都不可信，而且测试套件全部通过，往往也无法证明工具确实可用。本技能总结了 此类问题反复出现时的处理模式。

快速入门（FastMCP）

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def read_config(path: str) -> dict:
    """Read a config file. `path` must be absolute."""
    p = Path(path)
    if not p.is_absolute():
        return {"error": "path must be absolute"}
    if not p.exists():
        return {"error": f"no such file: {path}"}
    return {"data": p.read_text()}

if __name__ == "__main__":
    mcp.run()

错误约定：返回错误，不要抛出错误，并保持一致

未捕获的异常会以不透明的协议错误形式呈现给大模型，令其无法 推理。应返回形状可预测的结构化结果，并要求调用方检查。

• 选定一种错误格式，并在所有地方使用。 常见做法是使用带有 "error" 键的字典。请说明调用方必须检查此键。
• 批处理工具必须报告跳过项，不要将其忽略。 最常见的不一致是：单项工具会收集每项的错误，但同类的“对整个目录执行此操作”工具却会在文件加载失败时悄悄执行 continue。对于自动化流程，这会造成无法察觉的数据丢失。每个批处理工具都应返回结果，以及按项列出的 skipped/errors 列表。

@mcp.tool()
def validate_dir(path: str) -> dict:
    results, skipped = {}, []
    for f in Path(path).glob("*.md"):
        try:
            results[f.name] = _validate(f)
        except Exception as e:
            skipped.append({"file": f.name, "error": str(e)})  # never silently continue
    return {"results": results, "skipped": skipped}

类型陷阱：YAML 会自动将未加引号的 ISO 日期（date: 2025-06-15）解析为 datetime.date，而非 str 或 datetime.datetime。只处理 str/datetime 的验证器会对原生 YAML 日期产生误报。验证解析后的值时，应列出解析器实际可能生成的每一种类型。

封装子进程 / CLI

很多 MCP 服务都会调用其他工具的 Shell 命令。以下三类故障反复出现：

1. 非零退出时不要丢弃 stdout。 许多 CLI 有意以非零状态退出（例如检查器或变异测试工具报告问题），并将实际输出写入标准输出，同时 stderr 为空。若封装程序在 returncode != 0 时返回 f"Error: {result.stderr}"，就会向大模型报告一次成功运行，但内容却是空的 "Error: "。

def run_tool(args: list[str]) -> dict:
    r = subprocess.run(args, capture_output=True, text=True)
    return {                       # hand BOTH streams to the model; let it judge
        "returncode": r.returncode,
        "stdout": r.stdout,
        "stderr": r.stderr,
    }

2. 固定使用实际封装的版本，并验证子命令是否存在。 如果服务基于某工具的 2.x CLI 编写，而项目固定使用 3.x，那么服务会调用已不存在的子命令和标志——所有封装工具都会在运行时失效。请检查已安装版本的 --help，不要依赖记忆。

3. 安全解析参数。 使用 str.split() 拆分额外参数字符串，会破坏带引号且包含空格的参数——应使用 shlex.split()。绝不要将客户端提供的字符串插入 Shell 命令；应向 subprocess.run 传递 argv 列表（不要使用 shell=True）。如果工具接受目标或路径，请记住这是由大模型（或它读取的内容）选择的——应进行验证。

不要使用模块级全局状态（这会让服务无法测试）

在导入时解析 CLI 参数并将其存入模块全局变量 （WORKING_DIR、MAKEFILE_PATH、缓存等），会迫使每个测试都必须执行 del sys.modules["server"]，并在修改后的 sys.argv 下重新导入，只为重置状态——这种做法脆弱且容易出错。请将配置保存在对象中或传递配置，并通过工厂构造工具。

def build_server(config: Config) -> FastMCP:
    mcp = FastMCP("my-server")

    @mcp.tool()
    def do_thing(x: str) -> dict:
        return {"result": _work(x, config)}   # config captured, not global

    return mcp

这也能避免重复注册：如果模块级的“创建所有工具”循环 和 main() 内的相同循环同时存在，那么直接运行文件（如 Claude Desktop 执行的 uv run server.py）时，与通过控制台入口点运行时不同，会导致每个工具都注册两次。只在一个地方注册。

避免在协议事件循环中执行阻塞工作

不要假设框架会将同步工具函数移到工作线程中运行。 有些 FastMCP 运行时会在协议事件循环上直接调用这些函数。因此，耗时 五秒的 SQLite 查询、文件系统遍历、依赖项遍历或同步 HTTP 调用， 都可能阻塞 ping 操作以及同一时间内的所有其他请求。

将工具设为异步，并且只将阻塞操作移到线程中：

import asyncio

@mcp.tool()
async def find_dependents(item_id: int) -> dict:
    rows = await asyncio.to_thread(repository.find_dependents, item_id)
    return {"items": [row.to_dict() for row in rows]}

请留意连接的所有权。不要在事件循环线程上创建 SQLite 连接，然后将该 连接交给工作线程。在 repository.find_dependents 内部打开并关闭连接， 或者使用其并发约定明确允许跨线程传递的连接池／驱动程序。对共享的、 线程绑定的连接套上一层线程封装，只会把事件循环饥饿换成间歇性的 数据库错误。

测试响应能力，而不只是检查慢速工具的结果。启动一个故意阻塞的仓库调用， 在释放它之前调用一个轻量级工具（或协议 ping），并要求轻量级请求先完成：

slow = asyncio.create_task(call_tool("find_dependents", {"item_id": 42}))
await entered_worker.wait()

healthy = await asyncio.wait_for(call_tool("health", {}), timeout=0.2)
assert healthy == {"ok": True}

release_worker.set()
await slow

仅对慢速调用进行计时断言无法检测事件循环饥饿；真正的回归问题是独立的 协议流量无法继续推进。

采样是可选的客户端能力 — 在工具内部处理故障

即使工具本身已成功调用，也不保证 ctx.sample(...) 可用。连接的客户端可能 不支持采样，也可能在处理请求时由其采样处理器引发异常。这两种情况在框架层面 属于不同的故障模式，但在工具层面的结果相同：无法生成所请求的分析。

在采样边界内的工具中捕获异常，并将其转换为服务的常规错误格式。不要依赖 框架外层的异常包装；到那时，调用方收到的将是不透明的协议／工具错误，而不是 你所记录的行为约定。

async def sample_or_error(ctx: Context, prompt: str) -> dict:
    try:
        response = await ctx.sample(prompt)
    except Exception as exc:
        return {"error": f"sampling failed: {exc}"}
    return {"result": response.text or ""}

保持 try 代码块范围狭窄，避免将无关的编程错误误标为采样故障。明确测试这两种 边界情况：不支持采样的客户端，以及会引发异常的已配置采样处理器。如果工具承诺 返回空字符串或其他回退值，也要测试空的采样响应。

分发方式：单文件还是打包

MCP 服务通常以单文件形式启动（uv run server.py），因此有两类打包陷阱很容易在 未察觉的情况下引入：

• 模块／包名称冲突。 顶层同时存在 server.py 和 server/ 包目录时， import server 会解析到包（从而遮蔽模块），因此像 server:main 这样的控制台 入口找不到 main，并导致失败。只有直接运行文件才有效。请选择一种名称。
• 构建包含范围过窄。 类似 only-include = ["server.py"] 的构建配置只会生成 一个仅包含该文件的 wheel — 任何通过 pip install 安装的用户都会在 import server.analyzers 时遇到 ModuleNotFoundError，尽管 uv run server.py 在本地可正常运行。如果要发布包，请包含包本身及其数据文件。

对于 PEP 723 单文件服务，请在内联的 # /// script 标头中固定明确的版本，并且确保它们与 pyproject.toml 保持同步；仅作为传递依赖项存在的依赖（已导入但从未声明）会在中间依赖不再 提供它时立即导致故障。

测试：证明工具确实可用

模拟子进程／传输层并断言 argv 包含某些标记，会锁定被包装工具中可能并不存在的命令——测试套件仍会全部通过，但所有工具在运行时都已损坏。在这里，CI 通过并不意味着服务可用。

• 至少保留一个端到端调用真实被包装工具（或真实示例文件）的集成测试。
• 在名称冲突的情况下，单纯的 import server 冒烟测试毫无意义（它可能导入一个空包）。应断言工具能够运行并返回预期输出。
• 测试错误约定：格式错误的输入应返回你的 "error" 结构，批处理工具应填充 skipped。

将所有工具 I/O 视为不可信内容（提示词注入）

工具输入、文件内容，尤其是其他工具的描述，都可能受到攻击者影响，并流入模型的上下文。将此类文本反馈给第二次大模型调用的服务本身就是提示词注入攻击面——其输出仅供参考，不具权威性。不要赋予工具超出所需范围的文件系统／网络访问权限；验证路径，绝不要将工具输出视为可信指令。

MCP 服务检查清单

Contract:
- [ ] One consistent error shape; documented that callers check it
- [ ] Batch tools return a per-item skipped/errors list (never silent continue)
- [ ] Inputs validated (absolute paths, allowed types) before use
- [ ] Sampling failures (unsupported client and handler exception) normalized inside the tool

Subprocess:
- [ ] Both stdout and stderr returned; non-zero exit not assumed to be failure
- [ ] Pinned to the wrapped tool's actual version; subcommands verified
- [ ] shlex.split for arg strings; argv list (no shell=True)

Structure:
- [ ] No module-level CLI parsing / global state
- [ ] Tools registered in exactly one place
- [ ] Blocking database/filesystem/network work moved off the protocol event loop
- [ ] Concurrency test proves a lightweight request completes while a slow tool is blocked

Distribution & tests:
- [ ] No server.py / server/ name collision; build includes the whole package
- [ ] PEP 723 header deps pinned and synced with pyproject
- [ ] An integration test exercises a real tool (not just mocked argv)

了解更多

本技能基于 Will McGinnis 编写的 Python 高质量库开发指南。