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

> Effortlessly handle streaming responses (SSE) with the Anthropic Python SDK. Leverage Stream and AsyncStream for real-time Claude API data processing with minimal code.

- Repository: [Anthropic/anthropic-sdk-python](https://github.com/anthropics/anthropic-sdk-python)
- Tags: how-to-guide
- Published: 2026-02-23

---

**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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/anthropic/resources/completions.py), lines 384-418) that yields partial completions as `Completion` model instances:

```python
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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/_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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/anthropic/resources/completions.py), lines 784-809):

```python
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()`:

```python
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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/_client.py), lines 53-66). The iterator terminates immediately and closes the underlying HTTP connection:

```python
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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/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`](https://github.com/anthropics/anthropic-sdk-python/blob/main/_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.