FastAPI自定义中间件获取请求Body触发Content-Length错误如何解决

报错根因：Response 内置的 body_iterator 是单次消费的异步流，你在中间件中遍历读取后，流内没有剩余内容，返回给客户端时实际响应长度和响应头的 Content-Length 不匹配，就会触发该 RuntimeError。同时请求的 Body 流也是单次消费的，读取后需要重新回填，否则后续业务路由无法正常获取请求参数。

完整实现代码 #

以下实现仅在中间件层完成，无需自定义APIRouter、无业务侵入，同时支持请求Body鉴权、日志采集能力：

from fastapi import FastAPI, Request, HTTPException, Response
import json

app = FastAPI()

# 自行实现数据库密钥校验逻辑
def check_secret_valid(secret: str) -> bool:
    # 示例：查询数据库校验密钥合法性
    # valid_secrets = db.query(SecretModel).all()
    # return secret in [item.key for item in valid_secrets]
    return secret == "your_valid_secret"

@app.middleware("http")
async def custom_auth_log_middleware(request: Request, call_next):
    # 1. 读取请求Body用于鉴权、日志采集
    req_body_bytes = await request.body()
    # 兼容非JSON请求场景，避免解析报错
    try:
        req_body = json.loads(req_body_bytes)
    except json.JSONDecodeError:
        req_body = {}
    
    # 2. 密钥鉴权逻辑，校验不通过直接拦截返回
    req_secret = req_body.get("secret_key", "")
    if not check_secret_valid(req_secret):
        raise HTTPException(status_code=401, detail="密钥校验失败，无访问权限")

    # 3. 回填请求Body，避免后续业务路由拿不到请求参数
    request = Request(
        scope=request.scope,
        receive=lambda: {"type": "http.request", "body": req_body_bytes}
    )

    # 4. 执行业务逻辑获取响应
    response = await call_next(request)

    # 5. 如需采集响应日志，读取响应Body后重新构造返回
    resp_body_bytes = b""
    async for chunk in response.body_iterator:
        resp_body_bytes += chunk
    # 日志打印按需调整
    print(f"请求BODY: {req_body}")
    print(f"响应BODY: {resp_body_bytes.decode('utf-8')}")

    # 6. 回填响应Body，解决Content-Length不匹配报错
    return Response(
        content=resp_body_bytes,
        status_code=response.status_code,
        headers=dict(response.headers),
        media_type=response.media_type
    )

注意事项 #

• 大文件上传、二进制流接口建议添加路径白名单，跳过Body读取逻辑，避免内存占用过高
• 如无需采集响应日志，可删除第5、6步的响应处理逻辑，直接返回await call_next(request)即可
• 密钥校验逻辑建议添加本地缓存，降低频繁查询数据库的性能开销

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