Lesson 5: Re-streaming Through Your Own API

StreamingResponse and SSE endpoints

Your service sits between the user and the provider. The stream must pass through it.

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.post("/chat")
async def chat(body: ChatRequest, request: Request) -> StreamingResponse:
    async def event_stream() -> AsyncIterator[str]:
        async for event in provider.stream(body.messages, model=body.model):
            if event.kind == "content":
                yield f"data: {json.dumps({'text': event.text})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

Both headers matter in practice. Cache-Control: no-cache stops intermediaries caching a stream. X-Accel-Buffering: no tells nginx not to buffer the response, and without it a proxy may collect the entire stream and deliver it at once, which reproduces the non-streaming experience while your server believes it is streaming. That symptom, streaming that works locally and not in production, is almost always proxy buffering.

sse-starlette provides EventSourceResponse, which formats events for you and includes heartbeat support.

Proxying without buffering the whole response

The mistake that undoes everything:

# Wrong: collects the entire upstream response before sending anything
async def event_stream() -> AsyncIterator[str]:
    chunks = [event.text async for event in provider.stream(messages)]
    for chunk in chunks:
        yield f"data: {chunk}\n\n"

The comprehension consumes the whole upstream stream before the first yield, so time-to-first-token becomes total latency and the streaming is decorative.

Anything that materialises the stream has this effect: collecting to a list, joining into a string, or validating the complete response before forwarding. Forward each event as it arrives, and accumulate alongside if you need the whole text, which is what stream_and_collect from Lesson 3 does.

Backpressure and slow clients

The upstream provider produces at its own pace. A client on a poor mobile connection consumes more slowly. The difference has to go somewhere.

In a correctly written async chain it goes nowhere, because backpressure propagates: your generator yields, the server writes to the socket, the write blocks when the client's buffer is full, so the generator does not advance, so nothing is pulled from upstream. Memory stays flat.

What breaks it is putting an unbounded buffer in the middle.

# Wrong: unbounded queue between producer and consumer
queue = asyncio.Queue()          # no maxsize

An unbounded queue lets the producer run ahead indefinitely, so a slow client causes memory growth until the process fails. If you need a queue, bound it:

queue: asyncio.Queue[StreamEvent] = asyncio.Queue(maxsize=100)

A bounded queue makes put wait when full, which restores backpressure.

Limit concurrent streams. Each open stream holds an upstream connection and a server task. Bound them explicitly and reject beyond capacity rather than degrading everyone:

MAX_CONCURRENT_STREAMS = 40
stream_slots = asyncio.Semaphore(MAX_CONCURRENT_STREAMS)


@app.post("/chat")
async def chat(body: ChatRequest, request: Request) -> StreamingResponse:
    if stream_slots.locked():
        raise HTTPException(
            status_code=503,
            detail="at capacity, retry shortly",
            headers={"Retry-After": "5"},
        )
    await stream_slots.acquire()

    async def guarded() -> AsyncIterator[str]:
        try:
            async for event in provider.stream(body.messages, model=body.model):
                yield format_sse(event)
        finally:
            stream_slots.release()        # runs on completion, error, and cancellation
    ...

The finally is the whole point. Releasing in the normal path only would leak a slot on every client disconnect, and the service would silently lose capacity over hours until it stopped accepting streams.

Client disconnect and cancelling upstream

A user closes the tab. Without handling, your server keeps consuming the provider stream and you keep paying for tokens nobody will read.

Modern Starlette cancels the response generator when the client disconnects, which raises CancelledError inside it. Because cancellation propagates into what the generator is awaiting, the upstream request is cancelled too, provided you have not broken the chain.

async def event_stream() -> AsyncIterator[str]:
    try:
        async for event in provider.stream(body.messages, model=body.model):
            yield format_sse(event)
    except asyncio.CancelledError:
        logger.info("client disconnected, cancelling upstream")
        raise                            # always re-raise
    finally:
        stream_slots.release()

What breaks the chain. Running the upstream call in a detached create_task and reading from a queue means cancelling your generator does not cancel that task, so it continues consuming and paying. Catching CancelledError without re-raising has the same effect. And using a non-streaming upstream call means there is nothing to cancel, since the request is already in flight as a unit.

request.is_disconnected() can also be checked explicitly, which is useful for loops doing expensive work between yields, since cancellation only takes effect at an await point.

Verify this rather than assuming it. Start a long generation, close the client, and confirm from your provider dashboard or logs that the upstream call stopped. This is a cost bug that produces no error and no user complaint, so it is only ever found deliberately.

Errors after headers are sent

This is the failure that makes streaming genuinely harder than a normal response.

You returned a 200 status and sent forty chunks. Now the provider returns an error. You cannot send a 500, because the status line is already on the wire.

The answer is an error channel inside the stream.

async def event_stream() -> AsyncIterator[str]:
    try:
        async for event in provider.stream(body.messages, model=body.model):
            yield format_sse(event)
        yield 'event: done\ndata: {"status":"complete"}\n\n'
    except ProviderError as exc:
        logger.exception("upstream failed mid-stream")
        yield (
            'event: error\n'
            f'data: {json.dumps({"error": "generation_failed", "partial": True})}\n\n'
        )
    except asyncio.CancelledError:
        raise

Three requirements for this to be useful.

Distinguish completion from failure explicitly. A stream that simply stops is ambiguous: the client cannot tell a finished response from a dropped connection. Send an explicit terminal event in both cases, using the SSE event field so clients can dispatch on it.

Say that the output is partial. A client showing 400 tokens of an answer that was going to be 900 must indicate that it stopped early, or the user reads a truncated answer as a complete one, which is a correctness problem rather than a cosmetic one.

Do not leak internal detail. The error event goes to the browser. Log the full exception and send an error code, which is the same discipline as tool errors in Module 6.

The client must handle it. An error arriving after a 200 will not trigger a normal HTTP error path, so the client has to inspect events and act on the error type. A client that ignores the error event displays a truncated answer as though it were complete.

Heartbeats and idle-connection timeouts

Load balancers, proxies, and CDNs close connections that appear idle, commonly at around sixty seconds. A model thinking for ninety seconds before its first token looks idle, and the connection is killed by infrastructure neither end controls.

Heartbeats keep it alive. An SSE comment is the standard mechanism, since a line beginning with a colon is ignored by conforming clients.

async def with_heartbeat(
    events: AsyncIterator[str], *, interval: float = 15.0
) -> AsyncIterator[str]:
    """Emit a comment line when nothing has been sent for `interval` seconds."""
    iterator = events.__aiter__()
    while True:
        try:
            async with asyncio.timeout(interval):
                yield await iterator.__anext__()
        except StopAsyncIteration:
            return
        except TimeoutError:
            yield ": heartbeat\n\n"

Set the interval well under the shortest idle timeout in your path, and remember there may be several: your application server, a load balancer, a CDN, and possibly a corporate proxy at the client end.

Test through the full stack. Streaming that works against a local server and fails in production is almost always an intermediary buffering the response or closing it on idle, and neither is reproducible locally.