Streaming Responses Basics

9 examples to get you started with Streaming Responses - 6 basic and 3 intermediate.

Prerequisites

• Install the SDK: pip install anthropic.
• Set your API key in the environment: export ANTHROPIC_API_KEY=sk-ant-....
• All examples use client = anthropic.Anthropic(), which reads the key from the environment automatically.

Basic Examples

1. Open a Stream and Print Text

The simplest possible streaming call: open a stream and print each text fragment as it arrives.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "Write a haiku about rivers."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

• client.messages.stream(...) returns a context manager, not a response object.
• stream.text_stream is a convenience iterator that yields only the text fragments, skipping other event types.
• flush=True matters here - without it, Python may buffer output and defeat the purpose of streaming.

Related: How Server-Sent Events Power Claude's Streaming Responses - what's happening under this loop.

2. Iterate Raw Events Instead of Text

Drop down from text_stream to the raw event iterator to see every event type as it arrives.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "Name three prime numbers."}],
) as stream:
    for event in stream:
        print(event.type)

• Iterating stream directly (instead of stream.text_stream) yields every SSE event: message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop.
• This is the level you need whenever you care about anything other than plain text - tool calls, thinking, or usage totals.
• stream.text_stream is built on top of this same iterator, filtering for text_delta events.

3. Get the Final Assembled Message

After streaming, retrieve the complete Message object exactly as a non-streaming call would return it.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "Summarize photosynthesis in one sentence."}],
) as stream:
    for _ in stream.text_stream:
        pass
    final_message = stream.get_final_message()
 
print(final_message.content[0].text)
print(final_message.usage)

• get_final_message() must be called after the stream has been fully consumed (inside or after the with block finishes iterating).
• It returns the same Message shape you'd get from client.messages.create(...) without streaming - useful when you want live rendering and the final object for logging.
• final_message.usage gives you the token counts that only become final once generation stops.

4. Check the Event Type Before Acting

Filter on event.type explicitly instead of relying on text_stream, which is the pattern you'll extend for tool use and thinking.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "What is the boiling point of water at sea level?"}],
) as 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_stop":
            print("\n[stream complete]")

• content_block_delta is a generic envelope - always check event.delta.type too, since it can be text_delta, input_json_delta, or thinking_delta.
• message_stop is a reliable place to run "stream finished" cleanup logic (closing a UI spinner, flushing a buffer).
• This explicit form is what you reach for once a response can contain more than plain text.

5. Stream with a System Prompt and Multiple Messages

Streaming works with the same request shape as a normal call - system prompts and multi-turn history included.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=300,
    system="You are a terse technical writer. Answer in two sentences or fewer.",
    messages=[
        {"role": "user", "content": "What is a content block?"},
        {"role": "assistant", "content": "It's one unit of a message's content, like text or a tool call."},
        {"role": "user", "content": "And what identifies which block a delta belongs to?"},
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

• system and messages are passed exactly as in a non-streaming client.messages.create(...) call - streaming changes delivery, not the request shape.
• Multi-turn history is just prior user/assistant entries in the messages list, same as always.
• Keep max_tokens reasonable for a streaming demo - a smaller cap means a shorter, easier-to-read live output.

6. Handle a Basic Streaming Error

Wrap the stream in a try/except so a mid-stream API error doesn't crash the whole process silently.

import anthropic
 
client = anthropic.Anthropic()
 
try:
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=200,
        messages=[{"role": "user", "content": "Explain rate limiting briefly."}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()
except anthropic.APIStatusError as e:
    print(f"\n[stream failed: {e.status_code} - {e.message}]")

• anthropic.APIStatusError (and its subclasses like RateLimitError) can surface at any point during iteration, not just when the request is first sent.
• Wrapping the whole with block, not just the initial call, is what catches errors raised mid-stream.
• This is the minimum safety net; see Streaming Best Practices for reconnect and retry strategies.

Intermediate Examples

7. Track Streaming Progress with a Callback-Style Loop

Combine text_stream with a running character count to drive a simple progress indicator.

import anthropic
 
client = anthropic.Anthropic()
 
def stream_with_progress(prompt: str) -> str:
    chars = 0
    chunks = []
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=500,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            chunks.append(text)
            chars += len(text)
            print(f"\r{chars} chars received", end="", flush=True)
    print()
    return "".join(chunks)
 
result = stream_with_progress("Describe the water cycle in a short paragraph.")
print(result)

• Accumulating chunks in a list and joining once at the end is cheaper than repeated string concatenation for longer responses.
• The \r carriage return lets the progress counter update in place in a terminal.
• This pattern - accumulate, report progress, return the final string - is the basis for wiring a stream into a UI callback instead of print.

8. Extract Usage Totals from message_delta

Read the token usage that only becomes available once the model finishes generating.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "List three benefits of streaming APIs."}],
) as 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":
            usage = event.usage
            print(f"\n[output tokens so far: {usage.output_tokens}]")

• message_delta events carry stop_reason and cumulative usage - fields that aren't known until generation is wrapping up.
• Input token counts arrive earlier (in message_start); output token counts finalize in message_delta.
• Logging usage here, rather than only after get_final_message(), lets you track cost in near-real time for long-running streams.

9. Stream Two Requests Concurrently with asyncio

Use the async client to run two independent streams at the same time instead of one after another.

import asyncio
import anthropic
 
async_client = anthropic.AsyncAnthropic()
 
async def stream_one(prompt: str, label: str) -> None:
    async with async_client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=200,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for text in stream.text_stream:
            print(f"[{label}] {text}", end="", flush=True)
 
async def main() -> None:
    await asyncio.gather(
        stream_one("Give a one-line fun fact about Mars.", "mars"),
        stream_one("Give a one-line fun fact about Venus.", "venus"),
    )
 
asyncio.run(main())

• anthropic.AsyncAnthropic() mirrors the sync client's API but every stream method is awaited with async with / async for.
• asyncio.gather runs both streams concurrently over separate connections, which is faster than sequential sync calls when you have multiple independent prompts.
• Interleaved output from two labeled streams is a preview of the buffering concerns a real multi-user chat backend has to solve.

Related: Choosing Between Anthropic and AsyncAnthropic in Production Code - when to reach for the async client.

Stack versions: Written against the Claude model lineup current as of ~June 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (the default), and Claude Haiku 4.5 - and the official anthropic Python SDK (latest 0.x release). Model names, SDK versions, and pricing move quickly - verify current specifics at platform.claude.com/docs before relying on them.