首页 › tutorials › 造文件搜索MCP配调试测试

造文件搜索MCP配调试测试：FastMCP 全流程实战

概述

本教程带你用 Python 的 FastMCP 框架，从零构建一个本地文件搜索 MCP 服务器，并在真实开发链路中跑通它：写工具（Tools）、资源（Resources）、提示词（Prompts）三原语，用 MCP Inspector 调试 JSON-RPC 交互，写单元测试与端到端测试，最后接入 Claude Code 让模型真正调起你的搜索能力。适用人群：有 Python 基础、想让 Claude Code 长出"私有技能"的后端开发者与运维工程师。学完后你掌握的不是又一个域封装，而是可复用的 MCP 服务器开发流水线。

前置条件

为什么用 FastMCP 而不是直接写底层 SDK？FastMCP 1.0 在 2024 年被合并进官方 MCP Python SDK，现在由 Prefect 团队独立维护、每天下载量百万级，约 70% 的各语言 MCP 服务器跑在某个版本的 FastMCP 上。它的核心价值是：你只要写带类型注解的普通函数，JSON Schema、参数校验、协议处理全自动生成。

一、先把心智模型立起来

MCP（Model Context Protocol）本质上是"为 LLM 设计的 Web API"。一个 MCP 服务器对外暴露三类能力，称为三原语：

原语 类比 何时用 本教程实现
Tools（工具） POST 接口 / RPC 模型要执行动作、产生副作用、查询动态数据 search_files、read_snippet
Resources（资源） GET 接口 / 静态文件 模型要读取结构化数据，只读 search://stats 索引统计
Prompts（提示词） 可复用模板 固化高频交互模板，带参数注入 summarize_hits 命中汇总模板

一个请求的生命周期是这样的：

sequenceDiagram
    participant LLM as Claude（MCP 客户端）
    participant Host as Claude Code 宿主
    participant Srv as 你的 MCP 服务器（stdio）
    LLM->>Host: 我要调 search_files(query="TODO")
    Host->>Srv: tools/call（JSON-RPC over stdio）
    Srv->>Srv: FastMCP 校验参数 → 执行函数 → Context 上报日志
    Srv-->>Host: 返回结构化结果
    Host-->>LLM: 把结果喂回模型
    LLM-->>LLM: 基于结果继续推理/回答

记住一句话：类型注解就是契约，docstring 就是给模型看的说明书。这一条贯穿整个开发过程。

二、步骤 1：装环境、搭骨架

在工作目录初始化一个 uv 项目：

uv init file-search-mcp
cd file-search-mcp
uv add fastmcp
uv add --dev pytest pytest-asyncio

创建 server.py，先跑一个最小可运行的服务器：

# server.py
from fastmcp import FastMCP

mcp = FastMCP("FileSearch MCP")

@mcp.tool
def ping() -> str:
    """健康检查：返回 pong，用于验证服务器是否活着。"""
    return "pong"

if __name__ == "__main__":
    mcp.run()  # 默认 stdio 传输

操作说明：@mcp.tool 装饰器把普通函数注册成 MCP 工具。注意这里没有写任何 JSON Schema——def ping() -> str 这个签名本身就是 schema，docstring 会自动成为工具描述。

预期结果：用 FastMCP 自带的 dev 命令打开 Inspector 验证。

uv run fastmcp dev server.py

终端会提示打开浏览器（通常是 http://localhost:6274）。在 Inspector 左侧看到 FileSearch MCP，点 List Tools 能看到 ping，点 Run Tool 返回 "pong"，就说明骨架通了。

常见问题： - 报错 command not found: fastmcp：你没装上，或没用 uv run 前缀。FastMCP 安装后提供 fastmcp CLI，但 uv 项目里必须用 uv run fastmcp ...。 - Inspector 打开是空白页：Node 版本太低，升级到 18+。

三、步骤 2：实现搜索工具（Tools）

这是服务器的核心。我们要实现两个工具：按文件名 + 内容检索，以及读取某行附近的代码片段。

# server.py —— 工具部分
from pathlib import Path
from fastmcp import FastMCP
from fastmcp.server.context import Context

mcp = FastMCP("FileSearch MCP")

# 允许搜索的根目录，防止路径穿越越权
ALLOWED_ROOT = Path(__file__).resolve().parent.parent

def _safe_resolve(raw_path: str) -> Path:
    """把用户输入解析成绝对路径，并校验必须落在 ALLOWED_ROOT 内。"""
    p = (ALLOWED_ROOT / raw_path).resolve()
    if ALLOWED_ROOT not in p.parents and p != ALLOWED_ROOT:
        raise PermissionError(f"路径越界，禁止访问: {p}")
    return p

