跳到主要内容

第 6 章:Context 能力、中间件、边界与代码地图

收尾章。讲三样把 FastMCP 从「函数转 schema 的玩具」变成「生产框架」的东西:工具运行时的 Context 能力、横切逻辑的中间件、以及诚实的边界说明。最后给一张全局代码地图。

6.1 Context:工具与客户端的双向对话门​

第 3 章讲了 Context 怎么被注入。这里讲它能干什么。Context(server/context.py:137)是工具在运行时拿到的一个把手,让本来「只能返回一个值」的工具能反过来和客户端/用户交互:

能力方法干什么定义处
日志ctx.debug/info/warning/error把结构化日志推给客户端context.py:720-:784
进度ctx.report_progress长任务上报进度条context.py:389
Samplingctx.sample反向请求客户端的大模型帮忙生成context.py:917
Elicitationctx.elicit运行中途向用户要一段输入context.py:1023
读资源ctx.read_resource工具内部读本服务器的资源context.py:533
会话状态ctx.set_state/get_state跨中间件/工具传值context.py:1249-:1305
rootsctx.list_roots拿客户端暴露的根目录context.py:784

两个最有意思的是 sampling 和 elicitation,因为它们反转了控制方向:

Sampling:工具反过来用大模型​

通常是「模型调工具」。sampling 让「工具调模型」:工具在执行中可以请求客户端那边的大模型生成一段文本。

# 示意,非源码。工具内部借客户端的模型做一次生成
@mcp.tool
async def summarize(text: str, ctx: Context) -> str:
    resp = await ctx.sample(f"用一句话总结:{text}")   # 反向调用客户端的 LLM
    return resp.text

价值:服务器不用自己接 OpenAI/Anthropic、不用自己的 API key——复用客户端已有的模型。

Elicitation:运行中途问用户​

工具执行到一半发现缺信息,可以 ctx.elicit 弹一个结构化请求给用户填,拿到答案再继续(context.py:1023,一堆重载支持不同返回类型)。适合「危险操作前二次确认」「缺参数时补问」。

6.2 中间件:按 MCP 方法分派的洋葱​

第 3 章讲了中间件链怎么组装成洋葱(_run_middleware)。这里讲怎么写一个。Middleware(server/middleware/middleware.py:88)提供一组按 MCP 方法分的钩子(middleware.py:140-:212):

  • 通用:on_message / on_request / on_notification;
  • 具体:on_call_tool / on_list_tools / on_read_resource / on_get_prompt / on_list_resources …

每个钩子拿到 MiddlewareContext(middleware.py:46)和一个 call_next,可以在调用前后加逻辑、改参数、短路返回、或放行:

# 示意,非源码。一个计时中间件
class TimingMiddleware(Middleware):
    async def on_call_tool(self, context, call_next):
        start = time.time()
        result = await call_next(context)      # 放行到内层
        log(f"{context.message.name} 耗时 {time.time()-start:.3f}s")
        return result

仓库内置了一批可直接用的(server/middleware/ 目录):authorization.py(鉴权)、rate_limiting.py(限流)、caching.py(缓存)、timing.py(计时)、logging.py、error_handling.py、response_limiting.py。

6.3 怎么跑起来:传输与运行​

mcp.run()(server/mixins/transport.py:91)是同步入口,内部 anyio.run 跑 run_async(transport.py:57)。它按 transport 参数分流:

run(transport=?)
   ├── "stdio"                → run_stdio_async   (transport.py:198)  子进程/CLI 场景,读写标准输入输出
   └── "http"/"sse"/"streamable-http" → run_http_async (transport.py:240)  起 Starlette ASGI 应用
  • stdio(默认):服务器通过标准输入/输出收发 JSON-RPC,适合被 Claude Desktop 这类客户端当子进程拉起。
  • http:run_http_async 起一个 Starlette 应用(server/http.py),用官方 SDK 的 StreamableHTTPSessionManager 管会话。你还能用 @mcp.custom_route(transport.py:114)加健康检查、OAuth 回调等自定义 HTTP 端点。

6.4 边界与局限(诚实说)​

边界说明依据
工具函数不能有 *args / **kwargs无法映射成固定 JSON schema,注册期直接报错tools/function_parsing.py:187 起
lambda 必须显式起名匿名函数没有稳定 nametools/function_tool.py:317
bytes 返回值没有输出 schema无法表示成结构化 JSON,被跳过tools/function_parsing.py:39 _contains_bytes_type
同步函数 + timeout + run_in_thread=False 组合被拒内联同步执行无取消检查点,超时会静默失效tools/function_tool.py:326 起
错误详情默认被遮蔽_mask_error_details 默认开,客户端看不到原始异常server/server.py call_tool 的 except 分支
强依赖 pydantic 的类型系统schema 生成能力等于 pydantic 的表达力;pydantic 表示不了的类型就表示不了第 2 章
v3 是 workspace 布局真代码在 fastmcp_slim/,顶层 fastmcp 是聚合壳pyproject.toml [tool.uv.workspace]

这套设计刻意不做的事: 它不替你决定「工具该干什么」,只解决「怎么把你的函数暴露成合规的 MCP」。业务正确性、幂等性、副作用安全仍是你的责任。

6.5 横向对比(同 shelf 视角)​

FastMCP 在 protocols 区里是「协议的框架实现」这一类的代表,和「协议规范本身 / 底层 SDK」形成对比:

  • 相对官方 MCP Python SDK 的 LowLevelServer: FastMCP 把它当引擎包在里面(server/low_level.py),自己负责「函数 → schema → 处理函数」的高层人体工程学。SDK 给你原始 JSON-RPC 手柄,FastMCP 给你装饰器。
  • 相对 FastAPI(设计谱系上的兄长): 同样是「类型注解 + 装饰器 + pydantic schema」,但目标协议从 HTTP/OpenAPI 换成 MCP;from_fastapi 直接把这层关系变成一行代码。

6.6 全局代码地图(跨章导航)​

想找文件路径符号名章
服务器门面fastmcp_slim/fastmcp/server/server.pyFastMCP1
三个装饰器fastmcp_slim/fastmcp/server/server.pyFastMCP.tool/resource/prompt1
本地注册表fastmcp_slim/fastmcp/server/providers/local_provider/local_provider.pyLocalProvider1
函数→schemafastmcp_slim/fastmcp/tools/function_parsing.pyParsedFunction.from_function2
TypeAdapter 双重身份fastmcp_slim/fastmcp/tools/function_tool.pyFunctionTool._execute2,3
调用主干fastmcp_slim/fastmcp/server/server.pyFastMCP.call_tool3
协议接线fastmcp_slim/fastmcp/server/mixins/mcp_operations.py_setup_handlers3
依赖注入fastmcp_slim/fastmcp/server/dependencies.pytransform_context_annotations、without_injected_parameters3
Provider 抽象fastmcp_slim/fastmcp/server/providers/base.pyProvider4
组合fastmcp_slim/fastmcp/server/server.pymount、as_proxy、from_openapi4
客户端fastmcp_slim/fastmcp/client/client.pyClient5
传输推断fastmcp_slim/fastmcp/client/transports/inference.pyinfer_transport5
Context 能力fastmcp_slim/fastmcp/server/context.pyContext.sample、Context.elicit、Context.report_progress6
中间件fastmcp_slim/fastmcp/server/middleware/middleware.pyMiddleware6
运行/传输fastmcp_slim/fastmcp/server/mixins/transport.pyrun、run_stdio_async、run_http_async6