A Stream Carries More Than Just Text Deltas

Lesson 2 focused on the single most common event type — response.output_text.delta — since displaying incrementally arriving text is streaming's primary use case. A real streamed response, however, carries a richer sequence of event types, marking the lifecycle of the request: creation, progress, completion, and several more specific signals in between. Understanding this full sequence matters for building a genuinely robust streaming integration, rather than one that happens to work as long as nothing unusual occurs.

def log_all_event_types(prompt: str) -> None:
    """Print every distinct event type encountered during one streamed request,
    to see the actual lifecycle a stream goes through."""
    seen_types = []
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
    for event in stream:
        if event.type not in seen_types:
            seen_types.append(event.type)
    for t in seen_types:
        print(t)

log_all_event_types("Write two sentences about volcanoes.")

Running this against a real request typically surfaces a sequence resembling: response.created, one or more response.output_text.delta events, response.output_text.done, and response.completed — with additional event types appearing for more complex responses, such as those involving function calls (Unit 8) or multiple output segments.

The Core Lifecycle Events

response.created fires once, immediately, signaling that the request has been accepted and generation has begun. This is a useful hook for starting a "typing" or "generating" indicator in a user interface, distinct from the moment the first actual content arrives.

for event in stream:
    if event.type == "response.created":
        show_typing_indicator()
    elif event.type == "response.output_text.delta":
        hide_typing_indicator_if_visible()
        display_text(event.delta)

response.output_text.delta fires repeatedly, once per chunk of generated text, exactly as Lesson 2 covered — this is the workhorse event for any application whose primary goal is displaying text progressively.

response.output_text.done fires once a particular text output segment has finished generating completely, carrying the full, final text of that segment as a convenience — useful when you want the complete text of one output block without needing to have manually accumulated every delta yourself.

final_text = None
for event in stream:
    if event.type == "response.output_text.done":
        final_text = event.text  # the complete text for this output segment, already assembled

This event is worth knowing about specifically because it can simplify code that would otherwise need to maintain its own accumulator purely to reconstruct the full text — if all you need is the complete final text once generation finishes, and you don't need to display anything incrementally along the way, listening only for response.output_text.done is simpler than manually summing every delta.

response.completed fires once, at the very end, signaling that the entire response — potentially including multiple output segments, tool calls, or other content types beyond plain text — has finished. This event carries the complete, final response object, exactly the same shape as what a non-streaming call would have returned directly.

final_response = None
for event in stream:
    if event.type == "response.completed":
        final_response = event.response

print(f"Total tokens used: {final_response.usage.output_tokens}")
print(f"Response ID for chaining: {final_response.id}")

This is the event to listen for when you need anything that lives on the complete response object but isn't part of the streamed text itself — the response ID (needed for Unit 4, Lesson 3's chaining, as Lesson 2 of this unit demonstrated), the final usage statistics (needed for cost tracking, as in Unit 4, Lesson 5's project extension), or any other response-level metadata.

Handling Errors Mid-Stream

Beyond the successful lifecycle events, a stream can also emit an explicit error event if something goes wrong during generation — distinct from an exception being raised by the iteration itself, which Lesson 2 covered. Checking for this event type, where the SDK version supports it, lets your application distinguish a clean, successful completion from one that ended in an error state the server reported explicitly.

def stream_with_explicit_error_check(prompt: str) -> dict:
    collected = ""
    error_info = None
    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
        elif event.type == "response.error":
            error_info = event.error
            break

    return {"text": collected, "error": error_info}

result = stream_with_explicit_error_check("Write a short poem about autumn.")
if result["error"]:
    print(f"Stream reported an error: {result['error']}")
else:
    print(result["text"])

Distinguishing an explicit response.error event from a raised Python exception matters because the two represent different failure origins: an exception typically indicates a problem at the transport or client level (a dropped connection, a timeout), while an explicit error event indicates the server itself began processing the request and then encountered a problem partway through generation — both are worth handling, but they may call for different retry or fallback strategies (Unit 12 covers this systematically), and conflating them into identical handling can obscure which failure mode is actually occurring in production.

Filtering to Only the Events You Need

A well-structured streaming consumer typically only reacts to the small subset of event types actually relevant to its purpose, ignoring everything else explicitly rather than accidentally mishandling an event type it wasn't designed to expect. This is best expressed as a clear, exhaustive-feeling conditional structure rather than a single check for the one event type a first implementation happened to need.

def handle_stream(prompt: str, on_start=None, on_delta=None, on_done=None, on_error=None) -> None:
    """A general-purpose stream handler dispatching to whichever callbacks the
    caller actually cares about, ignoring event types none of them handle."""
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)
    for event in stream:
        if event.type == "response.created" and on_start:
            on_start()
        elif event.type == "response.output_text.delta" and on_delta:
            on_delta(event.delta)
        elif event.type == "response.completed" and on_done:
            on_done(event.response)
        elif event.type == "response.error" and on_error:
            on_error(event.error)
        # Any other event type is silently ignored by design, not by oversight —
        # a genuinely comprehensive handler should still know what it's choosing not to act on.

