Streaming Responses with the Python SDK

Streaming lets your application render Claude's answer as it is generated, instead of waiting for the full response.

The anthropic package exposes this through client.messages.stream(), a context manager built specifically for consuming a response incrementally.

Summary

A non-streaming call to messages.create() returns one complete Message object after the model has finished generating.

For long responses, that means your application sits idle, showing nothing, until the very last token is ready.

messages.stream() returns a context manager instead, exposing events (and a convenient text_stream iterator) as they arrive from the server.

The final, fully assembled message is still available once the stream ends, so streaming adds a way to observe the response incrementally without losing anything you'd get from a normal call.

Streaming is also the recommended way to request large outputs, because non-streaming requests risk hitting client-side HTTP timeouts once max_tokens climbs into the tens of thousands.

Recipe

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Write a short poem about the ocean."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
 
    final_message = stream.get_final_message()
 
print()
print(f"\nStop reason: {final_message.stop_reason}")

When to reach for this:

• Chat interfaces where you want to show text as it's generated, matching how users expect a conversational UI to feel.
• Any request with max_tokens above roughly 16,000, where a non-streaming call risks a client-side timeout.
• Long-form generation (articles, reports, large code files) where partial output is still useful to the user before the full response completes.
• CLI tools that print output progressively rather than pausing silently.

Working Example

import anthropic
 
client = anthropic.Anthropic()
 
 
def stream_answer(question: str) -> str:
    """Stream a response to stdout and return the full text once done."""
    full_text = ""
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=4096,
        system="You are a concise technical writer.",
        messages=[{"role": "user", "content": question}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            full_text += text
 
        final_message = stream.get_final_message()
 
    print()  # newline after the streamed output
    print(f"[stop_reason={final_message.stop_reason}, "
          f"output_tokens={final_message.usage.output_tokens}]")
 
    return full_text
 
 
if __name__ == "__main__":
    stream_answer("Explain what a context window is, in three sentences.")

What this demonstrates:

• client.messages.stream(...) is used as a context manager (with ... as stream:), which handles opening and closing the underlying connection.
• stream.text_stream yields plain text chunks as they arrive - no manual event parsing required for the common case.
• stream.get_final_message() (called after the for loop finishes) returns the same complete Message object a non-streaming call would have returned, including stop_reason and usage.
• Accumulating chunks into full_text lets you both print progressively and still return the complete string to the caller.

Deep Dive

How It Works

• Under the hood, messages.stream() opens an HTTP connection and reads a series of server-sent events (message_start, content_block_delta, message_delta, message_stop, and others) as they arrive.
• text_stream is a convenience iterator built on top of those raw events - it filters for text deltas and yields just the string content, so you don't parse event types yourself for basic use.
• The stream object buffers everything it has seen, which is what lets get_final_message() return a complete, typed Message after the loop ends, with all the same fields (content, stop_reason, usage) as a non-streaming response.
• Token usage and billing are unaffected by streaming - you pay for the same output tokens whether you receive them as one block or as many small chunks.
• If the connection drops mid-stream, the exception surfaces from the iteration itself (inside the with block), so wrap streaming calls in the same retryable-error handling you'd use for any other API call.

Event Types You Can Access Directly

For anything beyond plain text, iterate the stream object itself instead of text_stream:

Python Notes

# Inspecting raw events instead of using text_stream - useful when you
# need thinking content or tool_use blocks as they stream, not just text.
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Explain recursion."}],
) 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":
            # carries incremental usage and stop_reason updates
            pass

Async Streaming

import asyncio
import anthropic
 
client = anthropic.AsyncAnthropic()
 
 
async def stream_answer(question: str) -> None:
    async with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": question}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
 
    print()
 
asyncio.run(stream_answer("What is a coroutine, in two sentences?"))

AsyncAnthropic() streaming uses async with and async for in place of with and for - everything else about the API is the same.

Gotchas

• Using with instead of async with on the async client. AsyncAnthropic().messages.stream(...) returns an async context manager; opening it with plain with raises a TypeError. Fix: use async with and async for whenever you're on AsyncAnthropic().
• Forgetting flush=True when printing streamed text. Without it, Python may buffer stdout and the output appears to arrive in bursts rather than smoothly. Fix: pass flush=True to print(), or write directly to sys.stdout and flush explicitly.
• Reading stream.text_stream twice. The iterator is consumed once; iterating it again after the loop finishes yields nothing. Fix: accumulate the text into a variable during the first pass if you need it again afterward.
• Calling get_final_message() before the stream has been fully consumed. If you break out of the for loop early, the final message may be incomplete or the call may block waiting for the rest of the stream. Fix: let the iteration finish naturally, or explicitly close the stream first if you intend to abandon it early.
• Assuming streaming avoids the max_tokens truncation problem. Streaming changes how you receive output, not how much of it the model is allowed to generate; stop_reason == "max_tokens" can still happen. Fix: size max_tokens for the task regardless of whether you stream.
• Not handling network errors during iteration. An interrupted connection surfaces as an exception raised from inside the for text in stream.text_stream: loop, not before it starts. Fix: wrap the with block in the same typed-exception handling you use for non-streaming calls.

Alternatives

FAQs

Related

• The Anthropic Python SDK Mental Model - how streaming fits with sync/async clients
• Python SDK Basics - your first non-streaming request, for comparison
• Handling Streaming Tool Use in Async Python Workflows - streaming combined with tool_use
• Python SDK Retry and Timeout Configuration Reference - timeout behavior during long streams

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.