Streaming Intro

What Streaming Is

Without streaming, client.messages.create() waits until Claude has generated the entire response, then returns it all at once. For a long answer, the user sees nothing for many seconds, then everything appears.

With streaming, the response is delivered piece by piece while it is being generated. Your program receives small chunks of text, usually a few words at a time, and can show them immediately:

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain what an API is in three sentences."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

The answer appears on screen progressively, the way chat interfaces such as the Claude app display responses.

How Streaming Works

Streaming uses server-sent events (SSE), a standard way for a server to send a sequence of messages over one long-lived HTTP connection. When you request streaming ("stream": true in the raw API), the API keeps the connection open and sends events as generation progresses:

message_start          → a Message shell with empty content
content_block_start    → a new content block begins (text, thinking, tool_use, ...)
content_block_delta    → a small piece of that block's content (repeated many times)
content_block_stop     → the block is complete
   ... more content blocks ...
message_delta          → top-level changes: stop_reason and cumulative usage
message_stop           → the stream is finished

ping events can also appear at any time, to keep the connection active. Lessons 53 and 54 look at events in detail. For most applications, the SDK's helpers hide the event details completely.

Why Stream?

Two Kinds of Latency

Streaming does not make Claude generate faster. It makes the waiting visible and useful, which is what users notice.

Note: with thinking models, the first visible text may take a moment to appear, because the model can think before it starts writing the answer (Lesson 16). A brief "thinking…" indicator helps here (Lesson 53).

When to Use Streaming

Three Ways to Stream with the SDK

The Stream Helper (Recommended)

with client.messages.stream(model="claude-opus-5-5", max_tokens=1024, messages=messages) as stream:
    for text in stream.text_stream:          # Only the answer text, chunk by chunk
        print(text, end="", flush=True)
    message = stream.get_final_message()     # The complete Message, like create() returns

messages.stream() takes the same parameters as messages.create(). The with block makes sure the connection is closed properly, even if an error occurs.

Final Message Only

For long requests where you do not need to display progress, streaming can still be used internally:

with client.messages.stream(model="claude-opus-5-5", max_tokens=64000, messages=messages) as stream:
    message = stream.get_final_message()

According to the documentation, this returns a Message identical to what create() would return, and it is especially useful for large max_tokens values, where the SDK requires streaming to avoid HTTP timeouts.

Raw Events

stream = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=messages,
    stream=True,
)
for event in stream:
    print(event.type)

With stream=True, create() returns an iterator of raw events, without the helper's text iteration or message accumulation. Lesson 53 uses events in detail.

Practical Example: Measuring the Difference

This script sends the same request twice, once without streaming and once with streaming, and measures time to first text and total time for each. It shows concretely what streaming changes, and what it does not.

"""streaming_vs_blocking.py: compare perceived latency with and without streaming."""

import time

from anthropic import Anthropic

client = Anthropic()
REQUEST = {
    "model": "claude-opus-5-5",
    "max_tokens": 2048,
    "messages": [{"role": "user", "content": "Explain how HTTPS protects data, in about 250 words."}],
}


def without_streaming() -> None:
    start = time.perf_counter()
    message = client.messages.create(**REQUEST)
    total = time.perf_counter() - start
    text = "".join(b.text for b in message.content if b.type == "text")
    print(f"[blocking]  first text after {total:.1f}s, finished after {total:.1f}s, "
          f"{len(text.split())} words")


def with_streaming() -> None:
    start = time.perf_counter()
    first_text_at = None
    words = 0
    with client.messages.stream(**REQUEST) as stream:
        for text in stream.text_stream:
            if first_text_at is None and text.strip():
                first_text_at = time.perf_counter() - start
            words += len(text.split())
    total = time.perf_counter() - start
    print(f"[streaming] first text after {first_text_at:.1f}s, finished after {total:.1f}s, "
          f"~{words} words")


if __name__ == "__main__":
    without_streaming()
    with_streaming()

Example output (timings vary):

[blocking]  first text after 9.8s, finished after 9.8s, 252 words
[streaming] first text after 2.1s, finished after 9.6s, ~251 words

How It Works

1. Both functions send the same request, using the shared REQUEST dictionary, so the comparison is fair.
2. Without streaming, nothing is available until create() returns, so the first text and the full text arrive at the same moment.
3. With streaming, first_text_at records when the first non-empty chunk arrives. text.strip() skips chunks that contain only whitespace.
4. time.perf_counter() is a high-resolution timer for measuring elapsed time.
5. The word count from chunks is approximate (~), because a chunk can end in the middle of a word, which split() then counts twice or not at all. Use the final message for exact text.

The result shows the key point: the total time is about the same, but with streaming, the user starts reading after about 2 seconds instead of waiting nearly 10.

Common Mistakes

Expecting Streaming to Make Generation Faster

What happens: disappointment when total time does not change.

Fix: use streaming to improve perceived latency and reliability. To reduce total time, shorten outputs, lower effort, or choose a faster model.

Forgetting flush=True

What happens: text appears in bursts, or only at the end, because Python buffers output.

Fix: print chunks with print(text, end="", flush=True).

Not Using the with Block

What happens: connections may not be closed properly when errors occur, or when your code stops reading early.

Fix: always use with client.messages.stream(...) as stream: (Lesson 60).

Streaming Batch or Background Work

What happens: extra complexity with no user-facing benefit.

Fix: use regular requests for short background tasks, and the Batches API for bulk work.

Building the Final Text from Chunks for Storage

What happens: stored text is fine for plain answers, but thinking blocks, tool calls, and usage are lost.

Fix: store stream.get_final_message() (Lesson 55).

Best Practices

• Stream every user-facing response that may take more than a second or two.
• Use the messages.stream() helper with a with block.
• Show progress immediately, including a thinking indicator for thinking models.
• Use get_final_message() for storage, usage, and stop reasons.
• Use streaming for large max_tokens values, even when you do not display progress.
• Keep non-streaming requests for short background tasks.

Exercise

1. Run streaming_vs_blocking.py with a prompt that asks for about 50 words, then with one that asks for about 600 words. How does the difference in time to first text change?
2. Modify with_streaming() to also print the number of chunks received.
3. Use stream.get_final_message() inside with_streaming() to print the exact word count and usage.output_tokens.
4. Explain why first_text_at can be several seconds even with streaming, especially for complex questions.

Step 3: call get_final_message() after the loop, still inside the with block:

with client.messages.stream(**REQUEST) as stream:
    for text in stream.text_stream:
        ...
    message = stream.get_final_message()

final_text = "".join(b.text for b in message.content if b.type == "text")
print(f"exact words: {len(final_text.split())}, output tokens: {message.usage.output_tokens}")

Step 4: current models may think before writing the answer. Thinking tokens are generated first, and text_stream yields only answer text, so the first visible text appears only after the thinking phase. Complex questions usually involve more thinking. A visible "thinking…" indicator, or a lower effort setting for simple tasks, keeps the experience responsive.