@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
async def search_files(
    query: str,
    directory: str = ".",
    file_glob: str = "*",
    max_results: int = 50,
    ctx: Context = None,
) -> list[dict]:
    """在目录下递归搜索包含 query 的文件，同时支持按文件名 glob 过滤。

    Args:
        query: 要搜索的关键词，大小写不敏感。
        directory: 相对项目根的搜索起点目录，默认当前目录。
        file_glob: 文件名通配符，如 "*.py" 只搜 Python 文件。
        max_results: 最多返回多少条命中，默认 50。

    Returns:
        命中列表，每条含 path / line / snippet / matched_text。
    """
    root = _safe_resolve(directory)
    await ctx.info(f"开始搜索 query={query!r} under {root}")
    hits = []
    needle = query.lower()
    scanned = 0
    for path in root.rglob(file_glob):
        if not path.is_file():
            continue
        scanned += 1
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except (OSError, UnicodeDecodeError):
            continue
        for lineno, line in enumerate(text.splitlines(), start=1):
            if needle in line.lower():
                hits.append({
                    "path": str(path.relative_to(ALLOWED_ROOT)),
                    "line": lineno,
                    "snippet": line.strip()[:200],
                })
                if len(hits) >= max_results:
                    await ctx.info(f"达到上限 {max_results}，停止扫描")
                    return hits
    await ctx.info(f"扫描 {scanned} 个文件，命中 {len(hits)} 处")
    return hits

逐点说明：

  1. annotations：readOnlyHint=True 告诉客户端"这个工具不改环境"，openWorldHint=False 表示"封闭域"（只在本地文件系统内）。这是给客户端做权限/审批决策的元数据，Claude Code 会据此减少不必要的确认弹窗。
  2. ctx: Context = None：Context 通过类型注解自动注入，FastMCP 会把它从对外 schema 里剔除——客户端永远看不到 ctx 这个参数。Context 方法是异步的，所以函数用 async def。
  3. await ctx.info(...)：把日志回传给客户端，在 Inspector 和 Claude Code 里都能实时看到，是调试利器。
  4. _safe_resolve：MCP 服务器一旦被模型调用，输入就是"半不可信"的。路径穿越是 MCP 工具最常见的安全坑，必须用 resolve() + 父目录校验兜底，绝不能直接拼字符串。

预期结果：在 Inspector 里调 search_files(query="mcp")，右侧能看到分步的 info 日志，最终返回命中数组。

再加一个读片段的工具：

@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def read_snippet(
    path: str,
    line_number: int,
    context: int = 3,
    ctx: Context = None,
) -> str:
    """读取文件中指定行号附近若干行，带行号标注，供模型精读上下文。

    Args:
        path: 相对项目根的文件路径。
        line_number: 关注的行号（从 1 开始）。
        context: 上下各取多少行，默认 3。
    """
    p = _safe_resolve(path)
    if not p.is_file():
        raise FileNotFoundError(f"不是文件或不存在: {path}")
    await ctx.debug(f"读取 {p}:{line_number} ±{context}")
    lines = p.read_text(encoding="utf-8", errors="ignore").splitlines()
    start = max(1, line_number - context)
    end = min(len(lines), line_number + context)
    return "\n".join(f"{i:>4} │ {lines[i-1]}" for i in range(start, end + 1))

四、步骤 3：暴露只读资源（Resources）

资源用于"模型主动拉取的结构化数据"。我们把索引统计做成一个资源，再把单文件读取做成带参数的资源模板。

# server.py —— 资源部分
import json

@mcp.resource("search://stats")
def index_stats() -> str:
    """返回当前可搜索范围的统计信息：文件数、按扩展名分布。"""
    counts = {}
    total = 0
    for path in ALLOWED_ROOT.rglob("*"):
        if path.is_file() and ".git" not in path.parts:
            ext = path.suffix or "(无扩展名)"
            counts[ext] = counts.get(ext, 0) + 1
            total += 1
    return json.dumps({"total_files": total, "by_extension": counts}, ensure_ascii=False)

@mcp.resource("search://file/{path}")
def file_content(path: str) -> str:
    """读取指定相对路径文件的完整文本内容（只读）。"""
    p = _safe_resolve(path)
    return p.read_text(encoding="utf-8", errors="ignore")

