How to Handle Streaming Responses (SSE) with the Anthropic Python SDK

The Anthropic Python SDK automatically parses Server-Sent Events (SSE) from the Claude API into convenient Python iterators via the Stream and AsyncStream classes located in anthropic/_streaming.py, enabling real-time processing of partial responses with minimal boilerplate.

The anthropics/anthropic-sdk-python repository provides native support for streaming responses through both the legacy Completions API and the newer Messages API. When you enable streaming, the SDK handles the low-level SSE protocol parsing, connection lifecycle management, and type conversion, yielding strongly-typed objects like Completion or Message chunks as the server generates them.

Understanding the Streaming Architecture

The Core SSE Decoder

At the heart of the streaming implementation lies the ServerSentEvent class and SSEDecoder in anthropic/_streaming.py. The ServerSentEvent wrapper (lines 58-73) captures the event, data, id, and retry fields from raw SSE lines. The SSEDecoder.decode() method (lines 55-100) consumes bytes from the httpx.Response.iter_bytes() iterator (or aiter_bytes() for async), reconstructing complete events even when they span multiple network chunks.

Synchronous vs Asynchronous Streams

The SDK exposes two primary streaming abstractions:

• Stream[T]: Returned for synchronous endpoints when stream=True is passed. Implements __iter__ and __next__ to yield objects of type T. See the constructor implementation in anthropic/_streaming.py (lines 51-63).
• AsyncStream[T]: Returned for asynchronous endpoints. Implements __aiter__ and __anext__ for async iteration, defined in the same file (lines 68-80).

Both classes maintain a reference to the underlying httpx.Response and automatically close the connection when the iterator is exhausted via a finally block in their respective __stream__ methods.

How to Stream Completions

Synchronous Streaming with stream=True

For synchronous scripts, pass stream=True to client.completions.create(). The SDK instantiates a Stream[Completion] object (referencing anthropic/resources/completions.py, lines 384-418) that yields partial completions as Completion model instances:

from anthropic import Anthropic, HUMAN_PROMPT, AI_PROMPT

client = Anthropic()
question = "How can I recursively list all files in a directory in Python?"

stream = client.completions.create(
    prompt=f"{HUMAN_PROMPT} {question}{AI_PROMPT}",
    model="claude-sonnet-4-5-20250929",
    stream=True,
    max_tokens_to_sample=300,
)

for chunk in stream:
    # Each chunk is a Completion model; generated text is in chunk.completion

    print(chunk.completion, end="", flush=True)

Each chunk object is a fully parsed Completion model. The SDK handles the JSON decoding and synthetic type field insertion (lines 85-99 in _streaming.py) to normalize event payloads before model instantiation.

Asynchronous Streaming with AsyncAnthropic

For async applications, the pattern is identical but uses AsyncAnthropic and await. The method returns an AsyncStream[Completion] (see anthropic/resources/completions.py, lines 784-809):

import asyncio
from anthropic import AsyncAnthropic, HUMAN_PROMPT, AI_PROMPT

async def main() -> None:
    client = AsyncAnthropic()
    question = "Explain the difference between sync and async streaming in the SDK."

    stream = await client.completions.create(
        prompt=f"{HUMAN_PROMPT} {question}{AI_PROMPT}",
        model="claude-sonnet-4-5-20250929",
        stream=True,
        max_tokens_to_sample=300,
    )

    async for chunk in stream:
        print(chunk.completion, end="", flush=True)

asyncio.run(main())

Streaming with the Messages API

Using the stream() Context Manager

The Messages API provides a higher-level streaming interface via client.messages.stream(), which returns a MessageStream (or AsyncMessageStream) context manager. This abstraction builds upon AsyncStream but adds convenience methods like get_final_message():

import asyncio
from anthropic import AsyncAnthropic

async def main() -> None:
    client = AsyncAnthropic()

    async with client.messages.stream(
        model="claude-sonnet-4-5-20250929",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Say hello there!"}],
    ) as stream:
        async for event in stream:
            if event.type == "text":
                print(event.text, end="", flush=True)
            elif event.type == "content_block_stop":
                print("\n--- content block finished ---")
                print(event.content_block)

        # Access the fully assembled message after iteration

        final_msg = await stream.get_final_message()
        print("\n\nAccumulated message JSON:", final_msg.to_json())

asyncio.run(main())

The MessageStream class ensures that SSE events without explicit "type" keys receive synthetic type fields derived from the SSE event name (e.g., "message_start" or "content_block_delta"), enabling uniform model construction.

Error Handling in Streaming Responses

If the API returns an event: error line, the SDK raises an APIStatusError (or subclass) constructed via client._make_status_error (defined in _client.py, lines 53-66). The iterator terminates immediately and closes the underlying HTTP connection:

from anthropic import Anthropic, APIStatusError

client = Anthropic()
try:
    stream = client.completions.create(
        prompt="Hello",
        model="non-existent-model",  # Intentionally invalid

        stream=True,
        max_tokens_to_sample=10,
    )
    for chunk in stream:
        print(chunk.completion)
except APIStatusError as exc:
    print("Streaming failed:", exc.response.text)

Summary

• Core Implementation: The SSE parsing logic resides in anthropic/_streaming.py, featuring ServerSentEvent, SSEDecoder, and the Stream/AsyncStream generic iterators.
• Sync Usage: Pass stream=True to resource methods like completions.create() to receive a Stream[Completion] iterable.
• Async Usage: Use AsyncAnthropic with stream=True to obtain an AsyncStream[Completion] for non-blocking iteration.
• Messages API: Prefer client.messages.stream() for structured streaming with built-in event type handling and get_final_message() accumulation.
• Connection Management: Both sync and async streams automatically close the httpx.Response when exhausted or when an error occurs.
• Error Safety: Streaming errors raise APIStatusError immediately upon receiving an error event, ensuring no resource leaks.

Frequently Asked Questions

How do I know when a streaming response is finished?

The iterator raises StopIteration (sync) or terminates the async generator when the SSE stream sends a data: [DONE] signal or closes the connection. For the Messages API, the async with context manager automatically handles cleanup, and you can check stream.get_final_message() after the loop to access the complete accumulated response.

What is the difference between stream=True and client.messages.stream()?

The stream=True parameter returns a raw Stream or AsyncStream iterator over model objects (e.g., Completion), suitable for the Completions API. The client.messages.stream() method returns a specialized MessageStream context manager with additional conveniences like automatic event type normalization and the get_final_message() helper for the Messages API.

Does the SDK handle SSE connection retries automatically?

No, the SDK does not implement automatic retry logic for streaming connections. If the connection drops, you must catch the exception and re-initiate the request. The retry field in the ServerSentEvent class (parsed in _streaming.py) is stored but not currently used to trigger automatic reconnection attempts.

Can I cancel a streaming request mid-execution?

Yes. Exiting the for loop or async for loop early triggers the iterator's cleanup logic, which closes the underlying httpx.Response in a finally block. For async streams, you can also cancel the task containing the loop; the context manager protocol ensures the connection closes properly even if the coroutine is cancelled.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}