如何将FastAPI请求原样转发至另一服务器（非重定向）

要实现主应用处理请求的同时，把相同请求原样转发到另一服务器且不阻塞主流程，核心是用异步请求+后台任务/异步非阻塞调用，下面分两种场景给出具体实现：

单个端点的转发实现 #

如果只需要给特定端点添加转发逻辑，可以利用FastAPI的BackgroundTasks执行后台转发任务，不影响主请求的处理流程：

首先安装依赖：

pip install httpx

修改你的示例代码，添加转发逻辑：

from fastapi import BackgroundTasks, Request, Depends
import httpx

# 定义后台转发任务
async def forward_request(request: Request, target_base_url: str):
    # 复制原始请求的全部关键信息
    target_url = f"{target_base_url}{request.url.path}"
    query_params = dict(request.query_params)
    # 复制请求头，可按需排除不需要的字段（比如Host）
    headers = dict(request.headers)
    headers.pop("Host", None)
    # 获取原始请求体
    request_body = await request.body()

    # 异步发送请求，不等待响应
    async with httpx.AsyncClient() as client:
        await client.request(
            method=request.method,
            url=target_url,
            headers=headers,
            params=query_params,
            content=request_body
        )

@app.post("/my_endpoint/some_parameters")
def process_request(
    params: MyParamsClass,
    pwd: str = Depends(authenticate),
    request: Request = Request(...),
    background_tasks: BackgroundTasks = BackgroundTasks(...),
):
    # 添加转发任务到后台，指定目标服务器地址
    background_tasks.add_task(forward_request, request, "http://my_other_url")
    
    # 正常处理主请求
    return_value = process_the_request(params)
    return return_value.as_json()

全局所有端点的转发实现 #

如果需要所有请求都自动转发，推荐用FastAPI的中间件，一次性给所有端点添加转发逻辑：

from fastapi import Request
import httpx
import asyncio

@app.middleware("http")
async def forward_all_requests(request: Request, call_next):
    # 先处理主请求，拿到响应
    response = await call_next(request)

    # 定义转发逻辑的异步函数
    async def forward_task():
        target_url = f"http://my_other_url{request.url.path}"
        query_params = dict(request.query_params)
        headers = dict(request.headers)
        headers.pop("Host", None)
        request_body = await request.body()

        async with httpx.AsyncClient() as client:
            await client.request(
                method=request.method,
                url=target_url,
                headers=headers,
                params=query_params,
                content=request_body
            )
    
    # 启动异步任务，不等待完成直接返回主响应
    asyncio.create_task(forward_task())

    return response

注意事项 #

• 必须使用异步HTTP客户端（如httpx.AsyncClient），如果用同步客户端会阻塞主请求处理，违背需求
• 可根据需求过滤转发的请求，比如只转发POST请求、特定路径前缀的请求
• 如果目标服务器需要认证，可在headers中手动添加认证信息，或者保留原始请求的认证头
• 转发任务不会影响主请求的响应速度，因为不需要等待目标服务器的返回

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