如何使用httpx和FastAPI直接中继并流式传输HTTP响应块？

需求背景 #

我正在使用httpx和FastAPI开发一个中继处理器，目标是将上游服务器的HTTP响应块直接流式传输给客户端——收到响应块后立即发送，而非缓冲所有数据后一次性发送。

已尝试的方案及结果 #

1. 异步生成器流式传输 #

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import httpx
import logging

logger = logging.getLogger(__name__)
app = FastAPI()

@app.post("/relay")
async def relay_request():
    try:
        async with httpx.AsyncClient() as client:
            async with client.stream("POST", relayhandler.url, headers=preparedheaders, json=prepared_request) as response:
                logger.debug(f"Received response with status code {response.status_code}.")

                if response.status_code != 200:
                    response_text = await response.aread()
                    logger.error(f"Relay request failed with status code {response.status_code}: {response_text.decode()}")
                    raise HTTPException(status_code=response.status_code, detail=response_text.decode())

                async def stream_response():
                    try:
                        async for chunk in response.aiter_bytes():
                            logger.debug(f"Received chunk: {chunk}")
                            yield chunk
                    except httpx.StreamClosed:
                        logger.error("Upstream response stream was unexpectedly closed")
                        raise HTTPException(status_code=500, detail="Upstream response stream was unexpectedly closed")

                headers = {
                    "content-type": "text/event-stream; charset=utf-8",
                }

                return StreamingResponse(stream_response(), status_code=response.status_code, headers=headers, media_type="text/event-stream;charset=UTF-8")

    except httpx.RequestError as e:
        logger.error(f"HTTPX Request error: {e}")
        raise HTTPException(status_code=500, detail=str(e))
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise HTTPException(status_code=500, detail="Unexpected internal server error")

结果：连接意外断开。

2. 直接测试流式生成器 #

import httpx

timeout = httpx.Timeout(10.0, read=None)  # 设置read超时为None以支持无限流式传输

async with httpx.AsyncClient(timeout=timeout) as client:
    try:
        stream_gen = stream_response(client, url, headers, payload)
        print("Streaming response:")
        async for chunk in stream_gen:
            print(chunk.decode())
    except httpx.RequestError as e:
        print(f"Request error: {e}")
    except httpx.StreamClosed:
        print("Stream was unexpectedly closed")
    except Exception as e:
        print(f"Unexpected error: {e}")

async def stream_response(client, url, headers, payload):
    async with client.stream('POST', url, headers=headers, json=payload) as response:
        async for chunk in response.aiter_bytes():
            if chunk:
                yield chunk

结果：所有数据被收集后一次性打印，未逐块流式输出。

3. 直接返回StreamingResponse #

from fastapi.responses import StreamingResponse

async with httpx.AsyncClient() as client:
    return StreamingResponse(
        client.stream('POST', relayhandler.url, headers=preparedheaders, json=prepared_request),
        status_code=200,
        media_type="text/event-stream;charset=UTF-8"
    )

结果：抛出TypeError: '_AsyncGeneratorContextManager' object is not iterable错误。

4. 当前实现方案 #

async with httpx.AsyncClient() as client:
    response = await client.post(
        relay_handler.url,
        headers=prepared_headers,
        json=prepared_request,
    )
    logger.debug(f"Received response with status code {response.status_code}.")

    if response.status_code != 200:
        response_text = await response.aread()
        logger.error(f"Relay request failed with status code {response.status_code}: {response_text.decode()}")
        raise HTTPException(status_code=response.status_code, detail=response_text.decode())

    async def stream_response():
        async for chunk in response.aiter_bytes():
            logger.debug(f"Received chunk: {chunk}")
            yield chunk

    headers = {
        "content-type": "text/event-stream; charset=utf-8",
    }

    return StreamingResponse(stream_response(), status_code=response.status_code, headers=dict(headers), media_type="text/event-stream;charset=UTF-8")

结果：功能可用，但会缓冲所有数据后一次性流式输出。

目标 #

最终目标是将上游服务器的每个响应块通过StreamingResponse直接流式传输给接收方，求解决方法。

环境 #

