如何在FastAPI应用启动前添加中间件？解决启动后添加报错问题

问题场景 #

尝试在asynccontextmanager修饰的lifespan函数中，根据条件为FastAPI应用动态添加HTTP中间件，示例代码如下：

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    if some_condition:
        @app.middleware("http")
        async def my_middleware(request: Request, call_next):
            response = await call_next(request)
            return response
    yield
    pass

app = FastAPI(lifespan=lifespan)

运行时触发错误：

RuntimeError: Cannot add middleware after an application has started

虽然FastAPI生命周期事件文档提到：

你可以定义在应用启动前执行的逻辑（代码）。这意味着该代码会在应用开始接收请求前执行一次。

但实际操作仍报错，需要解决该问题。

原因分析 #

@app.middleware装饰器本质是调用app.add_middleware()方法，而FastAPI/Starlette在应用实例初始化完成后（即lifespan函数开始执行时），会锁定中间件注册系统，禁止后续添加中间件——此时应用已经完成启动前的准备工作，中间件栈已构建完毕，运行时修改会导致请求处理逻辑不一致，因此抛出错误。

可行解决方案 #

方案1：在应用实例初始化阶段添加中间件 #

如果some_condition是启动时就能确定的固定条件，直接在创建FastAPI实例后、启动应用前判断并添加中间件：

from fastapi import FastAPI, Request

some_condition = True  # 启动时确定的条件

# 先创建应用实例
app = FastAPI()

# 根据条件添加中间件
if some_condition:
    @app.middleware("http")
    async def my_middleware(request: Request, call_next):
        response = await call_next(request)
        # 这里添加自定义中间件逻辑
        return response

# 后续定义路由等逻辑
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

方案2：使用全局依赖替代中间件（运行时条件判断） #

如果需要根据请求级别的运行时条件决定是否执行类似中间件的逻辑，改用FastAPI的全局依赖实现：

from fastapi import FastAPI, Request, Depends

async def conditional_middleware_logic(request: Request):
    # 这里可以写运行时条件判断，比如根据请求头、路径等
    if request.headers.get("X-Custom-Header"):
        # 执行原本中间件的逻辑，比如修改请求或响应
        pass
    return request

# 将依赖全局应用到所有路由
app = FastAPI(dependencies=[Depends(conditional_middleware_logic)])

@app.get("/")
async def read_root():
    return {"message": "Hello World"}

方案3：手动修改中间件栈（不推荐） #

如果必须在lifespan中动态添加，可以直接操作FastAPI实例的middleware_stack属性（依赖Starlette内部实现，版本更新可能失效）：

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from contextlib import asynccontextmanager

# 先定义中间件类
class MyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        # 自定义中间件逻辑
        return response

@asynccontextmanager
async def lifespan(app: FastAPI):
    if some_condition:
        # 将自定义中间件插入到现有中间件栈中
        app.middleware_stack = MyMiddleware(app.middleware_stack)
    yield

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def read_root():
    return {"message": "Hello World"}

内容的提问来源于stack exchange，提问作者Dante