Stream Text Output

The Simplest Streaming Pattern

The most common streaming task is showing Claude's answer text as it arrives. The stream helper provides text_stream, an iterator that yields only the answer text, chunk by chunk:

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a four-line poem about Python."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()

Line by line:

What text_stream Contains

Chunks have no fixed size. A chunk can be a word, part of a word, several words, or just whitespace or a line break. Your code must never assume that a chunk is a complete word, sentence, or line.

Because text_stream contains only answer text, it is the right choice for displaying replies. For thinking, tool calls, or progress events, use the event interface (Lesson 53).

Why end="" and flush=True Matter

print(text, end="", flush=True)

• end="": print() normally adds a newline after its output. Chunks are fragments of continuous text, so a newline after each chunk would break the text apart.
• flush=True: Python buffers output, collecting it and writing it in larger batches for efficiency. Without flushing, the text may appear in sudden bursts, or only at the end, which defeats the purpose of streaming.

Collecting the Full Text While Streaming

Often you need both: display the text live and keep the complete text for later use. Collect the chunks in a list, then join them:

chunks = []
with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Give three tips for writing clean Python."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
        chunks.append(text)

full_text = "".join(chunks)
print(f"\n\n[{len(full_text)} characters received]")

Why a list instead of full_text += text: appending to a list and joining once is efficient for any length. Repeated string concatenation creates a new string each time, which becomes slow for very long outputs.

Getting the Final Text Directly

The stream helper also accumulates the text for you. After the stream finishes, get_final_text() returns the complete answer text:

with client.messages.stream(model="claude-opus-5-5", max_tokens=1024, messages=messages) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    full_text = stream.get_final_text()

To also get the stop reason, usage, and all content blocks, use get_final_message() (Lesson 55).

Writing Streamed Text to a File

To save output progressively, for example for long reports, write each chunk as it arrives:

from pathlib import Path

with Path("report.md").open("w", encoding="utf-8") as file, client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=8192,
    messages=[{"role": "user", "content": "Write a detailed guide to Python packaging."}],
) as stream:
    for text in stream.text_stream:
        file.write(text)
        file.flush()

• A single with statement manages both the file and the stream, so both are closed properly.
• file.flush() writes each chunk to disk immediately, so the partial output is preserved if the program stops unexpectedly.
• encoding="utf-8" handles any characters in the output (Lesson 15).

Rendering Streamed Markdown

Claude often formats answers in Markdown. Printing raw chunks shows symbols such as ** and #. The rich library can re-render Markdown live as text arrives:

python -m pip install rich

from rich.live import Live
from rich.markdown import Markdown

buffer = ""
with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Explain Python decorators with a short code example."}],
) as stream, Live(Markdown(""), refresh_per_second=8) as live:
    for text in stream.text_stream:
        buffer += text
        live.update(Markdown(buffer))

How it works: Live controls a region of the terminal that can be redrawn. Each chunk is added to buffer, and the whole buffer is re-rendered as Markdown. Incomplete Markdown, such as an unclosed code block, renders as plain text until it is completed. refresh_per_second=8 limits how often the screen is redrawn, which keeps the terminal smooth.

Re-rendering the whole buffer each time is fine for typical chat-length answers. For very long documents, redraw less often or print plain text.

Practical Example: A Streaming Terminal Assistant with Statistics

This tool streams an answer with live display and then reports useful statistics: time to first text, total time, output tokens, and the stop reason.

"""stream_cli.py: stream an answer to the terminal and report timing statistics.

Usage:
    python stream_cli.py "Explain Python list comprehensions with examples."
"""

import sys
import time

from anthropic import Anthropic

client = Anthropic()


