12.4 中间件:拦截整条流水线

12.4 中间件:拦截整条流水线

本节摘要:本节讲横切关注点的解决方案——中间件(Middleware)。日志、鉴权、限流、指标这些需求不该散落在每个处理器里。中间件让你在请求/通知流水线上叠加拦截层,可任意组合。本节讲透中间件的链式执行模型、它与第 11 章 Bearer 中间件的关系、以及如何写一个自定义中间件。读完本节,你能把横切关注点从业务代码里彻底分离。

一、为什么需要中间件

先看一个「没有中间件」的痛点。假设每个工具都要记日志、查权限、限流:

# 反模式:每个工具重复写横切逻辑
@mcp.tool()
async def op1(ctx):
    log("开始 op1")                  # 重复
    if not check_auth(ctx):          # 重复
        raise PermissionError()
    if rate_limited(ctx):            # 重复
        raise RateLimitError()
    return do_op1()

@mcp.tool()
async def op2(ctx):
    log("开始 op2")                  # 又重复
    if not check_auth(ctx):          # 又重复
        ...
​

每个工具都重复写日志、鉴权、限流——代码重复、难维护、易漏。中间件解决这个——把这些逻辑抽成可叠加的层,统一应用。

二、中间件的链式执行模型

中间件构成一条「链」,请求依次穿过每一层:

请求到达
   │
   ▼
┌─────────────┐
│ 日志中间件  │ ← 记录每个请求
└─────────────┘
   │
   ▼
┌─────────────┐
│ 鉴权中间件  │ ← 校验令牌
└─────────────┘
   │
   ▼
┌─────────────┐
│ 限流中间件  │ ← 检查频率
└─────────────┘
   │
   ▼
┌─────────────┐
│ 业务处理器  │ ← 实际工具
└─────────────┘
   │
   ▼ 返回响应,反向穿过各层
   (鉴权记录、日志记录完成时间等)
​

每层中间件可以:

• 前置处理(请求往下传之前):如日志记录开始时间
• 决定是否继续:如鉴权失败,直接返回 401,不往下传
• 后置处理(响应往回传时):如日志记录耗时

三、写一个自定义中间件

中间件用 ServerMiddleware 定义:

from mcp.server import MCPServer
from mcp.shared.middleware import ServerMiddleware, MiddlewareContext

class LoggingMiddleware(ServerMiddleware):
    """记录每个请求的中间件。"""

    async def on_request(self, ctx: MiddlewareContext, next):
        """前置:记录开始;调用 next 往下传;后置:记录耗时。"""
        start = time.time()
        method = ctx.method
        print(f"[开始] {method}")

        try:
            result = await next(ctx)           # 调用下一层(或业务)
            elapsed = time.time() - start
            print(f"[完成] {method} ({elapsed:.3f}s)")
            return result
        except Exception as e:
            elapsed = time.time() - start
            print(f"[失败] {method} ({elapsed:.3f}s): {e}")
            raise

# 挂到服务端
mcp = MCPServer("Demo", middleware=[LoggingMiddleware()])
​

关键点:

• next(ctx):调用下一层。不调就不往下传(如鉴权失败时)
• 前置:在 next 之前的代码(如记开始时间)
• 后置:在 next 之后的代码(如记耗时)
• 异常处理:try/except 包住 next,能捕获业务异常

四、中间件的典型应用

中间件适合各种「横切关注点」:

第 11 章的 Bearer 中间件就是鉴权中间件的具体实现——它拦截请求、校验 Bearer 令牌、失败返 401。理解了通用中间件模型,你就理解了 Bearer 中间件的工作原理。

五、中间件的组合

多个中间件可任意组合,按顺序执行:

mcp = MCPServer(
    "Demo",
    middleware=[
        LoggingMiddleware(),        # 第 1 层:日志(最外层)
        AuthMiddleware(),           # 第 2 层:鉴权
        RateLimitMiddleware(),      # 第 3 层:限流
        MetricsMiddleware(),        # 第 4 层:指标
    ]
)
​

执行顺序:请求从左到右穿过(日志→鉴权→限流→指标→业务),响应反向。顺序很重要:

顺序建议(从外到内):
   日志(最外,记录所有)
   → 鉴权(失败的不该进限流)
   → 限流(过鉴权但超频的挡掉)
   → 指标(统计最终到业务的)
   → 业务
​

💡 技巧:鉴权中间件要在限流之前。否则未授权的请求也会消耗限流配额,导致恶意请求打满限流、合法用户被挡。这个顺序细节体现了中间件设计的考量。

六、缓存中间件:可不调 next 的特例

大多数中间件都调 next 往下传。但缓存中间件是个特例——命中时直接返回,不调 next:

class CacheMiddleware(ServerMiddleware):
    async def on_request(self, ctx, next):
        key = make_cache_key(ctx)
        cached = self.cache.get(key)
        if cached is not None:
            return cached            # 命中,直接返回(不调 next)
        result = await next(ctx)     # 未命中,往下传
        self.cache.set(key, result)  # 存缓存
        return result
​

这个「短路」能力让中间件非常强大——它不只是「观察」,还能「决定」是否继续。缓存、鉴权(失败时短路)、限流(超限时短路)都用这个能力。

七、中间件 vs 扩展:分工

第 12.3 节的扩展与本节的中间件,分工不同:

简单说:扩展「加新功能」,中间件「管现有功能」。两者互补,常配合用——扩展定义新方法,中间件统一拦截(包括拦截扩展方法)。

本节要点回顾

1. 中间件解决横切关注点散落问题,把日志/鉴权/限流等抽成可叠加的层。
2. 链式执行:请求依次穿过各层,每层可前置/短路/后置。
3. 写法:ServerMiddleware 子类 + on_request(ctx, next),调 next 往下传。
4. next(ctx) 是关键:不调就短路(如鉴权失败)。
5. 典型应用:日志、鉴权、限流、指标、缓存、追踪;Bearer 中间件就是鉴权实例。
6. 顺序重要:日志(最外)→鉴权→限流→指标→业务;鉴权要在限流前。
7. 缓存中间件可短路,命中不调 next。
8. 扩展加新方法,中间件管现有方法,两者互补。

中间件清楚了,最后一节讲 Apps 与 OpenTelemetry。