OpenAI SDK: Streaming responses

A model generates output one token at a time. Without streaming you wait for the whole answer, perhaps 8 seconds for a long reply, and see nothing meanwhile. With streaming, the server sends each piece as soon as it exists (as server-sent events), so the user sees text after a few hundred milliseconds. Total time is the same, but time to first token is what users feel.

1. Same total time, very different experience

2. stream=True: iterate over events

from fake_openai import fake_client

client = fake_client(script=["Kochi, Bengaluru and Pune had the highest revenue in October."])
stream = client.responses.create(model="gpt-5.6-luna", input="Top 3 cities by revenue?", stream=True)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)          # print each piece as it arrives
    elif event.type == "response.completed":
        usage = event.response.usage
        print(f"\n[done: {usage.input_tokens} in / {usage.output_tokens} out]")

Output

Kochi, Bengaluru and Pune had the highest revenue in October.
[done: 6 in / 15 out]

all the event types in one stream

from collections import Counter
from fake_openai import fake_client

client = fake_client(script=["Streaming is just a sequence of typed events."])
events = list(client.responses.create(model="gpt-5.6-luna", input="Explain streaming", stream=True))
print(Counter(e.type for e in events))
print([e.sequence_number for e in events][:5], "…")

Output

Counter({'response.output_text.delta': 8, 'response.created': 1, 'response.output_item.added': 1, 'response.content_part.added': 1, 'response.output_text.done': 1, 'response.content_part.done': 1, 'response.output_item.done': 1, 'response.completed': 1})
[0, 1, 2, 3, 4] …

A stream is a sequence of typed events: the response is created, an output item is added, text deltas arrive, items finish, and the response completes (or fails, with an error or response.failed event). Tool calls stream too, as argument deltas. Handle only the event types you need and ignore the rest, so new event types never break your code.

3. The helper: client.responses.stream()

from fake_openai import fake_client

client = fake_client(script=["The helper accumulates the full response for you."])
with client.responses.stream(model="gpt-5.6-luna", input="Why use the helper?") as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="")
    final = stream.get_final_response()          # a normal Response object, as if not streamed
print("\nfinal.output_text:", final.output_text)

Output

The helper accumulates the full response for you.
final.output_text: The helper accumulates the full response for you.

The context manager closes the connection even if your loop breaks early, and get_final_response() gives you the assembled Response for logging or saving. It also supports text_format= for streamed structured output (part 06).

4. Streaming to a browser

FastAPI endpoint that relays tokens to the browser (sketch)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

@app.get("/ask")
async def ask(q: str):
    async def tokens():
        stream = await client.responses.create(model="gpt-5.6-luna", input=q, stream=True)
        async for event in stream:
            if event.type == "response.output_text.delta":
                yield event.delta
    return StreamingResponse(tokens(), media_type="text/plain")

Recap