def stream_answer(question: str) -> None:
    start = time.perf_counter()
    first_text_at = None

    with client.messages.stream(
        model="claude-opus-5-5",
        max_tokens=4096,
        messages=[{"role": "user", "content": question}],
    ) as stream:
        print("Claude: ", end="", flush=True)
        for text in stream.text_stream:
            if first_text_at is None and text.strip():
                first_text_at = time.perf_counter() - start
            print(text, end="", flush=True)
        message = stream.get_final_message()

    total = time.perf_counter() - start
    output_tokens = message.usage.output_tokens

    print("\n")
    print(f"[time to first text: {first_text_at:.1f}s]" if first_text_at else "[no text received]")
    print(f"[total time: {total:.1f}s | output tokens: {output_tokens} "
          f"| ~{output_tokens / total:.0f} tokens/s | stop: {message.stop_reason}]")
    if message.stop_reason == "max_tokens":
        print("[warning: the answer was cut off at max_tokens]")


if __name__ == "__main__":
    question = " ".join(sys.argv[1:]) or "Explain Python list comprehensions with two examples."
    stream_answer(question)

Example output:

Claude: A list comprehension builds a new list in a single expression...
...

[time to first text: 1.8s]
[total time: 7.4s | output tokens: 612 | ~83 tokens/s | stop: end_turn]

How It Works

1. print("Claude: ", end="", flush=True) shows a label immediately, before any text arrives, so the user knows a response is coming.
2. first_text_at records when the first non-whitespace chunk arrived, the time to first text from Lesson 51.
3. stream.get_final_message() is called inside the with block, after the loop. It returns the complete Message, including usage and stop_reason, which the text chunks alone do not provide.
4. Tokens per second is calculated from output_tokens, which include thinking, divided by the total time. It is an approximate throughput measure, useful for comparing models and settings.
5. The truncation warning follows the rule from Lesson 20: never present a cut-off answer as complete.
6. The conditional expression ... if first_text_at else ... handles responses with no text, for example if thinking used the whole max_tokens budget.

Common Mistakes

Printing Each Chunk on Its Own Line

What happens: print(text) adds a newline after every chunk, breaking words and sentences apart.

Fix: use end="".

Forgetting to Flush

What happens: text appears in bursts instead of smoothly.

Fix: use flush=True, or file.flush() when writing to files.

Assuming Chunks Are Words or Lines

What happens: logic such as "capitalize each chunk" or "process each line as it arrives" produces broken results.

Fix: accumulate text, and process complete units, such as lines split from the accumulated buffer.

Expecting Thinking or Tool Calls in text_stream

What happens: tool calls and thinking are silently missing.

Fix: use the event interface (Lesson 53) or the final message (Lesson 55).

Calling get_final_message() After the with Block

What happens: the stream is already closed, and the final message may not be available.

Fix: call it inside the with block, after consuming the text.

Best Practices

• Use text_stream for displaying answers.
• Print with end="" and flush=True.
• Collect chunks in a list and join once, or use get_final_text().
• Get the final message for stop reasons, usage, and storage.
• Show a label or indicator immediately, before the first text arrives.
• Render Markdown live only when the output is meant for human reading in a capable terminal.
• Warn about truncation after the stream ends.

Exercise

1. Run stream_cli.py with a short question and a long one. Compare time to first text and tokens per second.
2. Modify the file-writing example to print a dot (.) to the terminal for every chunk written, as a simple progress indicator.
3. Process streamed text line by line: accumulate chunks in a buffer, and print each complete line with a line number as soon as it is finished.
4. Combine rich Markdown rendering with the statistics from stream_cli.py.

buffer = ""
line_number = 0

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "List 8 Python built-in functions, one per line, with a short description."}],
) as stream:
    for text in stream.text_stream:
        buffer += text
        while "\n" in buffer:
            line, buffer = buffer.split("\n", 1)
            line_number += 1
            print(f"{line_number:>2}: {line}")

if buffer.strip():
    line_number += 1
    print(f"{line_number:>2}: {buffer}")

How it works: chunks are added to buffer. Whenever the buffer contains a line break, the text before it is a complete line, so it is printed and removed from the buffer with split("\n", 1). The while loop handles chunks that contain several line breaks. After the stream ends, any remaining text, the last line without a trailing newline, is printed.