handle_stream(
    "Explain the greenhouse effect.",
    on_start=lambda: print("[generating...]"),
    on_delta=lambda text: print(text, end="", flush=True),
    on_done=lambda response: print(f"\n[done, id={response.id}]"),
    on_error=lambda err: print(f"\n[error: {err}]"),
)

Structuring event handling this way — as a set of named, optional callbacks dispatched from one central loop — keeps a streaming consumer's logic organized and testable even as the number of event types an application cares about grows over time, and it makes explicit, in one place, exactly which events the application is choosing to act on versus silently pass over.

Handling Multiple Output Segments

A response can, in some cases, be composed of more than one output segment — for instance, a reasoning model's response (Unit 3, Lesson 4) may internally represent its reasoning process and its visible answer as conceptually separate segments, or a response involving a tool call (Unit 8) includes segments representing the tool call itself distinct from the final text answer. When this applies, delta events typically carry an index or identifier indicating which output segment they belong to, and a careful streaming consumer needs to track this rather than assuming every delta belongs to one single, undifferentiated block of text.

def stream_multi_segment(prompt: str, **kwargs) -> dict[int, str]:
    """Accumulate text per output segment, rather than assuming a single flat stream of text."""
    segments: dict[int, str] = {}
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True, **kwargs)
    for event in stream:
        if event.type == "response.output_text.delta":
            index = getattr(event, "output_index", 0)
            segments.setdefault(index, "")
            segments[index] += event.delta
    return segments

For a plain text-only response, there will typically be exactly one segment (index 0), and this distinction is invisible. It becomes directly relevant once a response involves multiple distinct output items — Unit 8's function calling and Unit 11's Agents SDK both introduce scenarios where a single response genuinely produces more than one kind of output, and a streaming consumer built only around Lesson 2's simplest single-accumulator pattern would silently mix content from different segments together if it isn't tracking segment identity explicitly.

Building a Reusable Streaming Response Handler

Pulling this lesson's patterns together, a small, reusable class captures the common lifecycle handling any streaming feature is likely to need, giving the rest of an application a clean interface rather than requiring every call site to re-implement event dispatching from scratch.

class StreamHandler:
    def __init__(self):
        self.text = ""
        self.response_id: str | None = None
        self.usage = None
        self.error = None

    def consume(self, stream, on_delta=None) -> "StreamHandler":
        for event in stream:
            if event.type == "response.output_text.delta":
                self.text += event.delta
                if on_delta:
                    on_delta(event.delta)
            elif event.type == "response.completed":
                self.response_id = event.response.id
                self.usage = event.response.usage
            elif event.type == "response.error":
                self.error = event.error
        return self


handler = StreamHandler().consume(
    client.responses.create(model="gpt-5.6-luna", input="Tell me a fun fact about otters.", stream=True),
    on_delta=lambda chunk: print(chunk, end="", flush=True),
)
print(f"\n\nResponse ID: {handler.response_id}")
print(f"Output tokens: {handler.usage.output_tokens if handler.usage else 'n/a'}")

This StreamHandler class packages exactly the information a typical application needs after a streaming call completes — the full text, the response ID for chaining, the usage statistics for cost tracking — behind one small, reusable object, rather than requiring every part of an application that streams a response to re-derive this bookkeeping independently.

A Quick Reference Table of Event Types

