Streaming responses through FastAPI

ShipFast's agents use a console. When an agent opens a message that the triage service could not draft a reply for, they can press "Draft for me". The first version waited for the whole reply and then showed it. A 90-word reply took about 3 seconds, and agents pressed the button again because nothing seemed to happen, which started a second paid call.

The model writes one token at a time, and the first tokens are ready after about half a second. Streaming sends them to the browser as they arrive. The total time does not change, but the agent sees words after 0.5 seconds instead of a blank box for 3 seconds, and stops pressing the button twice.

When to stream and when not to

ShipFast streams exactly one thing: the draft reply in the agent console. Classification and extraction return small JSON objects that code must validate as a whole. Streaming them would add complexity for no visible gain, because half a JSON object is useless.

There is a real trade-off here. You cannot fully validate a streamed reply until it has finished, so the agent may see words that your checks would reject. For an internal console where a trained agent reviews every draft, that is acceptable. For text going straight to a customer, it is not: generate, check, then send.

Adding stream() to LLMClient

The SDK's streaming helper gives you an async iterator of text chunks and, at the end, the final message with its usage. The client yields chunks and logs usage exactly like complete(), so streamed calls show up in the same cost reports.

1# shipfast/llm.py, inside class LLMClient2    async def stream(self, *, feature: str, system: str, messages: list[dict],3                     max_tokens: int = 600) -> AsyncIterator[str]:4        start = time.perf_counter()5        try:6            async with self._sdk.messages.stream(7                model=self.model, max_tokens=max_tokens, system=system, messages=messages,8                output_config={"effort": self.effort} if self.effort else anthropic.omit,9            ) as stream:10                async for chunk in stream.text_stream:11                    yield chunk12                final = await stream.get_final_message()13        except anthropic.APIError as err:14            raise _translate(feature, err) from err15        _log(feature, LLMResult(16            text="", model=self.model, input_tokens=final.usage.input_tokens,17            output_tokens=final.usage.output_tokens, stop_reason=final.stop_reason,18            latency_ms=round((time.perf_counter() - start) * 1000)))

The async with block matters. If the consumer stops iterating, for example because the browser disconnected, leaving the block closes the connection to the provider. Generation stops, and you stop paying for tokens nobody will read.

Serving it as server-sent events

Server-sent events (SSE) are a simple format for a one-way stream over HTTP: lines starting with event: and data:, separated by a blank line. Browsers read them with EventSource or fetch, and they pass through most proxies and load balancers.

1# shipfast/api.py (streaming part)2import json3from functools import lru_cache45from fastapi import Depends, FastAPI6from fastapi.responses import StreamingResponse7from pydantic import BaseModel, Field89from shipfast.config import settings10from shipfast.drafts import SYSTEM as DRAFT_SYSTEM, draft_problems11from shipfast.llm import LLMClient, LLMError1213app = FastAPI(title="ShipFast support API")141516@lru_cache17def get_llm() -> LLMClient:18    return LLMClient(settings.llm_model, timeout_s=settings.llm_timeout_s)192021class DraftRequest(BaseModel):22    text: str = Field(min_length=1, max_length=2_000)232425def sse(event: str, data: dict) -> str:26    return f"event: {event}\ndata: {json.dumps(data)}\n\n"272829@app.post("/v1/drafts/stream")30async def stream_draft(req: DraftRequest, llm: LLMClient = Depends(get_llm)):31    async def events():32        parts: list[str] = []33        try:34            async for chunk in llm.stream(35                    feature="draft_stream", system=DRAFT_SYSTEM,36                    messages=[{"role": "user", "content": f"<message>\n{req.text}\n</message>"}]):37                parts.append(chunk)38                yield sse("delta", {"text": chunk})39        except LLMError as err:40            yield sse("error", {"message": type(err).__name__})41            return42        yield sse("done", {"problems": draft_problems("".join(parts))})4344    return StreamingResponse(events(), media_type="text/event-stream",45                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

DRAFT_SYSTEM and draft_problems come from shipfast/drafts.py, which Section 4 builds; for now, read them as "the draft instructions" and "a function that lists problems in a draft, such as a promised refund". @lru_cache on get_llm means one client per process, so HTTP connections to the provider are reused.

Three details in this endpoint save you from real incidents.

Errors after the first byte. Once streaming starts, the HTTP status (200) has already been sent. You cannot change it to 503. So errors travel inside the stream as an error event, and the console shows "Draft failed, write it yourself" instead of hanging.

Checks at the end. The whole draft is checked once it is complete, and the result travels in the done event. If the model wrote "we will refund you", the console shows a warning next to the draft before the agent can send it.

Proxy buffering. Many reverse proxies, nginx by default, collect a response before forwarding it, which silently turns your stream into one late block. The X-Accel-Buffering: no header disables that for nginx; check your load balancer's documentation for the equivalent, and check its idle timeout too, because a slow stream can be cut off by a 30-second limit.

Testing a stream

The endpoint only needs something with a stream() method, so a test swaps in a fake that yields fixed chunks and uses FastAPI's TestClient to read the events.

1from fastapi.testclient import TestClient23from shipfast.api import app, get_llm456class FakeStreamLLM:7    model = "claude-haiku-4-5"89    async def stream(self, **_: object):10        for part in ["Hi, ", "we have ", "requested a refund."]:11            yield part121314def test_stream_flags_a_refund_promise():15    app.dependency_overrides[get_llm] = lambda: FakeStreamLLM()16    with TestClient(app).stream("POST", "/v1/drafts/stream", json={"text": "box wet"}) as r:17        body = "".join(r.iter_text())18    app.dependency_overrides.clear()19    assert body.count("event: delta") == 320    assert "forbidden" in body          # the done event lists the refund problem

The test checks the order of events and the final check without any network, which is exactly what can break when someone edits the endpoint.