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. [VOLATILE: verify the current API before publishing.]

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.

[IMAGE PROMPT M8-6
Purpose: Show how cancellation propagates from a disconnecting client through the service to the upstream provider, and what breaks the chain.
Visual type: Three-tier flow diagram with a propagation path and a broken variant.
Prompt: A clean educational diagram in two stacked panels, each showing three boxes in a horizontal row labelled "client", "your service", and "provider", connected by arrows. The upper panel is headed "Cancellation propagates" and shows a leftmost marker at the client labelled "closes tab", with a cancellation arrow travelling right through the service box, which contains an inner label "async for over upstream stream", and continuing to the provider box, which carries an annotation "upstream call cancelled, billing stops". Each arrow segment is labelled "CancelledError raised at the await point". The lower panel is headed "Chain broken" and shows the same three boxes, but the service box contains an inner element labelled "detached create_task plus queue", and the cancellation arrow from the client stops at that inner element with a cross marker labelled "cancellation does not reach the task". The provider box in this panel carries an annotation "still generating, still billing". A caption beneath both reads "verify by closing a client and checking the provider actually stopped".
Required elements: Three tiers in both panels, a continuous cancellation path above, a blocked path with a cross marker below, contrasting provider annotations, the caption.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector.
Layout: Two panels stacked vertically, each reading left to right.
Text labels: "client", "your service", "provider", "Cancellation propagates", "Chain broken", "closes tab", "async for over upstream stream", "detached create_task plus queue", "CancelledError raised at the await point", "cancellation does not reach the task", "upstream call cancelled, billing stops", "still generating, still billing", "verify by closing a client and checking the provider actually stopped".
Aspect ratio: 16:9
Accessibility: Mark the broken path with a cross symbol and text rather than colour, and label every arrow.
Avoid: Vendor logos, decorative icons, screenshots, tiny text, watermarks.
Alt text: Two diagrams showing cancellation travelling from a disconnecting client through the service to the provider and stopping billing, against a broken chain where a detached task keeps the upstream call generating and billing.
END IMAGE PROMPT]

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.