Streaming: Receive Responses in Real Time

Streaming lets you receive the response token by token instead of waiting for the entire generation to finish. This isn’t just a UI polish: it’s what separates a chatbot that feels sluggish from one that feels snappy, even when server performance is identical.

Why Stream

Without streaming, users sometimes stare at a blank screen for ten to thirty seconds before the full response appears at once. With streaming, the first words appear in under a second, and reading begins while generation continues. The total duration is the same; the perception is not. Enabling it requires just one parameter, stream=True, which turns the return value of create() into an event iterator.

from openai import OpenAI
client = OpenAI()

# WITHOUT streaming — wait for everything
response = client.responses.create(
    model="gpt-5.6-terra",
    input="Write a paragraph about artificial intelligence."
)
print(response.output_text)  # Appears all at once after the wait

# WITH streaming — token by token
stream = client.responses.create(
    model="gpt-5.6-terra",
    input="Write a paragraph about artificial intelligence.",
    stream=True
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
print()  # New line at the end

Typed Events, Not an Undifferentiated Stream

The stream doesn’t just carry text. It notifies you when the response is created, when each output item is added, when fragments arrive, and when completion occurs—the final event includes usage metrics. You filter on event.type to know what you’re handling. This structure lets you, for example, display a loading indicator as soon as response.created fires and tally tokens at the end without a second API call.

stream = client.responses.create(
    model="gpt-5.6-terra",
    input="Hello!",
    stream=True
)

for event in stream:
    match event.type:
        case "response.created":
            print(f"[Response created] ID: {event.response.id}")
        case "response.output_item.added":
            print(f"[New output item]")
        case "response.output_text.delta":
            print(f"[Text] {event.delta}", end="")
        case "response.output_text.done":
            print(f"\n[Text complete]")
        case "response.completed":
            print(f"[Done] Tokens: {event.response.usage.total_tokens}")

# Output:
# [Response created] ID: resp_abc123
# [New output item]
# [Text] Hello[Text] ! [Text] How[Text] can[Text] I...
# [Text complete]
# [Done] Tokens: 25

In a real application, you’ll almost always want to do two things at once: display the fragments and reconstruct the full text so you can archive it or return it to the caller. The rule is simple: accumulate as you iterate, because once the stream is consumed, it cannot be rewound.

def stream_and_collect(prompt: str) -> str:
    """Stream the response while collecting it."""
    stream = client.responses.create(
        model="gpt-5.6-terra",
        input=prompt,
        stream=True
    )

    full_text = ""
    for event in stream:
        if event.type == "response.output_text.delta":
            full_text += event.delta
            print(event.delta, end="", flush=True)

    print()  # New line
    return full_text

result = stream_and_collect("List 3 European capitals.")
print(f"\nCollected text ({len(result)} characters)")

Streaming and Function Calling

Function calls are also streamed, with their own events: arguments arrive in fragments via response.function_call_arguments.delta, and response.function_call_arguments.done signals that the JSON is complete. This has an important practical consequence—never try to deserialize the arguments before the done event, or you’ll only have truncated JSON.

stream = client.responses.create(
    model="gpt-5.6-terra",
    input="What time is it in Tokyo?",
    tools=[{
        "type": "function",
        "name": "get_time",
        "description": "Get the time in a city",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string"}
            },
            "required": ["city"]
        }
    }],
    stream=True
)

for event in stream:
    match event.type:
        case "response.output_text.delta":
            print(event.delta, end="")
        case "response.function_call_arguments.delta":
            print(f"[Args] {event.delta}", end="")
        case "response.function_call_arguments.done":
            print(f"\n[Function call complete]")

# Output:
# [Args] {"city": "Tokyo"}
# [Function call complete]

The Asynchronous Case

As soon as your application serves multiple users simultaneously, switch to AsyncOpenAI: the synchronous client would block the event loop for the entire generation, causing your other requests to wait. The syntax differs by just two keywords—await on creation and async for on iteration.

from openai import AsyncOpenAI
import asyncio

async_client = AsyncOpenAI()

async def stream_async(prompt: str):
    """Asynchronous streaming of the response."""
    stream = await async_client.responses.create(
        model="gpt-5.6-terra",
        input=prompt,
        stream=True
    )

    async for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
    print()

asyncio.run(stream_async("Tell a short joke."))

On the web, streaming is useless if it stops at your server: you need to extend it to the browser. FastAPI enables this with StreamingResponse, fed by a generator that relays fragments as they arrive.

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

app = FastAPI()
client = OpenAI()

@app.get("/chat")
async def chat(question: str):
    def generate():
        stream = client.responses.create(
            model="gpt-5.6-terra",
            input=question,
            stream=True
        )
        for event in stream:
            if event.type == "response.output_text.delta":
                yield event.delta

    return StreamingResponse(generate(), media_type="text/plain")

Errors Mid-Stream

One detail deserves your attention: in streaming, an error can occur after fragments have already been displayed. The user then sees a response that starts but gets cut off. Therefore, wrap the iteration itself, not just the stream creation, so you can cleanly signal the interruption rather than leaving a sentence hanging.

from openai import APIError

def stream_with_errors(prompt: str):
    """Streaming with robust error handling."""
    try:
        stream = client.responses.create(
            model="gpt-5.6-terra",
            input=prompt,
            stream=True
        )

        for event in stream:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
        print()

    except APIError as e:
        print(f"\nAPI error during streaming: {e.message}")
    except Exception as e:
        print(f"\nUnexpected error: {e}")

stream_with_errors("Explain streaming in 2 sentences.")

Key Takeaways

Enable streaming with stream=True for a real-time experience
Filter events by event.type to handle each data type correctly
response.output_text.delta contains text fragments
response.completed signals the end and includes usage metrics
Use AsyncOpenAI for streaming in asynchronous applications
Always handle errors, even in streaming mode