关键点： - @mcp.resource("search://stats") 里的 URI 是资源的唯一标识，懒加载——只有客户端真的去 resources/read 这个 URI 时，函数才执行。 - search://file/{path} 是资源模板（URI Template，RFC 6570 风格），{path} 是占位符。客户端列出模板后可按需实例化。

五、步骤 4：固化高频提示词（Prompts）

Prompt 是"带参数的对话模板"。当用户在 Claude Code 里输入 /mcp 调用 prompt 时，模板内容会作为预设消息注入对话。

# server.py —— 提示词部分
from fastmcp.prompts import Message

@mcp.prompt
def summarize_hits(query: str, top_n: int = 10) -> list[Message]:
    """生成一段提示词，要求模型把刚才的搜索命中整理成结构化摘要。

    Args:
        query: 用户最初搜索的关键词。
        top_n: 只汇总前 N 条命中，默认 10。
    """
    return [
        Message(
            f"我刚才用关键词「{query}」在代码库里搜索，请调用 search_files "
            f"拿到前 {top_n} 条命中，然后按文件分组、给出每处的简要说明，"
            f"最后用一句话总结这个词在项目里的主要用途。"
        ),
    ]

返回 list[Message] 可以构造多轮对话（指定 role="assistant"），这里返回单条用户消息即可。

六、步骤 5：用 MCP Inspector 深度调试

Inspector 不仅是"能跑就行"的玩具，它是协议级调试器。开发时养成三轮自检习惯：

  1. List 阶段：点 Tools / Resources / Prompts 三个标签，确认数量和名字都对。少一个说明装饰器没生效（常见原因：函数被 if __name__ 包住、或导入路径错）。
  2. Schema 阶段：点开每个工具，看自动生成的输入 schema。max_results 应该是 integer 且有默认值 50；如果参数描述为空，说明你没写 docstring 的 Args: 段。
  3. Call 阶段：真实调用，重点看下方 Notifications / Logs 面板——这里能看到你用 ctx.info/debug 发出的全部日志，以及 JSON-RPC 的原始报文。
# 两种等价启动方式
uv run fastmcp dev server.py                          # FastMCP 封装版
npx @modelcontextprotocol/inspector uv run server.py  # 官方原版

常见问题： - 工具列表里没有新加的工具：server.py 里有语法错误，Inspector 启动时会在终端打印 traceback，回看终端即可。 - 调用一直转圈不返回：大概率是 rglob 扫到了超大目录或 node_modules。在 _safe_resolve 后加个 if ".git" in path.parts or "node_modules" in path.parts: skip。

七、步骤 6：单元测试 + 端到端测试

MCP 服务器也是普通 Python 代码，必须写测试。FastMCP 提供 Client，可以直接用内存传输连到服务器对象，省去起进程的开销，速度极快。

# test_server.py
import pytest
from fastmcp import Client
from server import mcp

@pytest.mark.asyncio
async def test_ping_tool_via_in_memory_client():
    async with Client(mcp) as client:
        tools = await client.list_tools()
        assert any(t.name == "ping" for t in tools)

        result = await client.call_tool("ping", {})
        # FastMCP 2.x 返回结构化内容
        assert result.data == "pong" or "pong" in str(result)

@pytest.mark.asyncio
async def test_search_files_returns_hits(tmp_path, monkeypatch):
    # 造一个临时目录当搜索根，避免污染真实项目
    import server
    (tmp_path / "a.py").write_text("print('hello mcp world')\n")
    (tmp_path / "b.txt").write_text("nothing here\n")
    monkeypatch.setattr(server, "ALLOWED_ROOT", tmp_path)

    async with Client(server.mcp) as client:
        res = await client.call_tool(
            "search_files",
            {"query": "mcp", "directory": "."},
        )
        paths = [h["path"] for h in res.data]
        assert "a.py" in paths
        assert "b.txt" not in paths

@pytest.mark.asyncio
async def test_read_snippet_respects_context(tmp_path, monkeypatch):
    import server
    f = tmp_path / "x.py"
    f.write_text("\n".join(f"line {i}" for i in range(1, 11)))
    monkeypatch.setattr(server, "ALLOWED_ROOT", tmp_path)

    async with Client(server.mcp) as client:
        res = await client.call_tool(
            "read_snippet", {"path": "x.py", "line_number": 5, "context": 2}
        )
        text = res.data if isinstance(res.data, str) else str(res.data)
        assert "line 3" in text and "line 7" in text
        assert "line 1" not in text  # 超出 context 范围

运行：

uv run pytest -v

