Use Stream Helpers

What the Stream Helpers Are

Lesson 53 processed raw events by hand: tracking block indexes, accumulating text, and joining partial JSON. The SDK's stream helpers do this work for you.

When you call client.messages.stream(...), the SDK returns a MessageStream object that:

• accumulates every event into a complete message as the stream progresses,
• offers convenient iterators and methods for common tasks, and
• emits extra, higher-level events, such as "text arrived" or "this block is complete", alongside the raw events.

with client.messages.stream(model="claude-opus-5-5", max_tokens=1024, messages=messages) as stream:
    ...  # stream is a MessageStream

This lesson covers each helper, when to use it, and how the helpers fit together.

Helper Overview

Helper Events

When you iterate over a MessageStream, you receive the raw events from Lesson 53 and additional events created by the SDK. The most useful helper events:

The content_block_stop and message_stop events here are richer versions of the raw events: they carry the fully accumulated block or message, so you do not need to build them yourself.

Note: the exact set of helper events can grow as the API adds features. Handle the events you need, and ignore the rest.

Example: Text with Snapshots

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a haiku about debugging."}],
) as stream:
    for event in stream:
        if event.type == "text":
            print(event.text, end="", flush=True)
        elif event.type == "content_block_stop" and event.content_block.type == "text":
            print(f"\n[block complete: {len(event.content_block.text)} characters]")

• event.text is the new chunk, the same as text_stream yields.
• event.content_block at content_block_stop is the complete TextBlock, so the full text of that block is available without manual accumulation.

Previewing Tool Inputs as They Stream

Tool inputs arrive as fragments of JSON (Lesson 53). The input_json helper event provides a snapshot: the input parsed as far as possible so far, as a Python object. That makes it easy to show a live preview of what a tool is about to receive:

for event in stream:
    if event.type == "input_json":
        print(f"\rTool input so far: {event.snapshot}", end="", flush=True)
    elif event.type == "content_block_stop" and event.content_block.type == "tool_use":
        block = event.content_block
        print(f"\nComplete tool call: {block.name}({block.input})")

• \r returns the cursor to the start of the line, so each update overwrites the previous preview instead of printing a new line.
• event.snapshot may be incomplete while streaming, for example a string that is still being written. Only act on the input at content_block_stop, where block.input is the complete, parsed dictionary.

Checking Progress Mid-Stream

current_message_snapshot returns the message as it has been accumulated up to now: its content blocks, and whatever usage is known so far:

for text in stream.text_stream:
    print(text, end="", flush=True)

snapshot = stream.current_message_snapshot
print(f"\nBlocks so far: {[block.type for block in snapshot.content]}")

This is useful for checks during a stream, for example stopping a response that has already produced enough text (Lesson 47), or saving partial progress periodically.

Final Results

After the stream completes, two methods return the accumulated result:

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)

    final_text = stream.get_final_text()        # Just the answer text
    final_message = stream.get_final_message()  # The complete Message

Both methods wait for the stream to finish if it has not already, so they can also be called without iterating first, as Lesson 55 shows.

Choosing the Right Helper

Practical Example: A Helper-Based Live Assistant

This example uses helper events to display a response with three kinds of live feedback: thinking status, streaming answer text, and a live tool-input preview. It finishes with the complete message from get_final_message().

"""stream_helpers.py: live feedback using MessageStream helper events."""

from anthropic import Anthropic

client = Anthropic()

SEARCH_TOOL = {
    "name": "search_courses",
    "description": "Search the course catalog by topic and level.",
    "input_schema": {
        "type": "object",
        "properties": {
            "topic": {"type": "string", "description": "Course topic, e.g. 'Django'"},
            "level": {"type": "string", "enum": ["beginner", "intermediate", "advanced"]},
        },
        "required": ["topic", "level"],
    },
}