Note: Exact event type names and the precise set of events emitted can vary by SDK version and by what kind of response is being generated (plain text vs. one involving tool calls or multiple segments) — treat this table as a map of the concepts to expect, and confirm exact names against your installed SDK version's documentation or by running the log_all_event_types() exploration function from earlier in this lesson against a live request.

This table is worth keeping close at hand when first building a new streaming feature, since it's easy to reach only for response.output_text.delta (the event every "hello world" streaming example demonstrates) and miss that the response ID, usage statistics, and error details all live on different, less obvious events entirely.

Testing Event-Type Handling with Fake Streams

Following the same fake-stream testing pattern Lesson 2 introduced, a more complete fake stream — one that emits the full lifecycle of event types, not just deltas — lets you verify that a stream handler correctly reacts to response.completed and response.error, not only to text deltas.

class FakeEvent:
    def __init__(self, type_: str, **attrs):
        self.type = type_
        for k, v in attrs.items():
            setattr(self, k, v)

class FakeResponse:
    def __init__(self, id_: str, output_tokens: int):
        self.id = id_
        class Usage:
            pass
        usage = Usage()
        usage.output_tokens = output_tokens
        self.usage = usage

def fake_full_lifecycle_stream():
    yield FakeEvent("response.created")
    yield FakeEvent("response.output_text.delta", delta="Hello")
    yield FakeEvent("response.output_text.delta", delta=", world!")
    yield FakeEvent("response.completed", response=FakeResponse("resp_fake_123", 42))

def test_stream_handler_captures_completion_data():
    handler = StreamHandler().consume(fake_full_lifecycle_stream())
    assert handler.text == "Hello, world!"
    assert handler.response_id == "resp_fake_123"
    assert handler.usage.output_tokens == 42
    print("PASS: StreamHandler correctly captures text, response ID, and usage")

def test_stream_handler_captures_error():
    def fake_error_stream():
        yield FakeEvent("response.output_text.delta", delta="Partial")
        yield FakeEvent("response.error", error="simulated failure")

    handler = StreamHandler().consume(fake_error_stream())
    assert handler.text == "Partial"
    assert handler.error == "simulated failure"
    print("PASS: StreamHandler correctly captures partial text and error")

test_stream_handler_captures_completion_data()
test_stream_handler_captures_error()

These tests exercise exactly the parts of StreamHandler that are easy to get wrong and hard to verify against a live API call reliably (since you can't easily force a live request to fail partway through on demand) — a fake stream lets you construct precisely the event sequence you want to test against, including failure scenarios that would otherwise require unreliable manual reproduction against the real API.

Debugging an Unexpected Event Sequence

When a streaming feature behaves unexpectedly — content appears out of order, the response ID is missing, an error isn't caught — the most direct diagnostic step is the same log_all_event_types()-style exploration shown at the start of this lesson, applied to the specific failing request rather than a generic example, ideally capturing not just the type but the full attributes of each event.

def debug_full_stream(prompt: str, **kwargs) -> None:
    """Print full event details, not just type names, for deep debugging of
    an unexpected streaming interaction."""
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True, **kwargs)
    for i, event in enumerate(stream):
        attrs = {k: v for k, v in vars(event).items() if not k.startswith("_")}
        print(f"[{i}] {event.type}: {attrs}")

Running this against the exact prompt and parameters that produced unexpected behavior — rather than a simplified reproduction — often reveals the actual cause quickly: perhaps an event type your handler doesn't recognize is appearing (a sign the SDK version emits something your code wasn't written to expect), or events are arriving in a different order than assumed, or a segment index is present where your handler assumed there would only ever be one segment. This is directly analogous to Unit 3, Lesson 5's general debugging discipline of reading the actual token-level or event-level evidence rather than guessing at a cause from the visible symptom alone.

A Worked Example: A Progress Indicator from Lifecycle Events

To tie the lifecycle events together in a realistic small feature, consider a command-line progress indicator that shows distinct phases of a streaming request — waiting, generating, and done — using the events this lesson has covered.

import sys
import time

