`PrefectHQ/fastmcp`：几行代码把 Python 函数变成 MCP 工具的框架¶

先用一句话说清楚它是什么：FastMCP 是一个让你几行代码就能把 Python 函数变成 MCP 工具的框架，不用手写协议细节。在构建基于 Python 智能体栈（如 LangChain, LlamaIndex, CrewAI）的应用时，直接编写原生 JSON-RPC 协议样板与处理 stdio 读写极其繁琐。FastMCP 借鉴了 FastAPI 的设计哲学，引入了声明式装饰器注册与自动化类型反射机制（Type Reflection），大幅降低了创建 MCP Server 的门槛。

一个容易混淆的历史遗留问题：为什么会有两个"FastMCP"？ 作者 Jonathan Lowin 最早在自己名下的 jlowin/fastmcp 仓库里开发了 FastMCP 1.0，2024 年这套最基础的 Server 构建能力被合并进了官方 MCP Python SDK，成为其内置的精简模块 mcp.server.fastmcp（有时被称为"FastMCP 1.0"）。与此同时，独立项目并未停止演进：它继续发展出 2.0/3.0，新增了官方 SDK 没有的 Client、Server 组合与代理（Proxy）、OpenAPI/FastAPI 一键转换、部署与鉴权（Auth）、内置测试工具等大量能力，仓库也随之从 jlowin/fastmcp 迁移到了 Prefect 官方组织名下的 PrefectHQ/fastmcp（GitHub 会自动重定向旧地址）。本文讲的是功能更完整、社区使用更广泛的这个独立 FastMCP 项目（from fastmcp import FastMCP），而不是官方 SDK 内置的精简版本（from mcp.server.fastmcp import FastMCP）——两者 API 相似但不完全相同，行文中会统一使用独立项目的写法。

0. 最小示例：一个函数就是一个工具¶

FastMCP 最基础的用法只需要三步——建实例、加装饰器、启动：

from fastmcp import FastMCP

mcp = FastMCP(name="MyFirstServer")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

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

就这几行代码，add 这个普通 Python 函数就变成了一个大模型可以调用的 MCP 工具——不需要手写 inputSchema，不需要手写 JSON-RPC 报文格式。FastMCP 在背后做的事情，就是把这个函数“翻译”成大模型能理解的工具描述。它是怎么做到的？

1. 核心设计思路：自动读函数签名，帮你生成 Schema¶

FastMCP 的核心价值在于：它会自动读你写的 Python 函数签名和 Docstring，帮你生成 MCP 协议需要的工具描述（Schema），不用你手写一份 JSON Schema。具体落地是一套基于 Python 类型提示（Type Hints）与 Docstrings 的反射引擎：

  +---------------------------------------------------------+
  |              FastMCP 声明式 Python 业务函数               |
  |                                                         |
  |   @mcp.tool()                                           |
  |   def query_db(user_id: int, date: str) -> str:         |
  |       """Query local DB for audit log."""              |
  +----------------------------+----------------------------+
                               | 
                               | Python Inspect 反射与 Pydantic 转换
                               v
  +---------------------------------------------------------+
  |              自动化生成 MCP JSON-RPC 2.0 Schema           |
  |                                                         |
  |   "name": "query_db",                                   |
  |   "description": "Query local DB for audit log.",       |
  |   "inputSchema": { "properties": { "user_id": { ... } } }|
  +---------------------------------------------------------+

具体到实现层面，这套反射链路分三步走：

1.1 反射与转换链路机理：¶

参数签名提取 (Signature Inspect)：FastMCP 使用标准库中的 inspect.signature 解析函数的形参定义，提取 user_id: int 与 date: str 的强类型信息。
Docstring 抽取：解析 PEP-257 规范的 Docstring，第一行提取为 Tool 的 description，后续参数说明转换为参数层面的 description 描述项，直接送入模型以引导其决策。
Pydantic 验证器自动合成：FastMCP 在内部动态合成一个 Pydantic 数据模型（Data Model）。当 Client 发来 tools/call 请求时，SDK 首先用该 Pydantic 模型对参数执行强校验（Pydantic Validation）。校验失败直接返回标准 JSON-RPC -32602 (Invalid params) 错误，保护业务逻辑免受脏输入冲击。

2. 除了工具，还能声明“数据源”和拿到运行时环境信息¶

Tool 之外，MCP 协议还定义了 Resources——大模型可以读取的数据源，比如本地日志流、系统状态。FastMCP 同样用声明式的方式把这类数据源暴露出去，同时还提供了一个拿到当前会话环境信息的入口：

路径模板匹配：通过 @mcp.resource("logs://{server_name}/audit") 语法，在大模型请求读取对应 URI 时，路径参数 server_name 会自动被捕获并传给 Python 业务函数，实现海量异构数据源的声明式映射。
运行时上下文注入 (Context Injection)：可在工具函数中声明一个 ctx: Context 类型的特殊参数。运行时会自动将当前的会话上下文（含 Session ID、采样钩子、客户端信息）注入此对象，方便工具在执行时获取环境状态与记录安全的审计追踪。

