1# Streaming — Python
2 
3## Quick Start
4 
5```python
6with client.messages.stream(
7    model="claude-opus-4-8",
8    max_tokens=64000,
9    messages=[{"role": "user", "content": "Write a story"}]
10) as stream:
11    for text in stream.text_stream:
12        print(text, end="", flush=True)
13```
14 
15### Async
16 
17```python
18async with async_client.messages.stream(
19    model="claude-opus-4-8",
20    max_tokens=64000,
21    messages=[{"role": "user", "content": "Write a story"}]
22) as stream:
23    async for text in stream.text_stream:
24        print(text, end="", flush=True)
25```
26 
27### Low-level: `stream=True`
28 
29`messages.stream()` (above) is the recommended helper — it accumulates state and exposes `text_stream` / `get_final_message()`. If you only need the raw event iterator and want lower memory use, pass `stream=True` to `messages.create()` instead:
30 
31```python
32for event in client.messages.create(
33    model="claude-opus-4-8",
34    max_tokens=64000,
35    messages=[{"role": "user", "content": "Write a story"}],
36    stream=True,
37):
38    print(event.type)
39```
40 
41No final-message accumulation is done for you in this form.
42 
43---
44 
45## Handling Different Content Types
46 
47Claude may return text, thinking blocks, or tool use. Handle each appropriately:
48 
49> **Fable 5 / Opus 4.8 / Opus 4.7 / Opus 4.6:** Use `thinking: {type: "adaptive"}`. On older models, use `thinking: {type: "enabled", budget_tokens: N}` instead.
50 
51```python
52with client.messages.stream(
53    model="claude-opus-4-8",
54    max_tokens=64000,
55    thinking={"type": "adaptive"},
56    messages=[{"role": "user", "content": "Analyze this problem"}]
57) as stream:
58    for event in stream:
59        if event.type == "content_block_start":
60            if event.content_block.type == "thinking":
61                print("\n[Thinking...]")
62            elif event.content_block.type == "text":
63                print("\n[Response:]")
64 
65        elif event.type == "content_block_delta":
66            if event.delta.type == "thinking_delta":
67                print(event.delta.thinking, end="", flush=True)
68            elif event.delta.type == "text_delta":
69                print(event.delta.text, end="", flush=True)
70```
71 
72---
73 
74## Streaming with Tool Use
75 
76The Python tool runner currently returns complete messages. Use streaming for individual API calls within a manual loop if you need per-token streaming with tools:
77 
78```python
79with client.messages.stream(
80    model="claude-opus-4-8",
81    max_tokens=64000,
82    tools=tools,
83    messages=messages
84) as stream:
85    for text in stream.text_stream:
86        print(text, end="", flush=True)
87 
88    response = stream.get_final_message()
89    # Continue with tool execution if response.stop_reason == "tool_use"
90```
91 
92---
93 
94## Getting the Final Message
95 
96```python
97with client.messages.stream(
98    model="claude-opus-4-8",
99    max_tokens=64000,
100    messages=[{"role": "user", "content": "Hello"}]
101) as stream:
102    for text in stream.text_stream:
103        print(text, end="", flush=True)
104 
105    # Get full message after streaming
106    final_message = stream.get_final_message()
107    print(f"\n\nTokens used: {final_message.usage.output_tokens}")
108```
109 
110---
111 
112## Streaming with Progress Updates
113 
114```python
115def stream_with_progress(client, **kwargs):
116    """Stream a response with progress updates."""
117    total_tokens = 0
118    content_parts = []
119 
120    with client.messages.stream(**kwargs) as stream:
121        for event in stream:
122            if event.type == "content_block_delta":
123                if event.delta.type == "text_delta":
124                    text = event.delta.text
125                    content_parts.append(text)
126                    print(text, end="", flush=True)
127 
128            elif event.type == "message_delta":
129                if event.usage and event.usage.output_tokens is not None:
130                    total_tokens = event.usage.output_tokens
131 
132        final_message = stream.get_final_message()
133 
134    print(f"\n\n[Tokens used: {total_tokens}]")
135    return "".join(content_parts)
136```
137 
138---
139 
140## Error Handling in Streams
141 
142```python
143try:
144    with client.messages.stream(
145        model="claude-opus-4-8",
146        max_tokens=64000,
147        messages=[{"role": "user", "content": "Write a story"}]
148    ) as stream:
149        for text in stream.text_stream:
150            print(text, end="", flush=True)
151except anthropic.APIConnectionError:
152    print("\nConnection lost. Please retry.")
153except anthropic.RateLimitError:
154    print("\nRate limited. Please wait and retry.")
155except anthropic.APIStatusError as e:
156    print(f"\nAPI error: {e.status_code}")
157```
158 
159---
160 
161## Stream Event Types
162 
163| Event Type            | Description                 | When it fires                     |
164| --------------------- | --------------------------- | --------------------------------- |
165| `message_start`       | Contains message metadata   | Once at the beginning             |
166| `content_block_start` | New content block beginning | When a text/tool_use block starts |
167| `content_block_delta` | Incremental content update  | For each token/chunk              |
168| `content_block_stop`  | Content block complete      | When a block finishes             |
169| `message_delta`       | Message-level updates       | Contains `stop_reason`, usage     |
170| `message_stop`        | Message complete            | Once at the end                   |
171 
172## Best Practices
173 
1741. **Always flush output** — Use `flush=True` to show tokens immediately
1752. **Handle partial responses** — If the stream is interrupted, you may have incomplete content
1763. **Track token usage** — The `message_delta` event contains usage information
1774. **Use timeouts** — Set appropriate timeouts for your application
1785. **Default to streaming** — Use `.get_final_message()` to get the complete response even when streaming, giving you timeout protection without needing to handle individual events
1796. **Large `max_tokens` without streaming raises `ValueError`** — The SDK refuses non-streaming requests it estimates will exceed ~10 minutes (idle connections drop). Pass `stream=True` / use `messages.stream()`, or explicitly override `timeout`, to suppress the guard.
180