FastMCP Middleware 实战：用 LoggingMiddleware 与 RateLimitingMiddleware 重塑执行逻辑

1. 为什么要在 FastMCP 里折腾 MiddlewareFastMCP 的 Middleware 是一层可以插在请求进入处理器之前、响应返回客户端之前的拦截与增强机制。它最直接的价值是不用改任何工具函数的业务代码就能给整个 MCP Server 加上日志、限流、缓存、耗时统计这些横切能力。适合谁适合已经把 FastMCP 跑起来、工具能正常调用但开始遇到“某个工具被疯狂重复调用”“线上出问题不知道谁在调什么”“想给不同客户端分配不同额度”这类问题的开发者。我试过把日志和限流直接写进每个工具函数里结果是十几个工具复制粘贴同一段代码改一次日志格式要动十几个文件。Middleware 解决的正是这个问题它把“守门”和“增强”从业务逻辑里抽出来注册一次全局生效。FastMCP 的中间件链是双向的。请求从客户端出发按注册顺序依次穿过每个中间件的__call__到达最终处理器处理器返回后结果再反向穿过这条链。所以一个中间件天然拥有“前置”和“后置”两个切面日志可以记录开始和结束限流可以在前置直接拒绝缓存可以在后置写入结果。这篇聚焦两个最典型的场景用LoggingMiddleware看清执行链路用RateLimitingMiddleware控制请求节奏。同时会顺带把ResponseCachingMiddleware的取舍讲清楚因为限流和缓存经常一起出现。所有代码都可以在本地直接跑通验证动作我会给出预期输出。2. 前置准备环境、依赖与 TaoToken 接入在写中间件之前先把运行环境固定下来。FastMCP 的中间件模块路径在较新版本里是fastmcp.server.middleware.*如果你用的是老版本导入路径可能不同建议先确认版本。python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install fastmcp2.0 httpx python -c import fastmcp; print(fastmcp.__version__)如果你打算把 MCP Server 接到真实模型上做端到端验证需要一个稳定的模型调用入口。TaoToken 提供 OpenAI 兼容的接口官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 。拿到 Key 的入口在控制台的 API Keys 页面https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有 base_url 和鉴权头的完整说明。注意Middleware 是 FastMCP Server 侧的能力和模型供应商无关。TaoToken 在这里的角色是让你有一个可用的模型后端来触发工具调用从而观察中间件的日志和限流行为。两者是配合关系不是替代关系。环境变量建议这样设置避免把 Key 写进代码export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api3. 可复制配置注册 LoggingMiddleware 与 RateLimitingMiddleware先给一个最小可运行的 Server 骨架把两个中间件都注册进去。注意注册顺序会影响执行顺序先注册的在外层请求先经过它。import asyncio import logging from fastmcp import FastMCP, Client from fastmcp.server.middleware.logging import LoggingMiddleware from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) mcp FastMCP(DemoServer) mcp.tool async def get_weather(city: str) - str: get weather info for a city. return fIts sunny and 25 degrees Celsius in {city}. mcp.tool async def long_running_task() - str: simulate a slow task. await asyncio.sleep(2) return done # 先注册日志再注册限流请求先过日志再过限流 mcp.add_middleware(LoggingMiddleware( include_payloadsTrue, max_payload_length500, methods[tools/call], # 只关心工具调用避免 initialize 刷屏 )) mcp.add_middleware(RateLimitingMiddleware( max_requests_per_second2, burst_capacity2, global_limitTrue, )) async def main(): async with Client(mcp) as client: await client.call_tool(nameget_weather, arguments{city: Suzhou}) if __name__ __main__: asyncio.run(main())LoggingMiddleware的关键参数值得单独说。include_payloadsTrue会把请求体打出来调试时非常有用但生产环境如果 payload 很大日志会爆炸所以配合max_payload_length截断。methods[tools/call]是我强烈建议加的否则 initialize、tools/list 这些握手请求也会被记录日志里全是噪音。estimate_payload_tokensTrue可以估算 token 消耗对大模型应用做成本监控很有价值。RateLimitingMiddleware用的是令牌桶算法。max_requests_per_second2表示长期平均每秒放行 2 个请求burst_capacity2表示允许瞬间最多 2 个请求通过。global_limitTrue是全局共享额度所有客户端共用一个桶设为False则按get_client_id返回的标识给每个客户端独立分桶。如果你需要更严格的窗口控制可以换成SlidingWindowRateLimitingMiddlewarefrom fastmcp.server.middleware.rate_limiting import SlidingWindowRateLimitingMiddleware mcp.add_middleware(SlidingWindowRateLimitingMiddleware( max_requests5, window_minutes1, ))它不允许突发流量1 分钟内第 6 个请求一定被拒。令牌桶适合“允许偶尔爆发但长期受限”的场景滑动窗口适合“硬上限、绝不超发”的场景选哪个取决于你的业务对突发流量的容忍度。4. 验证请求日志输出与限流触发先验证日志。运行上面的main()你会看到类似这样的输出2026-04-07 21:34:58 INFO eventrequest_start methodtools/call sourceclient payload{name:get_weather,arguments:{city:Suzhou}} payload_typeCallToolRequestParams 2026-04-07 21:34:58 INFO eventrequest_success methodtools/call sourceclient duration_ms6.21request_start和request_success成对出现duration_ms是这次调用的耗时。因为只过滤了tools/call所以看不到 initialize 的日志链路很干净。如果你把methods去掉会看到完整的握手过程包括initialize、tools/list、tools/call三段。再验证限流。把main()改成并发发起 5 个long_running_task调用async def main(): async with Client(mcp) as client: async def call_task(i): try: await client.call_tool(namelong_running_task, arguments{}) print(fCall {i} succeeded.) except Exception as e: print(fCall {i} rejected: {type(e).__name__}) await asyncio.gather(*[call_task(i 1) for i in range(5)]) asyncio.run(main())预期结果是 5 个并发请求里只有 2 个成功另外 3 个被限流中间件拦截并抛出异常。因为burst_capacity2桶里初始只有 2 个令牌5 个请求同时到达时只有 2 个能立刻拿到令牌其余的直接被拒。被拒的请求不会进入工具函数所以long_running_task里的asyncio.sleep(2)只执行了 2 次整体耗时约 2 秒而不是 10 秒。这里有个容易忽略的点限流中间件在请求进入处理器之前就拒绝了所以日志中间件如果注册在限流外层会记录到request_start但不会有request_success而是记录一个失败事件。这个顺序差异在排查“为什么请求没到工具函数”时非常关键。5. 本篇常见错排查导入路径报错ModuleNotFoundError: No module named fastmcp.server.middleware。这是版本问题。早期 FastMCP 的中间件模块路径不同先升级到 2.0 以上pip install -U fastmcp。如果升级后仍报错用python -c import fastmcp.server.middleware as m; print(dir(m))确认实际可用的子模块名。日志里看不到 payload。检查include_payloads是否设为True同时确认methods过滤没有把你要看的方法排除掉。另外max_payload_length太小会导致 payload 被截断成空字符串调大到 1000 以上再试。限流完全不生效所有请求都通过。最常见的原因是global_limit和get_client_id的组合。如果global_limitFalse且没有提供get_client_id默认行为可能不符合预期。先用global_limitTrue验证限流本身是否工作再切换到分用户模式。另一个原因是max_requests_per_second设得太大比如默认值 10而你只发了 5 个请求自然不会触发。并发测试时结果不稳定。asyncio.gather的调度顺序不保证加上令牌桶有时间维度每次运行成功的是哪 2 个可能不同。这是正常现象只要成功数量符合预期即可。如果要精确复现改用SlidingWindowRateLimitingMiddleware并固定窗口。缓存中间件导致工具返回旧数据。ResponseCachingMiddleware默认对所有操作开启缓存get_current_time这类工具会被缓存住三轮调用返回同一个时间。如果你需要实时数据要么给该工具设置call_tool_settings排除它要么把ttl设得很短。缓存是空间换时间用之前先想清楚哪些工具的结果可以复用。from fastmcp.server.middleware.caching import ResponseCachingMiddleware mcp.add_middleware(ResponseCachingMiddleware( call_tool_settings{ enabled: True, ttl: 30, excluded_tools: [get_current_time], }, ))6. 把中间件接到真实调用链上本地跑通之后下一步是把它接到真实的模型调用链上。你可以用 TaoToken 的模型对话入口做端到端验证https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 在对话里触发工具调用观察 Server 侧的日志和限流是否符合预期。如果要做长期编码或 Agent 场景Coding Plan 的入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合持续性的工具调用负载。接入时把 base_url 指向 https://taotoken.net/api 鉴权头用Authorization: Bearer $TAOTOKEN_API_KEY。完整的参数说明和示例在接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里建议对照着把LoggingMiddleware的methods过滤和RateLimitingMiddleware的get_client_id调成适合你业务的配置。最后留一个实操建议中间件的注册顺序就是执行顺序把日志放在最外层限流放在日志内层缓存放在最内层。这样日志能看到所有请求包括被限流拒绝的限流能在进入缓存和处理器之前拦截缓存只对真正执行的请求生效。这个顺序在排查问题时能省掉很多猜测。