Streaming Helpers, Final Messages, and Usage

Two ways to stream in Python

The SDK gives you two streaming entry points. client.messages.stream(...) is a context-manager-style helper with rich convenience methods (text_stream, get_final_message, on_event). client.messages.create(stream=True) is the lower-level path that returns the raw event iterator. Use the helper for chat UI; use the iterator when you want full control over event handling.

get_final_message gives you usage

While you are streaming text, you do not have token usage yet. Usage is finalized at message_stop. The helper exposes it via stream.get_final_message() after iteration completes; the iterator path requires you to capture message_delta events and accumulate.

Persisting the stream as you go

For applications that log or replay sessions (cwkPippa), write each text delta to durable storage before rendering it to the user. JSONL append-only with explicit flush after every line means a hard kill mid-stream still leaves a complete record of what the user actually saw.

Principle: Stream for UX, finalize for accounting. Treat usage as the closing accounting line, not a per-delta number.

Code

Helper-style streaming with usage·python

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Three tips for streaming UX."}],
) as stream:
    for chunk in stream.text_stream:
        print(chunk, end="", flush=True)
    final = stream.get_final_message()

print()
print("input:", final.usage.input_tokens, "output:", final.usage.output_tokens)

Low-level iterator with JSONL persistence·python

import json, pathlib
from anthropic import Anthropic

client = Anthropic()
session_log = pathlib.Path("/tmp/session.jsonl").open("a", buffering=1)  # line-buffered

stream = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Stream and persist"}],
    stream=True,
)

for event in stream:
    if event.type == "content_block_delta" and event.delta.type == "text_delta":
        session_log.write(json.dumps({"delta": event.delta.text}) + "\n")
        session_log.flush()
        print(event.delta.text, end="", flush=True)
    elif event.type == "message_stop":
        # Final usage is on the message; capture from the assembled response if needed
        print()

session_log.close()

Exercise

Rewrite a streaming chat handler so each text delta is written to a JSONL file before it is yielded to the HTTP stream. Crash the process mid-response with SIGTERM and verify the JSONL still contains every visible token.

Hint

Open the file with buffering=1 (line-buffered) and call flush() after each write.

Streaming Helpers, Final Messages, and Usage

Two ways to stream in Python

get_final_message gives you usage

Persisting the stream as you go

Code

External links

Exercise

Progress

Comments 0