FastAPI中间件修改请求头和查询参数后端点未生效，如何解决？

你遇到的问题核心在于：FastAPI/Starlette的Request对象的headers和query_params属性，并不是直接读取你修改的私有变量_headers和_query_params，而是从ASGI协议的scope字典中解析生成的。直接修改私有变量只会在中间件内部临时生效，后续端点处理时会重新从原始scope中读取数据。

正确修改Headers的方法 #

不要直接操作request._headers，而是修改request.scope["headers"]。这里的headers是字节数组形式，需要把键值对都转成字节：

• 先删除旧的同名header（如果需要）
• 再添加新的header到scope的headers列表中

正确修改Query Params的方法 #

Query参数是从request.scope["query_string"]解析来的，所以要修改这个字符串，而不是request._query_params。可以用urllib.parse来解析、修改，再重新编码成字节字符串。

修正后的完整代码 #

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from urllib.parse import parse_qsl, urlencode

class TestMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: FastAPI):
        super().__init__(app)

    def get_modified_query_string(self, original_query_str: bytes) -> bytes:
        # 解析原始query字符串为字典
        query_params = dict(parse_qsl(original_query_str.decode("utf-8")))
        # 这里做修改，比如添加或修改参数
        query_params["new_param"] = "new_value"
        if "existing_param" in query_params:
            query_params["existing_param"] = "updated_value"
        # 重新编码为字节字符串
        return urlencode(query_params).encode("utf-8")

    async def dispatch(self, request: Request, call_next):
        # 修改Headers
        header_key = b"x-device-info"  # ASGI规范中headers键为小写字节
        new_header_value = b"new device info"
        
        # 过滤掉旧的同名header
        request.scope["headers"] = [
            (k, v) for k, v in request.scope["headers"] if k != header_key
        ]
        # 添加新的header
        request.scope["headers"].append((header_key, new_header_value))

        # 修改Query Params
        original_query = request.scope.get("query_string", b"")
        modified_query = self.get_modified_query_string(original_query)
        request.scope["query_string"] = modified_query

        # 验证修改（可选）
        print("modified headers =>", request.headers)
        print("modified params =>", request.query_params)

        response = await call_next(request)
        return response

# 测试用FastAPI实例
app = FastAPI()
app.add_middleware(TestMiddleware)

@app.get("/test")
async def test_endpoint(request: Request):
    print("endpoint headers =>", request.headers)
    print("endpoint query params =>", request.query_params)
    return {"headers": dict(request.headers), "query_params": dict(request.query_params)}

关键细节说明 #

• Headers在ASGI scope中是以小写字节键存储的，所以修改时必须用小写的字节形式，否则会出现重复header或修改不生效的情况。
• 修改Query参数时，必须重新生成query_string，因为request.query_params每次访问都会从这个字符串重新解析，直接修改私有变量不会触发重新解析。
• 不要依赖Request对象的私有变量（比如_headers、_query_params），这些是内部实现细节，版本更新可能会变动，应使用操作scope的标准方式。

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