基于FastAPI实现基础反向代理的可行性及优化咨询

环境无法使用Nginx、Traefik等现成反向代理工具，也无法采用容器化技术，希望用Python + FastAPI实现一个基础反向代理，核心需求如下：

• 仅需透明转发HTTP请求与WebSocket连接
• 保证资源高效，例如转发大文件时无需将整个请求/响应存入内存
• 无需负载均衡、SSL终止等复杂功能

已完成概念验证实现，基础场景下可正常运行，但存在以下疑问：

1. 已按RFC 7230规范剥离Connection头部，是否还有其他需要处理的头部？
2. WebSocket部分未处理头部，但双向通信能正常工作，是否存在潜在问题？
3. 这种方案是否可行，还是注定会失败？

import contextlib

import anyio
import fastapi
import httpx
import starlette.background
import starlette.responses
import uvicorn
import websockets

client = httpx.AsyncClient()


@contextlib.asynccontextmanager
async def lifespan(app: fastapi.FastAPI):
    yield
    await client.aclose()


service_mapping = {
    "x": "127.0.0.1:8001",
    "y": "127.0.0.1:8002",
}

app = fastapi.FastAPI(lifespan=lifespan)


def strip_connection_headers(headers: dict[str, str]) -> dict[str, str]:
    """RFC 7230 section 5.7 and 6.1"""
    lower_to_original_map = {k.strip().lower(): k for k in headers.keys()}
    result = headers.copy()

    if "connection" not in lower_to_original_map:
        return result

    connection_header_value = result[lower_to_original_map["connection"]]
    del result[lower_to_original_map["connection"]]

    for c in connection_header_value.split(","):
        c = lower_to_original_map.get(c.strip().lower())
        if c:
            del result[c]

    return result


@app.api_route(
    "/services/{service_name}/{path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"],
)
async def forward_http_request(
    request_from_client: fastapi.Request,
    service_name: str,
    path: str,
):
    if service_name not in service_mapping:
        raise fastapi.HTTPException(
            status_code=fastapi.status.HTTP_404_NOT_FOUND,
            detail=f"Service {service_name} not found",
        )

    base_url = f"http://{service_mapping[service_name]}"

    request_to_backend = client.build_request(
        method=request_from_client.method,
        url=f"{base_url}/{path}",
        headers=strip_connection_headers(dict(request_from_client.headers)),
        content=request_from_client.stream(),
        params=request_from_client.query_params,
    )

    response_from_backend = await client.send(request_to_backend, stream=True)

    return starlette.responses.StreamingResponse(
        content=response_from_backend.aiter_raw(),
        status_code=response_from_backend.status_code,
        headers=strip_connection_headers(dict(response_from_backend.headers)),
        background=starlette.background.BackgroundTask(
            response_from_backend.aclose
        ),
    )


@app.websocket("/ws/{service_name}/{path:path}")
async def forward_websocket_connection(
    websocket_with_client: fastapi.WebSocket,
    service_name: str,
    path: str,
):
    if service_name not in service_mapping:
        await websocket_with_client.close(
            code=fastapi.status.WS_1008_POLICY_VIOLATION,
            reason=f"service '{service_name}' unknown",
        )
        return

    await websocket_with_client.accept()

    base_url = f"ws://{service_mapping[service_name]}/{path}"

    websocket_with_backend = await websockets.connect(uri=base_url)

    async def forward_to_backend():
        while True:
            message = await websocket_with_client.receive()
            if message["type"] == "websocket.receive":
                message_content: str | bytes
                if "text" in message:
                    message_content = message["text"]
                else:
                    assert "bytes" in message
                    message_content = message["bytes"]
                await websocket_with_backend.send(message_content)
            else:
                assert message["type"] == "websocket.disconnect"
                task_group.cancel_scope.cancel()
                with anyio.CancelScope(shield=True):
                    await websocket_with_backend.close(
                        code=message["code"],
                        reason=message.get("reason", ""),
                    )
                    return

    async def forward_to_client():
        while True:
            try:
                message = await websocket_with_backend.recv()
            except websockets.exceptions.ConnectionClosed as e:
                task_group.cancel_scope.cancel()
                with anyio.CancelScope(shield=True):
                    await websocket_with_client.close(
                        code=e.code,
                        reason=e.reason,
                    )
                    return
            if isinstance(message, str):
                await websocket_with_client.send_text(message)
            else:
                assert isinstance(message, bytes)
                await websocket_with_client.send_bytes(message)

    async with anyio.create_task_group() as task_group:
        task_group.start_soon(forward_to_backend)
        task_group.start_soon(forward_to_client)


def main():
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        log_level="info",
        reload=True,
    )


if __name__ == "__main__":
    main()

HTTP头部处理补充 #

除了Connection头部，还有以下头部需要关注或调整：

• Host头部：当前直接转发客户端的Host，但后端服务可能依赖自身的Host地址（比如多域名部署的服务），建议替换为后端服务的Host（例如从service_mapping中提取127.0.0.1:8001作为Host值）
• *Forwarded/X-Forwarded-系列头部：如果后端需要获取客户端真实IP、请求协议或原始Host，建议添加X-Forwarded-For（客户端IP）、X-Forwarded-Proto（http/https）、X-Forwarded-Host（客户端请求的Host），让后端能拿到真实请求上下文
• Content-Length/Transfer-Encoding：流式请求/响应下，不要手动修改这两个头部，httpx和StreamingResponse会自动处理块编码和长度计算，避免出现响应异常
• Cookie头部：直接转发无问题，但如果后端有Cookie域名限制，需根据场景调整（内部服务一般无需处理）

WebSocket部分的潜在问题 #

当前实现能正常通信，但存在几个可优化点：

• 握手头部转发：客户端发起WebSocket握手时的Sec-WebSocket-Key、Sec-WebSocket-Version、Origin等头部，当前代码未转发给后端，websockets.connect会自动生成新的Sec-WebSocket-Key，若后端有Origin校验逻辑会失效。建议在连接后端时传递客户端的WebSocket握手头部：

  # 提取客户端WebSocket握手头部，保留关键字段
  client_ws_headers = {
      k: v for k, v in dict(websocket_with_client.headers).items()
      if k.lower() in ['sec-websocket-key', 'sec-websocket-version', 'origin', 'sec-websocket-protocol']
  }
  # 连接后端时传入头部
  websocket_with_backend = await websockets.connect(uri=base_url, headers=client_ws_headers)
  

• 异常场景处理：需增加websockets.connect失败的处理逻辑（比如后端服务不可达），此时应给客户端返回标准关闭码（如1006），避免服务崩溃
• 控制帧处理：websockets库会自动处理ping/pong控制帧，无需手动转发，但要确保代码不会误将控制帧当作普通消息处理

方案可行性判断 #

这个方案完全可行，适配你的场景：

• 资源效率：HTTP部分采用流式请求/响应，大文件转发不会占用大量内存；WebSocket采用异步双向转发，内存开销低
• 功能覆盖：基础透明转发需求已满足，补充头部处理后可应对绝大多数内部服务场景
• 局限性说明：
  • 性能：Python异步框架的性能不如Nginx等C语言实现的代理，但中小流量场景（内部服务、低并发）完全够用
  • 兼容性：部分边缘HTTP特性（如HTTP/2服务器推送）需额外处理，但标准HTTP方法和WebSocket场景无问题
  • 维护性：需自行维护代理逻辑，但代码量小，维护成本低

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