Use Stream Events

Beyond Text: The Event Stream

text_stream (Lesson 52) gives you only the answer text. A stream actually carries much more: when content blocks begin and end, thinking progress, tool call inputs, the stop reason, and token usage. All of this arrives as a sequence of events.

Working with events lets you:

• show status indicators, such as "Thinking…" or "Calling get_weather…",
• display tool inputs as they are generated,
• read the stop reason and usage as soon as they are known, and
• build integrations that forward events to other systems, such as a web browser.

The Event Sequence

According to the streaming documentation, every stream follows this flow:

message_start
│
├── content_block_start   (index 0)
│   ├── content_block_delta  (index 0)   ← repeated
│   └── content_block_stop   (index 0)
│
├── content_block_start   (index 1)
│   ├── content_block_delta  (index 1)   ← repeated
│   └── content_block_stop   (index 1)
│   ... more blocks ...
│
├── message_delta         ← stop_reason and cumulative usage
└── message_stop

Each content block has an index that matches its position in the final message's content list. Blocks are streamed one after another, in order.

Event Types and Their Data

Delta Types

Thinking in Streams

By default, thinking text is not shown. According to the documentation, a thinking block then opens, receives a thinking_delta with an empty thinking string and a single signature_delta, and closes. With summarized thinking (thinking={"type": "adaptive", "display": "summarized"}), thinking_delta events carry summary text.

Either way, the content_block_start of a thinking block tells you that Claude is thinking. That is ideal for showing a "Thinking…" indicator.

Tool Input Arrives as Partial JSON

For tool_use blocks, input_json_delta events carry fragments of a JSON string, such as '{"location": "San Fra'. The fragments only form valid JSON once the block is complete, so accumulate them and parse at content_block_stop. The SDK's helpers do this for you (Lesson 54).

Other Events

• ping events can appear anywhere in the raw stream. In the Python SDK, they are handled for you.
• Error events, such as overloaded_error during periods of high usage, can arrive mid-stream. The SDK raises them as exceptions (Lesson 57).
• New event types may be added over time. Your code should ignore event types it does not recognize, instead of failing.

Iterating Over Events

Use create(..., stream=True) to receive the raw events directly:

from anthropic import Anthropic

client = Anthropic()

stream = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Say hello in three languages."}],
    stream=True,
)

with stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            print(f"\n[stop_reason={event.delta.stop_reason}, "
                  f"output_tokens={event.usage.output_tokens}]")

How it works:

• stream=True makes create() return an iterator of events instead of a Message.
• with stream: closes the connection when the block ends, even if an error occurs (Lesson 60).
• Text is taken from content_block_delta events whose delta is a text_delta.
• message_delta provides the stop reason and the cumulative output token count.

Note: you can also iterate over the stream helper: with client.messages.stream(...) as stream: for event in stream:. The helper yields these raw events plus extra convenience events, such as text events, covered in Lesson 54.

Tracking Blocks by Index

Because deltas refer to blocks by index, you can rebuild each block's content as it streams:

blocks: dict[int, dict] = {}

for event in stream:
    if event.type == "content_block_start":
        blocks[event.index] = {"type": event.content_block.type, "text": "", "json": ""}
    elif event.type == "content_block_delta":
        block = blocks[event.index]
        if event.delta.type == "text_delta":
            block["text"] += event.delta.text
        elif event.delta.type == "input_json_delta":
            block["json"] += event.delta.partial_json

This is what the SDK's accumulation helpers do internally. In practice, use get_final_message() (Lesson 55) for the complete result, and use events for live reactions.

Practical Example: A Live Status Display

This script streams a request that can use a tool. It shows live status lines: "Thinking…" when a thinking block starts, the answer text as it arrives, and tool calls with their inputs once each is complete. At the end, it prints the stop reason and usage from the message_delta event.

"""stream_events.py: react to stream events with live status updates."""

import json
import time

from anthropic import Anthropic

client = Anthropic()

WEATHER_TOOL = {
    "name": "get_weather",
    "description": "Get the current weather for a city.",
    "input_schema": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}