• Python 3.9
• httpx 0.26.0
• FastAPI 0.103.0
• Uvicorn 0.20.0

解决方案 #

核心问题分析 #

1. 方案1连接断开：大概率是超时设置问题——httpx默认的read超时会中断未及时返回数据的流；另外FastAPI/uvicorn的响应超时也可能导致连接被切断。
2. 方案2和4出现缓冲：await client.post()会默认缓冲整个响应，即使后续用aiter_bytes()也只是从本地缓冲读取，而非实时从上游拉取；若上游服务器本身一次性返回所有数据，也会导致无流式输出的假象。
3. 方案3的错误：client.stream()返回的是上下文管理器，不能直接传给StreamingResponse，需先进入上下文获取响应生成器。

修正后的实现代码 #

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import httpx
import logging

logger = logging.getLogger(__name__)
app = FastAPI()

# 全局设置httpx超时：连接超时10s，read超时设为None（无限制）
HTTPX_TIMEOUT = httpx.Timeout(10.0, read=None)

@app.post("/relay")
async def relay_request():
    try:
        # 初始化异步客户端，使用全局超时设置
        async with httpx.AsyncClient(timeout=HTTPX_TIMEOUT) as client:
            # 发起流式请求，注意不要用await，直接进入上下文
            async with client.stream(
                "POST", 
                relayhandler.url, 
                headers=preparedheaders, 
                json=prepared_request
            ) as response:
                logger.debug(f"Received upstream response status: {response.status_code}")

                # 处理上游错误响应
                if response.status_code >= 400:
                    error_content = await response.aread()
                    logger.error(f"Upstream error: {response.status_code} - {error_content.decode()}")
                    raise HTTPException(
                        status_code=response.status_code, 
                        detail=error_content.decode()
                    )

                # 定义流式生成器，直接从上游响应迭代块
                async def stream_generator():
                    try:
                        # 使用aiter_raw()替代aiter_bytes()，避免httpx的额外缓冲
                        async for chunk in response.aiter_raw():
                            if chunk:
                                logger.debug(f"Forwarding chunk size: {len(chunk)}")
                                yield chunk
                    except httpx.StreamClosed:
                        logger.error("Upstream stream closed unexpectedly")
                        raise HTTPException(status_code=500, detail="Upstream connection lost")

                # 复制上游的响应头，移除可能冲突的字段
                response_headers = {
                    k: v for k, v in response.headers.items()
                    if k.lower() not in ["content-length", "transfer-encoding"]
                }
                response_headers["content-type"] = "text/event-stream; charset=utf-8"

                # 返回StreamingResponse，禁用自动压缩避免缓冲
                return StreamingResponse(
                    stream_generator(),
                    status_code=response.status_code,
                    headers=response_headers,
                    media_type="text/event-stream; charset=utf-8"
                )

    except httpx.RequestError as e:
        logger.error(f"HTTP request failed: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Upstream connection failed: {str(e)}")
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

关键优化点 #

1. 超时配置：给httpx设置read=None的超时，避免流式传输时因长时间无数据被断开；启动uvicorn时可增加--timeout-keep-alive 600参数，延长服务器连接超时时间。
2. 使用aiter_raw()：替代aiter_bytes()，跳过httpx对响应块的额外处理和缓冲，直接获取原始字节流。
3. 避免预读取响应：绝对不能用await client.post()，必须用client.stream()并保持上下文活跃，直到所有块转发完成。
4. 响应头处理：复制上游响应头但移除Content-Length（流式传输长度未知），确保Transfer-Encoding: chunked被正确设置。
5. 禁用自动压缩：若FastAPI启用了Gzip中间件，会缓冲响应内容再压缩，需显式禁用针对text/event-stream的压缩。

测试验证建议 #

1. 确认上游服务器是分块返回数据：可自行编写一个测试上游服务（比如用FastAPI实现每秒返回一个事件的接口）。
2. 用curl测试中继接口：curl -N http://localhost:8000/relay，-N参数禁用curl缓冲，实时查看输出。
3. 检查日志：确认每个chunk的日志是实时打印的，而非一次性输出。

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