stream=True and Iterating Over Events

Turning On Streaming

Every example in this course so far has called client.responses.create() and received a single, complete response object back. Adding stream=True changes what the call returns: instead of a completed response, you get back an iterable stream of events, each representing a small piece of what's happening as the model generates its output.

stream = client.responses.create(
    model="gpt-5.6-luna",
    input="Write a short paragraph about the water cycle.",
    stream=True,
)

for event in stream:
    print(event.type)

Running this prints a sequence of event type names rather than a single block of text — something like response.created, then several response.output_text.delta events, then eventually response.completed. This is the fundamental shape change streaming introduces: a single request/response exchange becomes a request followed by a sequence of incremental updates, each of which your code processes as it arrives, rather than all at once at the end.

The Event Stream Is Just an Iterator

Mechanically, the object returned when stream=True is set behaves like any other Python iterable — you consume it with a for loop, exactly as you would iterate over a list or a file's lines. Each iteration yields one event object, and the loop naturally ends once the model has finished generating and the final event has been delivered.

def collect_streamed_text(prompt: str) -> str:
    """The most basic streaming consumption pattern: accumulate text as it arrives."""
    collected = ""
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
    for event in stream:
        if event.type == "response.output_text.delta":
            collected += event.delta
    return collected

result = collect_streamed_text("Explain what a semaphore is in concurrent programming.")
print(result)

This function produces a result functionally equivalent to what a non-streaming call's response.output_text would give you — the difference is entirely in how that text became available to your code: incrementally, piece by piece, as each response.output_text.delta event arrived, rather than all at once. Lesson 3 of this unit covers the full set of event types and what each represents; this lesson focuses on the mechanical pattern of consuming the stream correctly, regardless of which specific events you care about.

Displaying Streamed Content as It Arrives

The real value of streaming, per Lesson 1's argument, comes from displaying content to a user progressively rather than accumulating it silently and displaying it all at once — which defeats the entire purpose. A minimal terminal-based example makes this concrete:

def stream_to_terminal(prompt: str) -> None:
    """Print each piece of text as it arrives, rather than waiting to accumulate everything."""
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
    print()  # final newline once the stream completes

stream_to_terminal("Describe three interesting facts about deep-sea creatures.")

The end="" argument prevents each print() call from adding its own newline (which would otherwise break the output into one line per delta chunk rather than a continuous flow of text), and flush=True forces each piece to be written to the terminal immediately rather than being buffered — without flush=True, Python's standard output buffering can hold text in memory and only display it in larger batches, silently defeating the progressive-display effect streaming is meant to achieve. This is a genuinely easy mistake to make when first implementing streaming output to a terminal or a simple log, and it's worth checking for explicitly if a "streaming" implementation appears to display text in unexpectedly large chunks rather than smoothly.

Streaming to a Web Client

A terminal is a useful first example, but most real applications need to stream content to a web browser rather than a local terminal. The general pattern — consume events as they arrive, forward each piece to the client immediately — stays the same, but the mechanism for "forwarding" changes to whatever your web framework's streaming response support looks like. Using a simple example with Python's Flask framework and Server-Sent Events (SSE), a common technique for streaming text to a browser:

from flask import Flask, Response, request

app = Flask(__name__)

@app.route("/chat", methods=["POST"])
def chat():
    user_message = request.json["message"]

    def generate():
        stream = client.responses.create(
            model="gpt-5.6-luna",
            input=user_message,
            stream=True,
        )
        for event in stream:
            if event.type == "response.output_text.delta":
                yield f"data: {event.delta}\n\n"
        yield "data: [DONE]\n\n"

    return Response(generate(), mimetype="text/event-stream")

The generate() function here is itself a Python generator — it yields each piece of text formatted as an SSE message, and Flask's Response object streams those yields to the connected browser incrementally, rather than waiting for generate() to finish entirely before sending anything. This is the same underlying principle as the terminal example: consume the model's stream, and forward each piece onward immediately, rather than accumulating it server-side and sending one large response at the end — doing the latter would silently convert a "streaming" backend implementation back into an effectively non-streaming user experience, since the browser would still receive everything at once regardless of how the server internally consumed the model's output.

Accumulating Text While Also Streaming It