预期结果：3 个用例全绿。Client(mcp) 这种"客户端连服务器对象"的写法，本质是 FastMCP 用内存 transport 绕过了 stdio——这正是它能毫秒级跑完测试的原因。把它当作 MCP 版的"直接调函数"，覆盖率可以拉得很高。

端到端测试更进一步：用真实子进程拉起服务器，再连 stdio，验证协议层没问题。FastMCP 的 Client 也能接 command:// 形式：

@pytest.mark.asyncio
async def test_e2e_subprocess():
    async with Client(["uv", "run", "server.py"]) as client:
        res = await client.call_tool("ping", {})
        assert "pong" in str(res)

八、步骤 7：接入 Claude Code

测试通过后，把服务器注册给 Claude Code。在项目根目录建 .mcp.json（项目级，可提交版本库，团队共享）：

{
  "mcpServers": {
    "file-search": {
      "command": "uv",
      "args": ["run", "--directory", ".", "server.py"]
    }
  }
}

或者用命令行（用户级，仅本机生效）：

claude mcp add file-search -- uv run server.py
claude mcp list          # 确认已注册
claude mcp get file-search   # 查看详情

进入 Claude Code，输入：

帮我搜一下代码库里所有用到 mcp 的地方，并读取第一处的上下文

模型会自主决定调用 search_files → 拿到命中 → 调用 read_snippet 读上下文 → 给出汇总。整个过程你不需要手写任何"先搜后读"的编排逻辑——工具描述（docstring）就是模型理解能力的说明书，写清楚它就能串起来。

常见问题： - claude mcp list 显示 server failed to start：命令路径不对。uv run server.py 必须在项目目录执行；用 .mcp.json 时务必加 --directory 指明工作目录。 - 工具被调用但报参数错误：schema 推断和你预期不一致。回 Inspector 看 schema，多半是类型注解写成了 list（无参数化）或 dict，FastMCP 没法生成精确 schema，改成 list[dict] 即可。

九、实战案例完整代码

把前面片段拼起来，下面是完整的 server.py，复制即可运行：

"""FileSearch MCP —— 本地文件搜索服务器（FastMCP 实现）"""
import json
from pathlib import Path

from fastmcp import FastMCP
from fastmcp.server.context import Context
from fastmcp.prompts import Message

mcp = FastMCP("FileSearch MCP")
ALLOWED_ROOT = Path(__file__).resolve().parent.parent

_SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "dist", "build"}


def _safe_resolve(raw_path: str) -> Path:
    p = (ALLOWED_ROOT / raw_path).resolve()
    if ALLOWED_ROOT not in p.parents and p != ALLOWED_ROOT:
        raise PermissionError(f"路径越界: {p}")
    return p


@mcp.tool
def ping() -> str:
    """健康检查，返回 pong。"""
    return "pong"


@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def search_files(
    query: str,
    directory: str = ".",
    file_glob: str = "*",
    max_results: int = 50,
    ctx: Context = None,
) -> list[dict]:
    """递归搜索包含 query 的文件，大小写不敏感。

    Args:
        query: 关键词。
        directory: 相对根目录，默认当前目录。
        file_glob: 文件名通配符，如 "*.py"。
        max_results: 最大命中数，默认 50。
    """
    root = _safe_resolve(directory)
    await ctx.info(f"搜索 query={query!r} under {root}")
    needle, hits = query.lower(), []
    for path in root.rglob(file_glob):
        if not path.is_file() or _SKIP_DIRS & set(path.parts):
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        for lineno, line in enumerate(text.splitlines(), 1):
            if needle in line.lower():
                hits.append({
                    "path": str(path.relative_to(ALLOWED_ROOT)),
                    "line": lineno,
                    "snippet": line.strip()[:200],
                })
                if len(hits) >= max_results:
                    return hits
    return hits


@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def read_snippet(
    path: str, line_number: int, context: int = 3, ctx: Context = None
) -> str:
    """读取文件指定行号附近若干行，带行号。

    Args:
        path: 相对根的文件路径。
        line_number: 关注行号（从 1 开始）。
        context: 上下行数，默认 3。
    """
    p = _safe_resolve(path)
    if not p.is_file():
        raise FileNotFoundError(path)
    await ctx.debug(f"读取 {p}:{line_number} ±{context}")
    lines = p.read_text(encoding="utf-8", errors="ignore").splitlines()
    start = max(1, line_number - context)
    end = min(len(lines), line_number + context)
    return "\n".join(f"{i:>4} │ {lines[i-1]}" for i in range(start, end + 1))