def stream_with_progress_indicator(prompt: str) -> str:
    collected = ""
    phase = "waiting"

    def set_phase(new_phase: str) -> None:
        nonlocal phase
        phase = new_phase
        sys.stderr.write(f"\r[{phase}]" + " " * 20 + "\r")
        sys.stderr.flush()

    set_phase("waiting")
    stream = client.responses.create(model="gpt-5.6-luna", input=prompt, stream=True)

    for event in stream:
        if event.type == "response.created":
            set_phase("generating")
        elif event.type == "response.output_text.delta":
            collected += event.delta
            print(event.delta, end="", flush=True)
        elif event.type == "response.completed":
            set_phase("done")
        elif event.type == "response.error":
            set_phase("error")

    print()
    return collected

stream_with_progress_indicator("List three benefits of regular exercise.")

Writing the phase indicator to sys.stderr rather than sys.stdout is a deliberate choice here: it keeps the status indicator separate from the actual generated content being printed to standard output, a pattern common in command-line tools that need to show transient status information alongside a program's real, meaningful output — useful to know if this project pattern is extended into a more polished command-line tool than the basic examples shown so far.

Common Mistakes

Only handling response.output_text.delta and ignoring response.completed entirely, then being surprised that the response ID (needed for Unit 4's chaining) or usage statistics aren't available anywhere in the code. These live specifically on the completed response, delivered via response.completed, not reconstructable from delta events alone.

Treating every delta as belonging to a single undifferentiated text stream, without checking for segment or output-index information, in an application that may eventually involve function calling (Unit 8) or other multi-segment responses — this works fine until the first time a response genuinely has more than one segment, at which point content from different segments can be silently interleaved incorrectly.

Conflating a raised exception with an explicit response.error event, handling both identically without distinguishing a transport-level failure from a server-reported generation error — these can call for different remediation strategies and are worth telling apart in production error handling.

Writing a streaming consumer that reacts to exactly one event type with no explicit handling (even a deliberate no-op) for the others, making it unclear later whether an unhandled event type was intentionally ignored or simply overlooked during initial development — the handle_stream() pattern's explicit "ignored by design" comment above is a small habit that pays off when revisiting the code later or when a new engineer works on it.

Best Practices

Build a small, reusable stream-handling utility (a function or class) for your application, rather than re-implementing event dispatching inline at every call site — this keeps behavior consistent and makes it easy to add handling for a new event type in one place as your application's needs grow.

Explicitly capture the response ID and usage statistics from the response.completed event whenever your application needs either, rather than assuming they can be derived from delta events, since they genuinely cannot.

Track output segment identity for any response that might involve function calling, reasoning, or other multi-part output, even if your current use case only ever produces a single segment — this future-proofs a streaming consumer against a class of subtle bugs that only appears once a more complex response type is introduced.

Distinguish transport-level exceptions from explicit server-reported error events in your error handling, and consider logging them separately, so production debugging can quickly tell which category a given failure belongs to.

Write tests against fake streams that exercise the full event lifecycle, not just the happy-path text delta case. As shown above, constructing a fake stream that includes a response.error event or a response.completed event with specific usage figures lets you verify error handling and metadata capture deterministically, without needing to coax a live API call into failing on demand for testing purposes.

Log full event details, not just event type names, when debugging an unexpected streaming interaction. A type name alone often isn't enough to diagnose why a stream behaved unexpectedly — the full attributes of each event (as debug_full_stream() prints above) frequently reveal the actual cause, such as an unrecognized event type or an unexpected segment index, that a bare list of type names would leave invisible.

Why This Level of Detail Matters

It might seem like overkill, when first encountering streaming, to build out handling for response.created, response.output_text.done, response.error, and multi-segment tracking, when a minimal implementation only checking for response.output_text.delta appears to work in a first demo. The reason this lesson has gone through the fuller event lifecycle deliberately is that the gap between "works in a simple demo" and "works reliably in production" is almost entirely made up of exactly these less-obvious event types: the response ID a chained conversation (Unit 4, Lesson 3) needs to continue correctly, the usage statistics a cost-conscious application (Unit 4, Lesson 1's cost-growth argument, Unit 4, Lesson 5's cost-tracking extension) needs to monitor, the explicit error signal a robust user experience needs to handle gracefully, and the segment tracking a more complex response type (Units 8 and 11) will eventually require. A streaming implementation built with only response.output_text.delta in mind will need to be revisited and expanded the moment any of these needs arises in a real application — building the fuller event-handling structure from the start, even for a first, simple feature, avoids that later rework and produces code that scales naturally as an application's requirements grow beyond a basic chat demo.