def run(question: str) -> None:
    with client.messages.stream(
        model="claude-opus-5-5",
        max_tokens=2048,
        tools=[SEARCH_TOOL],
        messages=[{"role": "user", "content": question}],
    ) as stream:
        for event in stream:
            if event.type == "content_block_start" and event.content_block.type == "thinking":
                print("(thinking...)")
            elif event.type == "text":
                print(event.text, end="", flush=True)
            elif event.type == "input_json":
                print(f"\r  preparing tool input: {event.snapshot}", end="", flush=True)
            elif event.type == "content_block_stop":
                block = event.content_block
                if block.type == "text":
                    print()
                elif block.type == "tool_use":
                    print(f"\r  tool call ready: {block.name}({block.input})" + " " * 20)

        final = stream.get_final_message()

    tool_calls = [b for b in final.content if b.type == "tool_use"]
    print(f"\nstop_reason={final.stop_reason} | blocks={[b.type for b in final.content]} "
          f"| tool calls={len(tool_calls)} | output_tokens={final.usage.output_tokens}")


if __name__ == "__main__":
    run("I'm a beginner who wants to learn Django. Find me a suitable course.")

Example output:

(thinking...)
I'll search the catalog for beginner Django courses.
  tool call ready: search_courses({'topic': 'Django', 'level': 'beginner'})

stop_reason=tool_use | blocks=['thinking', 'text', 'tool_use'] | tool calls=1 | output_tokens=131

How It Works

1. The raw content_block_start event still arrives when iterating the helper. It is used here to detect a thinking block early and show "(thinking...)".
2. text helper events print the answer chunks.
3. input_json helper events update a single preview line using \r. The snapshot grows from {} to {'topic': 'Django'} to the full input.
4. content_block_stop helper events carry the complete block. For the tool call, block.input is the final, parsed dictionary, which is safe to pass to a real function. The extra spaces overwrite any leftover characters from the longer preview line.
5. get_final_message(), called inside the with block after the loop, returns the complete Message, the same structure create() returns, with all blocks, the stop reason, and usage.

No manual index tracking or JSON accumulation was needed, unlike Lesson 53. That is the value of the helpers.

Common Mistakes

Acting on Partial Tool Input

What happens: a tool runs with incomplete arguments taken from an input_json snapshot.

Fix: use snapshots only for display. Act on content_block.input at content_block_stop, or on the final message.

Re-implementing Accumulation

What happens: extra code that duplicates what the helper already does, with more room for bugs.

Fix: use snapshot, content_block_stop, current_message_snapshot, and get_final_message().

Mixing text_stream and Event Iteration on the Same Stream

What happens: both consume the same underlying events, so text appears in only one of them, or is missed.

Fix: choose one way to consume a stream: either text_stream or iterating events.

Calling Final Methods Outside the with Block

What happens: the stream is closed before the final result is retrieved.

Fix: call get_final_message() or get_final_text() inside the with block.

Best Practices

• Prefer the helpers to raw events for most applications.
• Use text_stream for simple display, and helper events for richer feedback.
• Use snapshots for live display, and complete blocks for actions.
• Consume each stream in one way.
• Retrieve final results inside the with block.
• Ignore helper events you do not need.

Exercise

1. Modify the haiku example to print the snapshot length after each text event, instead of the chunk itself.
2. Use current_message_snapshot to stop a response once it contains more than 300 characters of text.
3. In stream_helpers.py, add handling for message_stop events that prints the total number of content blocks from event.message.
4. Explain why the tool should be run with block.input from content_block_stop, not with the last input_json snapshot.

Step 2:

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "Write a long essay about the history of Python."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
        snapshot = stream.current_message_snapshot
        written = "".join(b.text for b in snapshot.content if b.type == "text")
        if len(written) > 300:
            print("\n[stopped early]")
            break

Breaking out of the loop and leaving the with block closes the stream (Lesson 60).

Step 4: the last snapshot before the block ends may still be missing the final characters or fields. Only at content_block_stop is the input JSON complete and fully parsed. Running a tool with partial input could pass wrong or missing arguments, for example a truncated city name, to a real system.