@mcp.resource("search://stats")
def index_stats() -> str:
    """返回可搜索范围的统计：文件数 + 按扩展名分布。"""
    counts, total = {}, 0
    for path in ALLOWED_ROOT.rglob("*"):
        if path.is_file() and not (_SKIP_DIRS & set(path.parts)):
            ext = path.suffix or "(无扩展名)"
            counts[ext] = counts.get(ext, 0) + 1
            total += 1
    return json.dumps({"total_files": total, "by_extension": counts},
                      ensure_ascii=False)


@mcp.resource("search://file/{path}")
def file_content(path: str) -> str:
    """读取指定相对路径文件的完整文本（只读）。"""
    return _safe_resolve(path).read_text(encoding="utf-8", errors="ignore")


@mcp.prompt
def summarize_hits(query: str, top_n: int = 10) -> list[Message]:
    """生成提示词，让模型把搜索命中整理成结构化摘要。

    Args:
        query: 原始搜索关键词。
        top_n: 汇总前 N 条，默认 10。
    """
    return [Message(
        f"用关键词「{query}」调 search_files 拿前 {top_n} 条命中，"
        f"按文件分组说明，再用一句话总结该词在项目中的主要用途。"
    )]


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

十、整体开发工作流

把前面 7 步串成一条可复用的流水线——以后做任何新 MCP 服务器，照这个套路走即可：

flowchart TD
    A[uv init + add fastmcp] --> B[写函数 + 类型注解 + docstring]
    B --> C[加 annotations<br/>readOnly/destructive/idempotent]
    C --> D[fastmcp dev 开 Inspector<br/>List→Schema→Call 三轮自检]
    D --> E{协议/数据对吗?}
    E -- 否 --> B
    E -- 是 --> F[pytest 内存 Client<br/>单测 + 子进程 E2E]
    F --> G{测试绿?}
    G -- 否 --> B
    G -- 是 --> H[.mcp.json / claude mcp add<br/>接入 Claude Code]
    H --> I[真实对话验证<br/>模型自主串工具]
    I --> J{效果达标?}
    J -- 否 --> K[优化 docstring 措辞<br/>调整 annotations]
    K --> I
    J -- 是 --> L[🚀 完成]

    style D fill:#fef3c7
    style F fill:#dbeafe
    style H fill:#dcfce7

注意虚线反馈环：调试反馈到代码、测试反馈到代码、真实效果反馈到 docstring/annotations——这是 MCP 开发和普通后端开发最大的区别，因为你的"接口文档"直接影响模型能不能正确用上能力。

十一、进阶要点与避坑清单

  1. Context 只在单次请求内有效：每次请求拿到的是新 Context，跨请求存状态要用 ctx.session 或外部存储（Redis/DB）。别在函数外缓存 Context。
  2. 同步函数自动进线程池：def 工具会被丢进线程池跑，不阻塞事件循环；但 I/O 密集场景仍推荐 async def，更高效。
  3. 资源是只读的：别在 @mcp.resource 里做写操作。需要写，就用工具。
  4. 输入永远是半不可信的：模型可能传奇怪参数（空字符串、超长字符串、负数）。对 max_results 加 min(max_results, 500) 上限，对 query 做 [:200] 截断，对路径做穿越校验——这三道防线能让服务器在生产里稳得多。
  5. 传输层选择：本教程用默认的 stdio（本地、零配置，适合 Claude Code 这种本地宿主）。若要做远程多租户，切 Streamable HTTP：mcp.run(transport="http", port=8000)，再配认证。
  6. 版本钉死：FastMCP 仍在快速演进，生产环境 uv add fastmcp==<具体版本>，别用浮点版本。

延伸阅读

参考来源

  1. MCP Python SDK（官方）— GitHub README — 官方 Python SDK，含 v2 的 MCPServer/Client 与 mcp dev Inspector 启动方式
  2. FastMCP — GitHub（PrefectHQ/fastmcp） — 本教程所用框架，"70% 的 MCP 服务器跑在某个版本的 FastMCP 上"
  3. FastMCP 官方文档（gofastmcp.com） — Tools / Resources / Prompts / Context 的权威 API 参考
  4. Build an MCP server — Model Context Protocol 官方文档 — 三原语（Resources / Tools / Prompts）的概念定义
  5. FastMCP in Python: Build a Real MCP Server（2026 Guide） — 含 GitHub OAuth 集成与生产部署的实战博客

训练营适用性评估

[!tip] 51学通信知识星球 星主：爱卫生（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 星球定位：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 当前规模：5200 篇主题 / 990 个回答，198 元/年。 加入福利：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 AI 加持：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

51学通信知识星球海报-扫码加入

📎 本分类最新文章