A common, practical need: display content to the user progressively (for the responsiveness benefit) while also keeping the complete final text available afterward, for logging, for saving to a conversation history (Unit 4's memory mechanisms), or for further processing once generation completes.

def stream_and_accumulate(prompt: str, on_delta) -> str:
    """Call on_delta(text_chunk) for each piece as it arrives, while also
    building up and returning the complete text once streaming finishes."""
    collected = ""
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
    for event in stream:
        if event.type == "response.output_text.delta":
            on_delta(event.delta)
            collected += event.delta
    return collected

full_text = stream_and_accumulate(
    "Write a two-paragraph summary of the causes of World War I.",
    on_delta=lambda chunk: print(chunk, end="", flush=True),
)

print(f"\n\n--- Full text is {len(full_text)} characters, ready to save or process further ---")

This pattern — a callback invoked per chunk, plus a returned complete accumulation — is a common shape for streaming utility functions specifically because it serves both needs at once: the callback handles the real-time display responsibility, while the returned complete string handles everything that needs to happen after generation is done, such as appending the finished response to a Unit 4 conversation history (Conversation.send() or ChainedConversation.send() from that unit's lessons would use this exact pattern internally to both stream to a user and correctly update stored conversation state).

Error Handling During a Stream

A stream can fail partway through — a network interruption, a server-side error mid-generation — and this failure mode is meaningfully different from a non-streaming call's failure, because some content may have already been delivered and displayed to the user before the failure occurred. Handling this gracefully means catching exceptions around the iteration itself, not only around the initial call.

def stream_with_error_handling(prompt: str) -> str:
    collected = ""
    try:
        stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
        for event in stream:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
                collected += event.delta
    except Exception as e:
        print(f"\n[stream interrupted: {e}]")
        # collected still holds whatever was successfully received before the failure
    return collected

Because collected accumulates incrementally as the loop runs, an exception raised partway through the for loop still leaves collected holding whatever text arrived before the interruption — a meaningful difference from a non-streaming call, where a failure produces nothing usable at all, since the entire response was pending as a single unit. This partial-success behavior is worth handling deliberately in a production application: a user-facing chat interface, for instance, might reasonably display a "response was interrupted" notice after whatever partial text did arrive, rather than discarding it entirely and showing only a generic error.

Consuming a Stream Asynchronously

For an application built on Python's async/await model — a common choice for a web backend handling many concurrent requests — the streaming interface has an async equivalent, following the same iteration pattern but using async for instead of a plain for loop.

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def stream_async(prompt: str) -> str:
    collected = ""
    stream = await async_client.responses.create(
        model="gpt-5.6-luna", input=prompt, stream=True,
    )
    async for event in stream:
        if event.type == "response.output_text.delta":
            collected += event.delta
    return collected

result = asyncio.run(stream_async("Explain the difference between TCP and UDP."))
print(result)

Unit 12 covers async clients and concurrency more thoroughly, including why an async approach matters for an application handling many simultaneous streaming connections, but it's worth knowing at this point that the streaming interface itself has both a synchronous and an asynchronous form, and the choice between them follows the same considerations as choosing between the synchronous and asynchronous client generally — not something specific to streaming itself.

Properly Closing a Stream

A stream that is only partially consumed — because your code broke out of the loop early, following the early-cancellation pattern from Lesson 1 — should be closed explicitly to release the underlying network connection promptly, rather than relying on it eventually being garbage-collected.

def stream_until_condition(prompt: str, stop_after_chars: int) -> str:
    collected = ""
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
    try:
        for event in stream:
            if event.type == "response.output_text.delta":
                collected += event.delta
                if len(collected) >= stop_after_chars:
                    break
    finally:
        stream.close()  # release the connection regardless of how the loop exited
    return collected

Wrapping the loop in a try/finally and explicitly calling stream.close() ensures the underlying connection is released whether the loop completes normally, breaks early, or exits due to an exception — a small habit that matters more as an application's request volume grows, since leaving many partially-consumed streams to be cleaned up only by garbage collection can tie up connection resources longer than necessary under real load. Many streaming clients also support being used as a context manager directly (with client.responses.create(..., stream=True) as stream:), which handles this closing automatically — check your specific SDK version's supported usage pattern and prefer the context-manager form when it's available, since it removes the risk of forgetting the explicit close() call in a finally block.

Chunking Behavior: What Arrives in Each Delta

It's worth setting a realistic expectation about the granularity of response.output_text.delta events: each one is not guaranteed to correspond to a single word, a single token, or any other fixed linguistic unit — the chunking is an implementation detail of the underlying transport and generation process, and can vary between a few characters and a longer phrase from one delta event to the next.

def inspect_chunking(prompt: str, max_events: int = 15) -> None:
    """Print each delta's exact content and length, to see chunking behavior directly."""
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
    count = 0
    for event in stream:
        if event.type == "response.output_text.delta":
            print(f"delta {count}: {event.delta!r} (length {len(event.delta)})")
            count += 1
            if count >= max_events:
                stream.close()
                break

inspect_chunking("Write one sentence about the ocean.")

Running this reveals that delta chunks don't reliably align with word boundaries — a chunk might end mid-word, and the next chunk continues it. This has a direct, practical implication: code that processes streamed text (looking for a specific keyword, checking whether a sentence has ended, as the early-cancellation example in Lesson 1 did with .endswith(".")) needs to operate on the accumulated text built up so far, not on each individual delta chunk in isolation, since a single delta chunk in isolation may contain a meaningless, arbitrarily-split fragment of a word or phrase.

# Wrong: checking each raw chunk in isolation can miss a condition split across chunks
for event in stream:
    if event.type == "response.output_text.delta":
        if "important" in event.delta:  # misses "impor" + "tant" split across two deltas
            handle_important_content()

# Right: check against the accumulated text
collected = ""
for event in stream:
    if event.type == "response.output_text.delta":
        collected += event.delta
        if "important" in collected:  # correctly catches the word regardless of chunk boundaries
            handle_important_content()

Combining Streaming with Instructions and Conversation Memory

Streaming is an orthogonal concern to everything covered in Units 3 and 4 — instructions, few-shot examples, and any of the three memory mechanisms all continue to work exactly as described in those units, simply with stream=True added and the response consumed as an event sequence rather than a single object.

def stream_chained_turn(prompt: str, instructions: str, previous_response_id: str | None) -> tuple[str, str]:
    """A streaming version of Unit 4 Lesson 3's chaining pattern — returns the
    full text and the new response ID once streaming completes."""
    kwargs = {"model": "gpt-5.6-luna", "instructions": instructions, "input": prompt, "stream": True}
    if previous_response_id:
        kwargs["previous_response_id"] = previous_response_id

    collected = ""
    response_id = None
    stream = client.responses.create(**kwargs)
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
            collected += event.delta
        elif event.type == "response.completed":
            response_id = event.response.id  # the completed response's ID, for chaining forward

    print()
    return collected, response_id


last_id = None
text, last_id = stream_chained_turn("My name is Priya.", "Be friendly and concise.", last_id)
text, last_id = stream_chained_turn("What's my name?", "Be friendly and concise.", last_id)

Notice that the response's id — needed to continue a previous_response_id chain, exactly as Unit 4, Lesson 3 described — becomes available specifically through the response.completed event once streaming finishes, rather than being available immediately the way it would be on a plain, non-streaming response object. This is an important practical detail: an application streaming a chained conversation needs to wait for the stream to fully complete before it can capture the ID needed for the next call in the chain, which Lesson 3 of this unit will cover in more detail as part of its full event-type reference.

Testing Streaming Code Without a Live Connection

Following the dependency-injection testing pattern used throughout this course, streaming consumption logic can be tested by substituting a fake stream — a plain Python generator yielding fake event objects — in place of an actual API call, letting you verify your event-handling logic quickly and deterministically.

class FakeDeltaEvent:
    def __init__(self, delta: str):
        self.type = "response.output_text.delta"
        self.delta = delta

def fake_stream(chunks: list[str]):
    """A generator standing in for a real streamed response, for testing."""
    for chunk in chunks:
        yield FakeDeltaEvent(chunk)

def collect_from_stream(stream) -> str:
    """The logic under test — accepts any iterable of delta-shaped events."""
    collected = ""
    for event in stream:
        if event.type == "response.output_text.delta":
            collected += event.delta
    return collected

def test_collect_from_stream_handles_split_words():
    fake = fake_stream(["The wat", "er cycle ", "is import", "ant."])
    result = collect_from_stream(fake)
    assert result == "The water cycle is important."
    print("PASS: correctly reassembles text split arbitrarily across chunks")

test_collect_from_stream_handles_split_words()

Structuring collect_from_stream() to accept any iterable of event-shaped objects (rather than being hardwired to call the API directly) is what makes this substitution possible — the same principle behind the dependency-injection tests in Unit 4, applied here to streaming-specific logic, letting you verify chunk-reassembly and event-handling correctness without needing a live API key or incurring any cost for every test run.

Common Mistakes

Forgetting to flush output when streaming to a terminal or a simple log, resulting in text appearing in large, delayed batches rather than smoothly, defeating the visible-progress benefit streaming is meant to provide. Always verify visually that a "streaming" implementation is actually displaying content incrementally, not just internally consuming it incrementally while still presenting it all at once.

Accumulating the full response server-side before forwarding anything to the client, as in a web backend that consumes the entire model stream into a string and only then sends a single, complete HTTP response to the browser. This silently converts a streaming backend into a non-streaming user experience — the actual benefit only exists if content is forwarded onward as it's received, at every layer of the pipeline, not just at the point where the model's own output is consumed.

Not handling exceptions raised during stream iteration, treating a stream exactly like a non-streaming call in terms of error handling and losing the opportunity to gracefully preserve and communicate whatever partial content was successfully delivered before an interruption.

Mixing synchronous and asynchronous streaming patterns inconsistently within one application, using async for in some places and a plain for loop with the synchronous client in others without a clear architectural reason, which tends to produce confusing, hard-to-maintain code as an application grows.

Best Practices

Always flush output explicitly when streaming to a terminal, log, or any other buffered destination, and verify visually (not just by reading the code) that content actually appears incrementally rather than in unexpected batches.

Design streaming utility functions to serve both the real-time display need and the need for the complete final text, using a callback-plus-return-value pattern (or an equivalent) so a single streaming call can drive a chat UI's live display while also producing text ready for Unit 4's conversation-history mechanisms once generation completes.

Handle exceptions around stream iteration explicitly, not just around the initial API call, and design your application's error handling to make good use of whatever partial content a stream successfully delivered before an interruption, rather than discarding it.

Choose synchronous or asynchronous streaming deliberately, based on your application's overall concurrency architecture, rather than mixing the two patterns without a clear reason — Unit 12 covers this decision in more depth once error handling, retries, and concurrency are all in view together.

Always operate on accumulated text, not individual delta chunks, when checking for a specific word, phrase, or condition within streamed content. As shown above, chunk boundaries do not respect word or sentence boundaries, and code that inspects each chunk in isolation will intermittently and unpredictably miss a condition that happens to be split across two adjacent chunks — a bug that is particularly frustrating to track down because it depends on chunking behavior that isn't fully within your control and can vary from one request to the next.

Explicitly close a stream whenever your code stops consuming it before it completes naturally, using a try/finally block or your SDK's context-manager support, to release the underlying connection promptly rather than relying on garbage collection — a habit that costs little to build in from the start and avoids a class of resource-leak issues that only becomes visible under real production load.

Capture the completed response's ID from the response.completed event when chaining streamed conversation turns, per Unit 4, Lesson 3's mechanism — remember that, unlike a non-streaming call, this ID is not available until the stream has fully finished, so any logic that needs it (storing it for the next chained call, logging it) must run after the iteration loop completes, not during it.

A Note on SDK Version Differences

The exact event type names, the precise shape of each event object, and the supported context-manager or closing patterns for streaming are all details that can shift between SDK versions more readily than the higher-level concepts this lesson covers. The pattern of iterating over a stream of typed events, accumulating text incrementally, and closing a stream you've stopped consuming early is stable across versions in spirit even when specific attribute names evolve — when working against a specific installed SDK version, it's worth confirming the exact event type strings and object shapes against that version's own documentation or by running the inspect_chunking()-style exploration function shown above against a real request, rather than assuming the exact names used in this lesson will never change. This mirrors the general caution Unit 2, Lesson 5 raised about model names and API shapes moving faster than course material can track perfectly — the underlying mental model is durable; the exact spelling of a given attribute is worth a quick, live confirmation before depending on it in production code.