Python for JS Devs · LLM APIs

Streaming LLM responses

The ChatGPT typewriter effect. Stream tokens to the client as they arrive — using async generators and FastAPI's StreamingResponse.

A full LLM reply can take many seconds. Streaming sends each token the moment the model produces it, so the user sees output immediately. Two new pieces: the SDK's async stream, and FastAPI's StreamingResponse driven by an async generator.

1 · The SDK stream (async iteration)

The async client's .stream() is an async context manager exposing text_stream — an async iterator of text chunks. This is async for from Lesson 10 in action.

const stream = client.messages
  .stream({ ... });
for await (const ev of stream) { ... }

async with client.messages.stream(
    model="claude-opus-4-8", max_tokens=1024,
    messages=[...],
) as stream:
    async for text in stream.text_stream:
        print(text, end="", flush=True)

2 · An async generator = the response body

Python's async generator (an async def with yield) is the perfect producer for a streaming HTTP body. Wrap it in StreamingResponse and FastAPI streams each yielded chunk to the client.

import anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()
client = anthropic.AsyncAnthropic()

class ChatIn(BaseModel):
    message: str

async def token_stream(message: str):
    async with client.messages.stream(
        model="claude-opus-4-8", max_tokens=1024,
        messages=[{"role": "user", "content": message}],
    ) as stream:
        async for text in stream.text_stream:
            yield text        # each chunk flushed to the client

@app.post("/chat/stream")
async def chat_stream(body: ChatIn):
    return StreamingResponse(
        token_stream(body.message),
        media_type="text/plain",
    )

3 · Server-Sent Events (for browsers)

For a JS frontend using EventSource, format each chunk as SSE and set the media type. Same generator, tiny change:

async def sse_stream(message: str):
    async with client.messages.stream(...) as stream:
        async for text in stream.text_stream:
            yield f"data: {text}\n\n"      # SSE frame

@app.post("/chat/sse")
async def chat_sse(body: ChatIn):
    return StreamingResponse(sse_stream(body.message),
                             media_type="text/event-stream")

Check yourself