MessageStream and AsyncMessageStream are iterable wrappers around raw SSE events that accumulate a final message and expose higher-level events (message_start, content_block_delta, etc.), text filters, and final-message accessors for synchronous and asynchronous streaming responses. The SDK provides paired context managers—MessageStreamManager (sync) and AsyncMessageStreamManager (async)—that defer the API request until entry, then yield a stream object for iterating events or draining text deltas with stream.text_stream. SSE (Server-Sent Events) is a protocol in which a server pushes a stream of newline-delimited text events over a single persistent HTTP connection, enabling real-time data delivery without repeated client polling. The Anthropic Python SDK wraps raw SSE events into higher-level ParsedMessageStreamEvent objects, so callers never need to parse the wire format directly.
client.messages.stream() returns a MessageStreamManager context manager that yields a MessageStream, which is iterable, emits events, and accumulates a final message object.[1] MessageStream (in src/anthropic/lib/streaming/_messages.py) is a synchronous, generic, context-manager-compatible iterator over ParsedMessageStreamEvent objects, wrapping a raw Stream[RawMessageStreamEvent].[2] AsyncMessageStream is the async counterpart to MessageStream, implementing __aiter__, __aenter__, and __aexit__ for use with async for and async with.[2]
MessageStreamManager is a synchronous context manager returned by .stream() that lazily invokes the API request on __enter__ and closes the stream on __exit__, deferring the actual HTTP call until the with block is entered.[2]
MessageStreamManager usage example — synchronous streaming context manager:
with client.messages.stream(...) as stream:
for chunk in stream:
...
AsyncMessageStreamManager is an async context manager wrapper returned by .stream() that does NOT require await-ing the original client call — the await is deferred to __aenter__.[2]
AsyncMessageStreamManager usage example — async streaming context manager:
async with client.messages.stream(...) as stream:
async for chunk in stream:
...
Inside MessageStream.__stream__, every raw SSE event is passed to accumulate_event() to build up __final_message_snapshot, and then build_events() maps the raw event plus snapshot into the higher-level ParsedMessageStreamEvent items that callers iterate over.[2] Iterating over a MessageStream (sync or async) yields ParsedMessageStreamEvent objects whose .type attribute follows the sequence message_start, content_block_start, interleaved content_block_delta/text pairs, content_block_stop, message_delta for a basic text response.[3] For a tool-use streaming response, ParsedMessageStreamEvent objects follow the sequence: message_start, text block events, content_block_stop, then a second block sequence with content_block_start, interleaved content_block_delta/input_json pairs, content_block_stop, message_delta.[3] anthropic.lib.streaming._messages exposes a TRACKS_TOOL_INPUT flag controlling whether tool input is tracked during streaming.[3]
MessageStream.text_stream is a synchronous Iterator[str] that yields only the text delta strings from content_block_delta events with delta.type == "text_delta", filtering out all other event types.[2]
MessageStream.text_stream usage example — iterate text-only deltas from a synchronous stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print()
AsyncMessageStream.text_stream is an AsyncIterator[str] that yields only text delta strings, mirroring the synchronous MessageStream.text_stream but for async usage.[2]
Canonical async streaming usage with client.messages.stream() and stream.text_stream:
async with client.messages.stream(
max_tokens=1024,
messages=[{"role": "user", "content": "Say hello there!"}],
model="claude-sonnet-5",
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
print()
MessageStream.get_final_message() calls until_done() to drain the stream before returning the accumulated ParsedMessage snapshot; it can also be called outside the context manager as long as the stream was fully consumed inside it.[2][1] MessageStream.get_final_text() blocks until the stream is fully consumed and returns all text-type content blocks concatenated together; the API currently returns only a single content block.[2][1] MessageStream.get_final_text() raises RuntimeError if the API response contains no text content blocks, with an error message listing the actual block types returned.[2] MessageStream.until_done() blocks until the stream has been read to completion without returning a value.[1] MessageStream.current_message_snapshot asserts that __final_message_snapshot is not None, so accessing it before the first SSE event has been processed will raise AssertionError.[2] MessageStream.request_id reads the request-id response header from the underlying httpx.Response, providing easy access to the API's request identifier for debugging.[2] MessageStream.close() delegates to self._raw_stream.close() and is automatically called when the response body is read to completion; the stream is also automatically cancelled when the context manager exits.[2][1]
Using a deprecated model with messages.stream() raises a DeprecationWarning matching "The model '{deprecated_model}' is deprecated"; the warning is triggered when the stream is consumed (e.g. via stream.until_done()).[3] The MessageStream object returned by messages.stream() is an instance of Stream (sync) or AsyncStream (async), but accessing the stream as such emits a DeprecationWarning.[3] When a refusal stop occurs, stream.get_final_message() returns a Message whose .stop_reason is "refusal", .stop_details.type is "refusal", and .stop_details carries category and explanation fields.[3] Tool-use streaming responses include cache_creation_input_tokens, cache_read_input_tokens, service_tier, and server_tool_use fields on message.usage in the final assembled message.[3]
Sources