3. 基于 FastMCP 的声明式智能体服务（Python 实践）¶

把工具注册、资源注册和上下文注入放在一起，一个接近生产环境的 MCP Server 大致是下面这样：

import logging
from typing import Annotated
from pydantic import Field
from fastmcp import FastMCP, Context

# 1. 初始化 FastMCP 服务实例，声明 Stdio / Streamable HTTP 通道配置
# FastMCP 会自动处理底层的 JSON-RPC 帧封装
mcp = FastMCP(
    name="LogAuditService",
    version="1.0.0",
)

# 配置内部日志流，严禁写往 sys.stdout，统一重定向至 stderr
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger("mcp")

# 2. 声明式工具注册，结合 Pydantic Field 强化参数描述边界
@mcp.tool()
def fetch_system_logs(
    limit: Annotated[int, Field(default=10, description="Max log rows to return, max 100.")] = 10,
    ctx: Context = None
) -> str:
    """
    Fetch the latest system audit logs from local runtime.

    :param limit: Maximum rows to return.
    """
    # 3. 动态使用运行时上下文注入，记录安全的审计追踪
    if ctx:
        logger.info(f"Tool called by client: {ctx.request_context}")

    if limit > 100:
        limit = 100

    logger.info(f"Fetching logs with limit: {limit}")
    return f"LOG: 2026-05-20 - User 42 successfully fetched auth token. (Rows returned: {limit})"

# 4. 动态资源注册，使用 Path Pattern 进行动态路由
@mcp.resource("system://metrics/{metric_name}")
def get_dynamic_metrics(metric_name: str) -> str:
    """
    Get dynamic system metrics by name.

    :param metric_name: Name of the metric (e.g. cpu, memory, disk).
    """
    if metric_name == "cpu":
        return "CPU Usage: 42.5% (Load Avg: 2.10, 1.85, 1.50)"
    elif metric_name == "memory":
        return "Memory Used: 12.4GB / 32GB (Swap: 1.2GB)"
    return f"Metric '{metric_name}' not found."

if __name__ == "__main__":
    # 5. 启动服务，支持 Stdio (命令行子进程模式) 或 Streamable HTTP (HTTP 服务模式，
    # 独立 FastMCP 项目里对应 transport="http"；这是当前 MCP 官方传输方式，
    # 早期 2024-11 版协议里的 HTTP+SSE 已被取代，WebSocket 从来不是官方传输方式)
    # 本地默认通过 stdio 管道运行
    mcp.run(transport="stdio")

4. 生产级失效排查与系统缺陷诊断¶

故障现象	底层诱因	系统级表现	防御与排查手段
`Schema mismatch / Hallucination` 错配	Python 缺少类型提示（如 `limit` 没有指定类型），导致反射引擎生成了 `type: "any"`。	大模型频繁生成不合法类型的参数值，引发本地工具执行持续报错。	1. 强制对所有注册函数引入强类型提示与 Annotation。 2. 严格利用 Pydantic `Field` 提供字段级描述。
隐式 `print` 导致通信崩溃	业务代码或依赖库中误用了 `print()` 输出调试信息，污染了标准输出。	智能体连接瞬间被 Host 切断，报错 `corrupted JSON frame`。	1. 严格使用 logging 模块且重定向流至 `sys.stderr`。 2. 启动时执行拦截，重新路由全局标准输出。
异步 Handler 执行挂起	大批并发请求命中同步的工具 Handler，引发 Python GIL 线程阻塞。	并发吞吐暴跌，请求时延呈线性攀升，超时拦截器频频超时报错。	1. 将高并发工具 Handler 定义为 `async def`。 2. 耗时 CPU 计算任务移出主进程运行。

5. 资深系统架构师面试表达方案¶

面试提问：使用 Python 构建 MCP 服务时，你是如何保证模型契约的严密性与网络传输稳定性的？

回答模版： 用 Python 写 MCP Server，FastMCP 最省心的地方是不用手写 JSON Schema——函数签名加类型提示、Docstring 写清楚参数说明，剩下的交给它的反射引擎去生成。但这也意味着类型提示这件事不能偷懒，我们早期有个工具函数的参数没标类型，FastMCP 只能退化生成一个 type: any 的 Schema，模型经常传错格式的参数进来，框架自己又不报错，一直到工具执行内部才炸，排查起来很绕。后来我们定了个规矩，所有 @mcp.tool 函数强制类型标注加 Pydantic Field 描述，这个问题基本消失了。

另一个真的踩过的坑是标准输出污染——不是听说来的，是自己踩过。团队里有人在工具函数里顺手加了一行 print 调试，stdio 模式下这行输出直接混进了 JSON-RPC 的数据流，客户端瞬间报 JSON 解析错误断连，现象是“连接莫名其妙断了”，一开始完全没往这个方向想，最后翻日志才发现是这一行 print 的锅。现在我们会在项目里强制把 logging 重定向到 stderr，同时把高并发的工具 Handler 都改成 async def，避免同步阻塞拖慢整体吞吐。

PrefectHQ/fastmcp：几行代码把 Python 函数变成 MCP 工具的框架¶