def run(question: str) -> None:
    start = time.perf_counter()
    block_types: dict[int, str] = {}
    tool_json: dict[int, str] = {}
    tool_names: dict[int, str] = {}

    stream = client.messages.create(
        model="claude-opus-5-5",
        max_tokens=2048,
        tools=[WEATHER_TOOL],
        messages=[{"role": "user", "content": question}],
        stream=True,
    )

    with stream:
        for event in stream:
            elapsed = f"{time.perf_counter() - start:5.1f}s"

            if event.type == "message_start":
                print(f"[{elapsed}] started (input tokens: {event.message.usage.input_tokens})")

            elif event.type == "content_block_start":
                kind = event.content_block.type
                block_types[event.index] = kind
                if kind == "thinking":
                    print(f"[{elapsed}] Thinking...")
                elif kind == "tool_use":
                    tool_names[event.index] = event.content_block.name
                    tool_json[event.index] = ""
                elif kind == "text":
                    print(f"[{elapsed}] Answer: ", end="", flush=True)

            elif event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    print(event.delta.text, end="", flush=True)
                elif event.delta.type == "input_json_delta":
                    tool_json[event.index] += event.delta.partial_json

            elif event.type == "content_block_stop":
                kind = block_types.get(event.index)
                if kind == "text":
                    print()
                elif kind == "tool_use":
                    arguments = json.loads(tool_json[event.index] or "{}")
                    print(f"[{elapsed}] Tool call: {tool_names[event.index]}({arguments})")

            elif event.type == "message_delta":
                print(f"[{elapsed}] stop_reason={event.delta.stop_reason}, "
                      f"output_tokens={event.usage.output_tokens}")

            elif event.type == "message_stop":
                print(f"[{elapsed}] done")


if __name__ == "__main__":
    run("Is it a good day for a walk in Chennai? Check the weather first.")

Example output (timings and wording vary):

[  0.6s] started (input tokens: 410)
[  0.6s] Thinking...
[  2.3s] Answer: I'll check the current weather in Chennai first.
[  3.1s] Tool call: get_weather({'city': 'Chennai'})
[  3.1s] stop_reason=tool_use, output_tokens=118
[  3.1s] done

How It Works

1. block_types records the type of each block by its index at content_block_start, so later content_block_stop events, which carry only an index, know which kind of block ended.
2. Thinking blocks trigger a "Thinking…" status line. The user sees immediately that work is in progress, even before any answer text exists.
3. Text deltas are printed live, exactly as with text_stream.
4. Tool input fragments (partial_json) are accumulated per block and parsed with json.loads only at content_block_stop, when the JSON is complete. or "{}" handles tools that were called with no input.
5. message_delta provides the stop reason (tool_use here, because Claude wants to run the tool) and the cumulative output token count.
6. Unhandled event types are simply skipped, because there is no final else that raises an error. That is how the code stays compatible with event types added in the future.

The stream ends with stop_reason=tool_use. To finish the task, your application would run the tool and send the result back (Lessons 85–88).

Common Mistakes

Parsing Partial JSON Too Early

What happens: json.loads() on an incomplete fragment raises JSONDecodeError.

Fix: accumulate partial_json and parse at content_block_stop, or use the SDK's helpers.

Ignoring the Block Index

What happens: text from different blocks, or tool inputs from several tool calls, gets mixed together.

Fix: track content per index.

Treating Usage in message_delta as Incremental

What happens: adding up usage values double-counts tokens.

Fix: the counts in message_delta are cumulative. Use the latest value.

Failing on Unknown Event Types

What happens: code with an else: raise ... branch breaks when the API adds a new event type.

Fix: ignore event types you do not handle.

Not Closing Raw Streams

What happens: connections stay open when your code stops iterating early or an error occurs.

Fix: use with stream: around the iteration.

Best Practices

• Use text_stream when you only need answer text, and events when you need status, tools, or metadata.
• Track blocks by index and record their types at content_block_start.
• Show status for thinking and tool calls, so users see progress before the answer starts.
• Accumulate tool input JSON and parse it when the block stops.
• Read stop reasons and usage from message_delta, remembering that usage is cumulative.
• Ignore unknown events for forward compatibility.
• Close streams with with.

Exercise

1. Run stream_events.py with a question that does not need the tool, such as "What is 12 × 12?". Which events appear?
2. Add thinking={"type": "adaptive", "display": "summarized"} to the request, and print thinking_delta text in a dimmer style, or with a [thinking] prefix.
3. Count how many content_block_delta events arrive for each block type, and print the counts at the end.
4. Record the time of the first text_delta and print it as the "time to first text".

Step 2: add the parameter to create(), then handle the delta type:

elif event.type == "content_block_delta":
    if event.delta.type == "thinking_delta" and event.delta.thinking:
        print(f"[thinking] {event.delta.thinking}", end="", flush=True)
    elif event.delta.type == "text_delta":
        print(event.delta.text, end="", flush=True)

The event.delta.thinking check skips empty thinking deltas.

Step 3:

from collections import Counter

delta_counts: Counter[str] = Counter()

# Inside the loop:
elif event.type == "content_block_delta":
    delta_counts[block_types[event.index]] += 1
    ...

# After the loop:
print(dict(delta_counts))   # e.g. {'thinking': 2, 'text': 14, 'tool_use': 6}