本教程教你用 Python 的 FastMCP 框架,从零打造一个"多能力"MCP 服务器——同时暴露工具(Tool)、结构化输出(Pydantic)、只读资源(Resource),并通过 Context 回传日志与进度,再用 MCP Inspector 完成独立调试,最后以正确的 scope 接入 Claude Code。适合已会基础 Python、想给自己 Agent 做专属工具链的开发者;Go SDK、远程 OAuth、Elicitation 等主题不在此展开。
Annotated)pip + venv:用于隔离依赖、生成可被 Claude Code 调起的可执行入口npx)MCP(Model Context Protocol)是一个开放协议,定义了"客户端 ↔ 服务器"之间的三种原语:工具(可执行的函数)、资源(只读数据)、提示词(可复用模板)。你写的服务器叫 MCP Server,Claude Code / Claude Desktop / Cursor 这些宿主是 MCP Client。
官方为各语言提供了 SDK。本教程选 FastMCP,原因有三:
@mcp.tool() 装饰器就完成"函数 → 工具"的转换,自动从类型注解生成 JSON Schema,无需手写协议帧。modelcontextprotocol/python-sdk,底层走同一套协议实现,不存在"野生框架"风险。与 Go SDK 的区别:Go 偏"显式注册 + 强类型结构体",FastMCP 偏"装饰器 + 类型推断"。如果你上一轮做的是 Go,这一轮正好补齐 Python 侧的肌肉记忆。
预期认知:读完这一步,你应能说清"工具是动词、资源是名词、Context 是服务器对客户端说话的渠道"。
# 1. 建项目目录
mkdir repo-doctor-mcp && cd repo-doctor-mcp
# 2. 用 uv 初始化(推荐,能生成确定性入口)
uv init --python 3.12
uv add fastmcp
# 3. 验证安装
uv run python -c "import fastmcp; print(fastmcp.__version__)"
预期结果:终端打印 FastMCP 版本号(当前 v3.x 系列),说明依赖就绪。
在项目根目录创建 server.py:
from fastmcp import FastMCP
mcp = FastMCP("repo-doctor")
@mcp.tool()
def ping() -> str:
"""健康检查,返回固定字符串。"""
return "pong"
if __name__ == "__main__":
mcp.run() # 默认 stdio 传输,正是 Claude Code 本地服务器所需的
先确认它能起来。stdio 服务器不会像 Web 服务那样持续打印,它启动后静默等待 JSON-RPC 消息,所以"看起来没反应"才是正常的。
常见问题
mcp.run() 报 RuntimeError: ... not in a request:你可能在交互式 REPL 里直接调了 run()。必须在 if __name__ == "__main__": 块内通过脚本入口运行。uv add fastmcp 很慢:换成国内 PyPI 镜像,uv add fastmcp --index-url https://pypi.tuna.tsinghua.edu.cn/simple。ping 只是冒烟测试。真正的工具要有参数、有文档、有明确返回类型。FastMCP 的关键约定是:函数签名即工具 Schema——参数名、类型、默认值、docstring 会被原样翻译给大模型。
from pathlib import Path
from typing import Annotated
from pydantic import Field
@mcp.tool()
def count_lines(
directory: Annotated[str, Field(description="要统计的目录绝对路径")],
extensions: Annotated[list[str] | None, Field(description="只统计这些后缀,如 ['py','md'];留空统计全部")] = None,
) -> dict[str, int]:
"""按后缀统计目录下所有文本文件的代码行数。"""
root = Path(directory)
result: dict[str, int] = {}
for p in root.rglob("*"):
if not p.is_file():
continue
ext = p.suffix.lstrip(".") or "noext"
if extensions and ext not in extensions:
continue
try:
with p.open("r", encoding="utf-8", errors="ignore") as f:
result[ext] = result.get(ext, 0) + sum(1 for _ in f)
except OSError:
continue
return result
为什么用 Annotated[..., Field(description=...)]:FastMCP 官方文档明确推荐这种写法——它让每个参数都带"自解释"说明,大模型在不看代码的情况下也能正确填参。普通 str 注解也能跑,但描述会缺失,模型容易瞎猜。
预期结果:当客户端列出工具时,会看到 count_lines 的 schema 里 directory 被标记为 required string,extensions 为 optional array,且每个字段都带中文描述。
常见问题
list[str] | None 在旧 Python 报错:这是 3.10+ 语法。低版本用 Optional[List[str]]。工具返回 dict 时,客户端拿到的只是松散文本。如果想让结果带结构化 schema(便于宿主做后处理或前端渲染),返回类型标注成 Pydantic 模型即可——FastMCP 会自动生成 outputSchema 并把数据放进结构化内容块。
from pydantic import BaseModel
class TodoItem(BaseModel):
file: str
line: int
text: str
severity: str # "high" | "medium" | "low"
@mcp.tool()
def find_todos(
directory: Annotated[str, Field(description="目标目录")],
marker: Annotated[str, Field(description="匹配标记,默认 TODO")] = "TODO",
) -> list[TodoItem]:
"""扫描源码中的 TODO/FIXME 标记,返回结构化清单。"""
root = Path(directory)
items: list[TodoItem] = []
for p in root.rglob("*.py"):
for i, line in enumerate(p.open("r", encoding="utf-8", errors="ignore"), 1):
if marker in line:
sev = "high" if "FIXME" in line or "XXX" in line else "medium"
items.append(TodoItem(file=str(p), line=i, text=line.strip(), severity=sev))
return items
预期结果:在 Inspector 里调用 find_todos,响应里除了文本,还有一个 structuredContent 字段,是严格的 {"items": [{file, line, text, severity}, ...]},宿主可以直接拿去渲染表格。
常见问题
list[BaseModel] 没生成 schema:确认 FastMCP 版本 ≥ 支持结构化输出的版本(v2.3+)。临时方案是包一层 class TodoList(BaseModel): items: list[TodoItem] 作为返回类型。X | Y union 返回类型导致崩溃:FastMCP 在 3.10+ union 注解的返回类型上曾有 bug(见 python-sdk issue #2591)。规避:返回类型用单一 Pydantic 模型,把多态收敛进模型字段。长任务(比如扫描十几万行的 monorepo)如果不回传进度,客户端会以为服务器卡死。FastMCP 用 Context 对象解决:把它作为参数声明出来,框架自动注入,你就能向客户端"喊话"。
from fastmcp import FastMCP, Context # 旧式类型注解注入,仍被官方支持
@mcp.tool()
async def list_functions(
directory: Annotated[str, Field(description="目标目录")],
ctx: Context,
) -> list[dict]:
"""列出所有顶层函数定义及所在行号,带实时进度。"""
root = Path(directory)
files = list(root.rglob("*.py"))
out: list[dict] = []
for idx, p in enumerate(files, 1):
await ctx.info(f"扫描 {p.name}({idx}/{len(files)})")
await ctx.report_progress(idx, len(files)) # 进度条
for i, line in enumerate(p.open("r", encoding="utf-8", errors="ignore"), 1):
if line.startswith("def ") or line.startswith("async def "):
out.append({"file": str(p), "line": i, "name": line.split("(")[0].replace("async def ", "").replace("def ", "").strip()})
await ctx.info(f"完成,共 {len(out)} 个函数")
return out
两个要点
info/debug/report_progress 等)都是 async,所以宿主函数要写成 async def。ctx.set_state / ctx.get_state(会话级)。预期结果:调用 list_functions 时,Inspector / Claude Code 的 /mcp 面板会实时滚动日志并显示进度百分比,而不是干等。
常见问题
ctx 注入成了 None:参数名随意,但类型必须是 Context(或 Context | None)。框架靠类型识别,不靠名字。from fastmcp.server.dependencies import get_context; ctx = get_context() 在请求上下文内任意位置取。注意它只能在请求处理期间调用,否则 RuntimeError。工具是"动作",资源是"被引用的数据"。把项目元信息做成资源,模型可以按 URI 拉取,而不必每次都调用工具。
import subprocess
@mcp.resource("repo://meta")
def repo_meta() -> dict:
"""暴露仓库元信息(分支、最近提交、统计)。只读。"""
root = Path(".")
def git(*args: str) -> str:
try:
return subprocess.run(["git", *args], capture_output=True, text=True, cwd=root).stdout.strip()
except Exception:
return ""
return {
"branch": git("rev-parse", "--abbrev-ref", "HEAD"),
"commit": git("rev-parse", "--short", "HEAD"),
"py_files": len(list(root.rglob("*.py"))),
}
还可以用资源模板让 URI 参数化:
@mcp.resource("repo://file/{path}")
def read_file(path: str) -> str:
"""按相对路径读取仓库内单个文件内容。"""
return Path(path).read_text(encoding="utf-8", errors="ignore")
预期结果:客户端能看到两条资源(一个静态 repo://meta,一个模板 repo://file/{path}),后者会自动展开成可枚举的资源列表或按需取。
常见问题
.. 路径穿越,防止读出敏感文件。写完先别急着接 Claude Code——先在MCP Inspector里单独把玩,能省掉"不知道是服务器错还是客户端错"的扯皮。
# Inspector 会拉起你的服务器作为子进程
npx @modelcontextprotocol/inspector uv run server.py
启动后终端会提示打开浏览器(默认 http://localhost:6274)。界面分两栏:左边是连接/配置,右边是工具、资源、提示词面板。
调试流程:
ping / count_lines / find_todos / list_functions 四个工具,各自带 schema。count_lines,填 directory,Run。看返回是否符合预期。list_functions 时,下方 Notifications 区会滚动 notifications/message(日志)和 notifications/progress(进度)。tools/call 的请求与响应,排查 schema 不匹配。预期结果:所有工具在 Inspector 里行为正确后,再接客户端几乎不会出问题。
常见问题
server.py,要用解释器入口,如 uv run server.py 或绝对路径 /full/path/.venv/bin/python server.py,否则 Node 找不到 Python。async def 且确实 await ctx.report_progress(...);同步函数里调 await 会直接语法错误。--server 6277 --client 6274 显式指定。调通后,把服务器注册到 Claude Code。核心命令:
claude mcp add --transport stdio repo-doctor -- uv run server.py
关于 --(双横线):这是硬性要求——它把 Claude 自己的选项(--transport、--env、--scope)和你服务器的命令隔开。-- 之后的一切原样交给服务器进程。忘掉 --,Claude 会把 server.py 当成自己的参数去解析而报错。
三种 scope 的取舍
| Scope | 生效范围 | 是否随仓库共享 | 存放位置 | 适用场景 |
|---|---|---|---|---|
local(默认) |
仅当前项目、仅你自己 | 否 | ~/.claude.json |
本地实验、含密钥的配置 |
project |
仅当前项目 | 是(随 Git) | 仓库根 .mcp.json |
团队共享工具链 |
user |
你的所有项目 | 否 | ~/.claude.json |
全局常用服务器 |
团队共享推荐 project scope:
claude mcp add --transport stdio --scope project repo-doctor -- uv run server.py
它会在仓库根生成 .mcp.json:
{
"mcpServers": {
"repo-doctor": {
"command": "uv",
"args": ["run", "server.py"]
}
}
}
首次拉取该仓库的同事,claude mcp list 里这条会显示 ⏸ Pending approval,交互式运行 Claude 时批准即可——这是项目 scope 的安全闸门。
环境变量:Claude Code 会给子进程注入 CLAUDE_PROJECT_DIR(项目根),你可在服务器内用 os.environ["CLAUDE_PROJECT_DIR"] 解析相对路径,不依赖工作目录。需要额外密钥时:
claude mcp add --env GITHUB_TOKEN=xxx --transport stdio repo-doctor -- uv run server.py
验证:
claude mcp list # 应看到 repo-doctor ✓ Connected
claude mcp get repo-doctor # 查看详细配置与 scope
然后在 Claude Code 里直接说:"用 repo-doctor 统计一下 src 目录的代码行数,再列出所有 TODO。" 模型会自动选 count_lines 和 find_todos。
常见问题
✗ Failed 或连不上:九成是路径或解释器问题。claude mcp get 看实际 command/args,复制到终端手动跑一遍;stdio 服务器有 stdout 杂物(如 print 调试)会污染 JSON-RPC,调试信息务必走 ctx.info 或 stderr。uv: command not found(在 Claude 子进程里):用 uv 安装的二进制要在 PATH 里,或直接用绝对路径 /home/you/.local/bin/uv。MCP_TIMEOUT=15000 claude;输出过大触发告警则 export MAX_MCP_OUTPUT_TOKENS=50000。repo-doctor 服务器把前面四块拼成一个文件 server.py,可直接 uv add fastmcp pydantic 后运行:
# server.py —— 代码仓库体检 MCP 服务器
import os, subprocess
from pathlib import Path
from typing import Annotated
from pydantic import BaseModel, Field
from fastmcp import FastMCP, Context
mcp = FastMCP("repo-doctor")
def _root() -> Path:
# Claude Code 注入的项目根;兜底用当前目录
return Path(os.environ.get("CLAUDE_PROJECT_DIR", "."))
# ---------- 模型 ----------
class TodoItem(BaseModel):
file: str
line: int
text: str
severity: str
# ---------- 工具 ----------
@mcp.tool()
def count_lines(
directory: Annotated[str, Field(description="目标目录")],
extensions: Annotated[list[str] | None, Field(description="只统计这些后缀")] = None,
) -> dict[str, int]:
"""按后缀统计代码行数。"""
root, result = Path(directory), {}
for p in root.rglob("*"):
if not p.is_file():
continue
ext = p.suffix.lstrip(".") or "noext"
if extensions and ext not in extensions:
continue
try:
with p.open(encoding="utf-8", errors="ignore") as f:
result[ext] = result.get(ext, 0) + sum(1 for _ in f)
except OSError:
continue
return result
@mcp.tool()
def find_todos(
directory: Annotated[str, Field(description="目标目录")],
marker: Annotated[str, Field(description="匹配标记,默认 TODO")] = "TODO",
) -> list[TodoItem]:
"""扫描 TODO/FIXME,返回结构化清单。"""
root, items = Path(directory), []
for p in root.rglob("*.py"):
for i, line in enumerate(p.open(encoding="utf-8", errors="ignore"), 1):
if marker in line or "FIXME" in line:
sev = "high" if "FIXME" in line else "medium"
items.append(TodoItem(file=str(p), line=i, text=line.strip(), severity=sev))
return items
@mcp.tool()
async def list_functions(
directory: Annotated[str, Field(description="目标目录")],
ctx: Context,
) -> list[dict]:
"""列出顶层函数,带实时进度。"""
files = list(Path(directory).rglob("*.py"))
out = []
for idx, p in enumerate(files, 1):
await ctx.report_progress(idx, len(files))
await ctx.info(f"扫描 {p.name}({idx}/{len(files)})")
for i, line in enumerate(p.open(encoding="utf-8", errors="ignore"), 1):
if line.startswith(("def ", "async def ")):
name = line.split("(")[0].replace("async def ", "").replace("def ", "").strip()
out.append({"file": str(p), "line": i, "name": name})
return out
# ---------- 资源 ----------
@mcp.resource("repo://meta")
def repo_meta() -> dict:
"""仓库元信息(分支、提交、Python 文件数)。只读。"""
def git(*a: str) -> str:
try:
return subprocess.run(["git", *a], capture_output=True, text=True, cwd=_root()).stdout.strip()
except Exception:
return ""
return {
"branch": git("rev-parse", "--abbrev-ref", "HEAD"),
"commit": git("rev-parse", "--short", "HEAD"),
"py_files": len(list(_root().rglob("*.py"))),
}
if __name__ == "__main__":
mcp.run()
一键上手脚本:
uv init --python 3.12
uv add fastmcp pydantic
# (把上面内容存为 server.py)
npx @modelcontextprotocol/inspector uv run server.py # 先调试
claude mcp add --transport stdio repo-doctor -- uv run server.py
claude mcp list
flowchart LR
A[编写 server.py<br/>FastMCP 装饰器] --> B[npx MCP Inspector<br/>独立调试工具/资源]
B -->|逐工具验证<br/>看协议帧| C{通过?}
C -->|否| A
C -->|是| D[claude mcp add<br/>选 local/project/user scope]
D --> E[.mcp.json 或 ~/.claude.json]
E --> F[Claude Code 客户端<br/>stdio 拉起子进程]
F --> G[模型自动调用工具<br/>读资源/收进度]
G --> H[CLAUDE_PROJECT_DIR<br/>解析项目相对路径]
sequenceDiagram
participant CC as Claude Code (Client)
participant SP as stdio 子进程
participant SR as FastMCP Server
CC->>SP: spawn: uv run server.py (注入 CLAUDE_PROJECT_DIR)
SP->>SR: mcp.run() 监听 stdin
CC->>SR: tools/list
SR-->>CC: 4 个工具 + schema
CC->>SR: tools/call list_functions
SR-->>CC: notifications/progress (10/100)
SR-->>CC: notifications/message "扫描 utils.py"
SR-->>CC: structuredContent {functions:[...]}
| 现象 | 根因 | 处理 |
|---|---|---|
| 工具不出现 | docstring 空 / 参数无描述 | 写清 docstring + Annotated 描述 |
| 返回 dict 无 schema | 返回类型未标 Pydantic 模型 | 用 BaseModel 标注返回类型 |
| 进度不刷新 | 函数是同步的 / 没 await |
改 async def,await ctx.report_progress |
| Inspector 连不上 | 命令缺解释器入口 | 用 uv run 或绝对路径 python |
Claude 端 Failed |
stdout 被 print 污染 | 调试输出走 ctx.info 或 stderr |
list[str]\|None 报错 |
Python < 3.10 | 升级或改 Optional[List[str]] |
py.sdk.modelcontextprotocol.io)—— FastMCP 底层实现,想绕过装饰器写"低层服务器"时查阅。github.com/modelcontextprotocol/inspector)—— 命令行参数、代理端口、CLI 模式细节。code.claude.com/docs/en/mcp)—— scope、超时、MAX_MCP_OUTPUT_TOKENS 等运维项。repo-doctor 换成"日志检索 / 发布检查"工具,强调 Context 进度与结构化输出在长任务里的价值。claude mcp add 三种传输、scope 层级、-- 分隔符与 CLAUDE_PROJECT_DIR 的权威说明。report_progress/info/会话状态的 API 细节。Annotated + Pydantic Field 自描述参数的最佳实践。[!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 等场景全覆盖。
