[Python] FastAPI StreamingResponse not streaming upstream data to client

I am trying to write the simple proxy for llama, In which I am trying to proxy stream response but it fails with ReadError in SSE stream: , If I try with simple print without StreamingResponse method it works well and print stream at server as well.

import logging
import yaml
import httpx
import json
from fastapi import FastAPI, Request, Response, HTTPException
from fastapi.responses import StreamingResponse
import asyncio
import uvicorn
from sse_starlette.sse import EventSourceResponse
import time

# Logger setup
logger = logging.getLogger("proxy")
handler = logging.StreamHandler()
logger.addHandler(handler)

class Upstream:
def __init__(self, url, cls):
self.url = url
self.cls = cls

class Config:
def __init__(self, log_level, upstreams: list = []):
self.log_level = log_level
self.upstream = []
logger.setLevel(logging.DEBUG if log_level == "development" else logging.INFO)
for upstream in upstreams:
self.upstream.append(Upstream(upstream.get("url"), upstream.get("cls")))

@staticmethod
def load_config(config_file):
with open(config_file, "r") as f:
yml = yaml.safe_load(f)
return Config(yml.get("log_level"), yml.get("upstreams"))

def pick_random_upstream(self) -> Upstream:
return self.upstream[0] # TODO: Implement actual load balancing

# Load config
cfg = Config.load_config("config.yaml")
app = FastAPI()

@app.api_route("/{full_pathath}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def proxy(request: Request, full_path: str):
body = await request.body()
try:
req_body = json.loads(body)
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="Invalid request body")

model = req_body.get("model")
if not model:
raise HTTPException(status_code=400, detail="Model is required")

logger.debug(f"Model: {model} - Path: {full_path} - Body: {req_body}")

upstream = cfg.pick_random_upstream()
upstream_url = upstream.url

# Prepare headers (excluding 'host' to avoid conflicts)
headers = {key: value for key, value in request.headers.items() if key.lower() != "host"}

async with httpx.AsyncClient(timeout=None) as client:
request_obj = client.build_request(
request.method, f"{upstream_url}/{full_path}", content=body, headers=headers
)

upstream_response = await client.send(request_obj, stream=True)

content_type = upstream_response.headers.get("Content-Type", "")

if "text/event-stream" in content_type:
async def sse_stream():
print(upstream_response.is_closed)
try:
async for line in upstream_response.aiter_lines():
if line:
print(f"SSE Received: {line}")
yield f"data: {line}\n\n"
except httpx.ReadError as e:
print(f"ReadError in SSE stream: {e}")
except Exception as e:
print(e)
finally:
await upstream_response.aclose() # Ensure cleanup

return StreamingResponse(sse_stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"})

response_data = await upstream_response.aread()
return Response(content=response_data, status_code=upstream_response.status_code, headers=upstream_response.headers)

if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)



What i am doing wrong? Thanks.

Continue reading...