Introduction to Streaming Responses with Claude Code
When building applications powered by large language models like Claude, the way you deliver responses to your users can make or break the user experience. Traditional request-response patterns force users to stare at loading spinners while the model generates an entire response. Streaming responses flip this model on its head by delivering tokens as they are generated, dramatically reducing perceived latency and creating a more interactive, conversational feel.
Claude Code, Anthropic's command-line tool and SDK for working with Claude models, provides robust streaming capabilities designed for production workloads. In this guide, we'll explore everything you need to know to implement, optimize, and scale streaming responses in real-world applications.
What Is Response Streaming?
Response streaming is a communication pattern where the server sends data to the client incrementally as it becomes available, rather than waiting until the entire payload is ready. In the context of Claude and other LLMs, this means sending individual tokens (or small groups of tokens) to the client the moment the model produces them.
Under the hood, Claude uses Server-Sent Events (SSE) as the transport mechanism for streaming. Each event contains metadata about the generation state — whether it's starting, progressing, or complete — along with the actual content tokens. This structured approach allows clients to render partial responses, handle errors gracefully, and provide real-time feedback to users.
Key Streaming Event Types
Claude's streaming API emits several distinct event types during a generation. Understanding these events is essential for building robust clients:
message_start— Fired when the generation begins, includes message metadata and the initial role.content_block_start— Indicates the beginning of a new content block (text, tool use, etc.).content_block_delta— Contains incremental content updates, the most frequently emitted event.content_block_stop— Signals the end of a content block.message_delta— Carries message-level metadata updates, including stop reason and usage statistics.message_stop— Fired when the entire message is complete.ping— Keep-alive events sent periodically to maintain the connection.error— Emitted when something goes wrong during generation.
Why Streaming Matters in Production
The benefits of streaming extend far beyond a nicer user experience. In production environments, streaming addresses several critical concerns that affect both users and infrastructure.
Perceived Latency Reduction
Time-to-first-token (TTFT) is one of the most important metrics for LLM applications. With traditional responses, users wait for the entire generation to complete before seeing anything. For a 500-token response that takes 10 seconds to generate, that's 10 seconds of dead air. With streaming, users see the first token in under a second, and content flows continuously thereafter. This transforms the experience from "waiting" to "reading."
Improved Cancellation Handling
Streaming enables true cancellation. When users can see output as it's generated, they can decide early that a response isn't what they wanted and cancel the request. This saves compute costs, reduces unnecessary token usage, and frees up capacity for other requests. Without streaming, you're committed to the full generation cost regardless of whether the user wants it.
Better Error Recovery
With streaming, partial responses are preserved even if the connection drops mid-generation. Your application can cache received tokens and resume from where things left off, rather than losing everything and starting over. This is particularly valuable for long-form content generation where a failure at the 90% mark shouldn't mean losing the entire response.
Real-Time Feedback and Interactivity
Streaming opens the door to interactive experiences. You can display typing indicators, show token counts in real time, implement progressive rendering of markdown or code blocks, and even trigger downstream processing as soon as relevant content arrives. This is impossible with batch responses.
Setting Up Your Environment
Before diving into code, ensure your environment is properly configured. You'll need an Anthropic API key and the appropriate SDK installed. The examples in this guide use Python and TypeScript, the two most common languages for Claude Code development.
Python Setup
# Install the Anthropic Python SDK
pip install anthropic
# Set your API key as an environment variable
export ANTHROPIC_API_KEY="your-api-key-here"
TypeScript Setup
// Install the Anthropic TypeScript SDK
npm install @anthropic-ai/sdk
// Set your API key as an environment variable
// In your shell or .env file:
// ANTHROPIC_API_KEY=your-api-key-here
Basic Streaming Implementation
Let's start with a simple streaming example to understand the fundamental pattern. We'll send a message to Claude and print tokens as they arrive.
Streaming in Python
import anthropic
client = anthropic.Anthropic()
def stream_basic_response():
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[
{"role": "user", "content": "Explain how database indexing works in simple terms."}
]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print() # Final newline
if __name__ == "__main__":
stream_basic_response()
The stream() context manager handles connection lifecycle automatically. The text_stream property provides a convenient iterator that yields only the text content, abstracting away the underlying event structure. The flush=True parameter ensures each token is printed immediately rather than being buffered.
Streaming in TypeScript
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
async function streamBasicResponse() {
const stream = client.messages.stream({
model: "claude-sonnet-4-20250514",
max_tokens: 1024,
messages: [
{ role: "user", content: "Explain how database indexing works in simple terms." }
],
});
for await (const event of stream) {
if (
event.type === "content_block_delta" &&
event.delta.type === "text_delta"
) {
process.stdout.write(event.delta.text);
}
}
console.log(); // Final newline
}
streamBasicResponse();
In TypeScript, the stream() method returns an async iterable. We iterate over raw events and filter for text_delta events to extract text content. This lower-level approach gives you full control over event handling.
Working with Raw Stream Events
While the simplified text stream is convenient, production applications often need access to the full event stream. This is necessary when you need to handle tool use, track usage statistics, or implement custom error handling.
Handling All Event Types in Python
import anthropic
import json
client = anthropic.Anthropic()
def stream_with_full_events():
collected_text = []
total_input_tokens = 0
total_output_tokens = 0
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[
{"role": "user", "content": "Write a Python function to merge two sorted lists."}
]
) as stream:
for event in stream:
if event.type == "message_start":
print(f"[Message started] Model: {event.message.model}")
elif event.type == "content_block_start":
print(f"\n[Content block started] Type: {event.content_block.type}")
elif event.type == "content_block_delta":
if event.delta.type == "text_delta":
collected_text.append(event.delta.text)
print(event.delta.text, end="", flush=True)
elif event.type == "content_block_stop":
print("\n[Content block complete]")
elif event.type == "message_delta":
if event.usage:
total_output_tokens = event.usage.output_tokens
if event.delta.stop_reason:
print(f"\n[Stop reason: {event.delta.stop_reason}]")
elif event.type == "message_stop":
print("[Message complete]")
print(f"\n\nTotal output tokens: {total_output_tokens}")
return "".join(collected_text)
if __name__ == "__main__":
stream_with_full_events()
This example demonstrates handling every event type in the stream. Notice how usage statistics arrive in the message_delta event, not message_start. The stop reason also arrives here, telling you why generation ended (e.g., end_turn, max_tokens, stop_sequence).
Streaming with Tool Use
Tool use adds complexity to streaming because the model interleaves text and structured tool-use blocks. You need to handle both content types and potentially execute tools mid-stream.
Streaming Tool Use in Python
import anthropic
import json
client = anthropic.Anthropic()
tools = [
{
"name": "get_weather",
"description": "Get the current weather for a given city.",
"input_schema": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "The city name"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["city"]
}
}
]
def get_weather(city: str, unit: str = "celsius") -> dict:
# Simulated weather API
return {"city": city, "temperature": 22, "unit": unit, "condition": "sunny"}
def stream_with_tools():
messages = [
{"role": "user", "content": "What's the weather in Tokyo and London?"}
]
while True:
tool_use_blocks = []
text_content = []
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
tools=tools,
messages=messages
) as stream:
current_tool_input = ""
for event in stream:
if event.type == "content_block_start":
block = event.content_block
if block.type == "tool_use":
print(f"\n[Tool call: {block.name}]")
current_tool_input = ""
elif event.type == "content_block_delta":
if event.delta.type == "text_delta":
text_content.append(event.delta.text)
print(event.delta.text, end="", flush=True)
elif event.delta.type == "input_json_delta":
current_tool_input += event.delta.partial_json
elif event.type == "content_block_stop":
if current_tool_input:
tool_input = json.loads(current_tool_input)
tool_use_blocks.append(tool_input)
current_tool_input = ""
response = stream.get_final_message()
# If no tool use, we're done
if response.stop_reason != "tool_use":
break
# Execute tools and continue conversation
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = get_weather(**block.input)
print(f"\n[Tool result: {result}]")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
messages.append({"role": "user", "content": tool_results})
if __name__ == "__main__":
stream_with_tools()
The key insight here is that tool input JSON arrives as partial fragments via input_json_delta events. You must accumulate these fragments and parse the complete JSON only when the content block stops. The outer while True loop handles multi-turn tool use, where Claude may call tools, receive results, and continue generating.
Building a Production Streaming Server
In production, you'll typically expose Claude's streaming through your own API. Let's build a FastAPI endpoint that proxies streaming responses to clients using Server-Sent Events.
FastAPI Streaming Endpoint
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from anthropic import Anthropic
import json
import asyncio
app = FastAPI()
client = Anthropic()
@app.post("/api/chat/stream")
async def stream_chat(request: Request):
body = await request.json()
user_message = body.get("message", "")
conversation_history = body.get("history", [])
messages = conversation_history + [
{"role": "user", "content": user_message}
]
async def event_generator():
try:
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=4096,
messages=messages,
system="You are a helpful, concise assistant."
) as stream:
for event in stream:
if event.type == "content_block_delta":
if event.delta.type == "text_delta":
# Send each text delta as an SSE event
sse_data = json.dumps({
"type": "text",
"content": event.delta.text
})
yield f"data: {sse_data}\n\n"
elif event.type == "message_delta":
if event.usage:
usage_data = json.dumps({
"type": "usage",
"output_tokens": event.usage.output_tokens
})
yield f"data: {usage_data}\n\n"
elif event.type == "message_stop":
yield f"data: {json.dumps({'type': 'done'})}\n\n"
except Exception as e:
error_data = json.dumps({
"type": "error",
"message": str(e)
})
yield f"data: {error_data}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable Nginx buffering
}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Several production considerations are baked into this endpoint. The X-Accel-Buffering: no header is critical when running behind Nginx, which buffers responses by default and would break streaming. The Cache-Control: no-cache header prevents intermediary proxies from caching stream data. The error handling wraps the entire stream in a try-except block, ensuring clients always receive a structured error event rather than a broken connection.
Client-Side Consumption with JavaScript
async function streamFromServer(message, history = []) {
const response = await fetch("/api/chat/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ message, history }),
});
if (!response.ok) {
throw new Error(`HTTP error: ${response.status}`);
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let fullText = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// Process complete SSE events (separated by double newlines)
const events = buffer.split("\n\n");
buffer = events.pop(); // Keep incomplete event in buffer
for (const eventStr of events) {
if (!eventStr.startsWith("data: ")) continue;
const data = JSON.parse(eventStr.slice(6));
switch (data.type) {
case "text":
fullText += data.content;
// Update UI with new text
document.getElementById("output").textContent = fullText;
break;
case "usage":
console.log(`Output tokens: ${data.output_tokens}`);
break;
case "done":
console.log("Stream complete");
break;
case "error":
console.error(`Stream error: ${data.message}`);
break;
}
}
}
return fullText;
}
// Usage
streamFromServer("Explain quantum computing in one paragraph.");
The client code uses the Fetch API with a streaming reader rather than the EventSource API. This is intentional — EventSource only supports GET requests, while most chat applications need POST requests with request bodies. The buffer-based SSE parsing handles cases where chunk boundaries fall in the middle of an event, which happens frequently in practice.
Error Handling and Resilience
Production streaming requires robust error handling. Network connections drop, rate limits trigger, and model errors occur. Your application must handle all of these gracefully.
Implementing Retry Logic with Exponential Backoff
import anthropic
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
client = anthropic.Anthropic()
def stream_with_retry(
messages,
max_retries=3,
base_delay=1.0,
max_delay=30.0
):
"""
Stream a response with automatic retry on transient failures.
Accumulated text is preserved across retries.
"""
accumulated_text = ""
last_error = None
for attempt in range(max_retries + 1):
try:
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=4096,
messages=messages
) as stream:
for text in stream.text_stream:
accumulated_text += text
print(text, end="", flush=True)
# Successfully completed
return accumulated_text
except anthropic.RateLimitError as e:
last_error = e
logger.warning(f"Rate limited on attempt {attempt + 1}")
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
time.sleep(delay)
except anthropic.APIConnectionError as e:
last_error = e
logger.warning(f"Connection error on attempt {attempt + 1}")
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
time.sleep(delay)
except anthropic.APIStatusError as e:
last_error = e
if e.status_code >= 500 and attempt < max_retries:
logger.warning(f"Server error {e.status_code} on attempt {attempt + 1}")
delay = min(base_delay * (2 ** attempt), max_delay)
time.sleep(delay)
else:
# Non-retryable error (4xx)
raise
raise last_error
# Usage
if __name__ == "__main__":
messages = [
{"role": "user", "content": "Write a detailed essay about renewable energy."}
]
stream_with_retry(messages)
This retry implementation distinguishes between retryable errors (rate limits, connection issues, 5xx server errors) and non-retryable errors (4xx client errors). The exponential backoff with a maximum delay prevents thundering herd problems when many clients retry simultaneously. Note that accumulated text is preserved across retries — if the connection drops after generating 500 tokens, the retry continues from the beginning but the client already has the first 500 tokens.
Handling Mid-Stream Failures Gracefully
import anthropic
from dataclasses import dataclass, field
client = anthropic.Anthropic()
@dataclass
class StreamResult:
text: str = ""
completed: bool = False
error: str | None = None
input_tokens: int = 0
output_tokens: int = 0
def safe_stream(messages, **kwargs) -> StreamResult:
"""Stream with comprehensive error capture."""
result = StreamResult()
try:
with client.messages.stream(
model=kwargs.get("model", "claude-sonnet-4-20250514"),
max_tokens=kwargs.get("max_tokens", 4096),
messages=messages,
system=kwargs.get("system", "")
) as stream:
for event in stream:
if event.type == "content_block_delta":
if event.delta.type == "text_delta":
result.text += event.delta.text
elif event.type == "message_start":
if event.message.usage:
result.input_tokens = event.message.usage.input_tokens
elif event.type == "message_delta":
if event.usage:
result.output_tokens = event.usage.output_tokens
elif event.type == "message_stop":
result.completed = True
except anthropic.APIStatusError as e:
result.error = f"API error {e.status_code}: {e.message}"
except anthropic.APIConnectionError as e:
result.error = f"Connection error: {str(e)}"
except Exception as e:
result.error = f"Unexpected error: {str(e)}"
return result
# Usage
if __name__ == "__main__":
result = safe_stream(
[{"role": "user", "content": "Explain recursion."}],
max_tokens=1024
)
if result.completed:
print(f"Success! {result.output_tokens} tokens generated.")
print(result.text)
else:
print(f"Stream failed: {result.error}")
if result.text:
print(f"Partial output ({len(result.text)} chars): {result.text[:200]}...")
The StreamResult dataclass captures everything you need to handle both successful and failed streams. Even when a stream fails, you retain the partial text, token counts, and a structured error message. This makes it easy to implement fallback logic, log failures for debugging, or present partial results to users with an appropriate warning.
Performance Optimization
Streaming at scale requires attention to performance at every layer. Here are the key optimizations to implement.
Connection Pooling
import anthropic
from anthropic import Anthropic
# Create a single client instance and reuse it
# The underlying HTTP client maintains a connection pool
client = Anthropic(
max_retries=2,
timeout=anthropic.Timeout(
connect=10.0,
read=120.0, # Long read timeout for streaming
write=10.0,
pool=5.0
)
)
# For high-throughput applications, configure the HTTP client
import httpx
custom_http_client = httpx.Client(
limits=httpx.Limits(
max_connections=100,
max_keepalive_connections=20,
keepalive_expiry=30
)
)
client = Anthropic(http_client=custom_http_client)
Connection pooling is critical for throughput. Creating a new HTTP connection for every request adds significant latency, especially with TLS handshake overhead. By reusing a single client instance with a configured connection pool, you amortize connection costs across many requests. The extended read timeout is essential for streaming, as long generations can take minutes.
Concurrent Streaming
import anthropic
import asyncio
from typing import List, Dict
client = anthropic.AsyncAnthropic()
async def stream_single_message(
message: str,
system_prompt: str = ""
) -> Dict:
"""Stream a single message and return the complete result."""
result = {"text": "", "error": None, "tokens": 0}
try:
async with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[{"role": "user", "content": message}],
system=system_prompt
) as stream:
async for text in stream.text_stream:
result["text"] += text
final = await stream.get_final_message()
result["tokens"] = final.usage.output_tokens
except Exception as e:
result["error"] = str(e)
return result
async def stream_concurrent(messages: List[str]) -> List[Dict]:
"""Stream multiple messages concurrently."""
tasks = [stream_single_message(msg) for msg in messages]
results = await asyncio.gather(*tasks)
return results
# Usage
if __name__ == "__main__":
prompts = [
"Summarize the French Revolution in 3 sentences.",
"Explain photosynthesis briefly.",
"What are the benefits of exercise?",
]
results = asyncio.run(stream_concurrent(prompts))
for i, result in enumerate(results):
print(f"\n--- Response {i + 1} ---")
if result["error"]:
print(f"Error: {result['error']}")
else:
print(result["text"])
print(f"Tokens: {result['tokens']}")
Using the async client (AsyncAnthropic) with asyncio.gather allows you to process multiple streams concurrently. This is invaluable for batch processing, multi-agent workflows, or any scenario where you need to generate multiple responses in parallel. Be mindful of rate limits — concurrent requests share the same rate limit pool, so you may need to implement a semaphore to control concurrency.
Buffered Output for UI Rendering
import anthropic
import re
client = anthropic.Anthropic()
class MarkdownStreamRenderer:
"""Buffers tokens to render markdown blocks progressively."""
def __init__(self):
self.buffer = ""
self.in_code_block = False
self.code_block_start = 0
def process_token(self, token: str) -> str:
"""Process a token and return renderable content."""
self.buffer += token
output = ""
# Detect code block boundaries
if "" in token:
if not self.in_code_block:
# Starting a code block - wait for language identifier
self.in_code_block = True
self.code_block_start = len(self.buffer) - token.index("")
return "" # Don't output yet
else:
# Ending a code block - output the complete block
self.in_code_block = False
code_block = self.buffer[self.code_block_start:]
output = f"\n[CODE BLOCK]\n{code_block}\n[/CODE BLOCK]\n"
return output
if self.in_code_block:
# Accumulate code block content
return ""
# For non-code content, output immediately
return token
def flush(self) -> str:
"""Flush any remaining buffered content."""
if self.in_code_block and self.buffer[self.code_block_start:]:
return self.buffer[self.code_block_start:]
return ""
def stream_with_rendering():
renderer = MarkdownStreamRenderer()
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[
{"role": "user", "content": "Write a Python function with explanation to check if a string is a palindrome."}
]
) as stream:
for text in stream.text_stream:
rendered = renderer.process_token(text)
if rendered:
print(rendered, end="", flush=True)
# Flush any remaining content
remaining = renderer.flush()
if remaining:
print(remaining, end="", flush=True)
if __name__ == "__main__":
stream_with_rendering()
When rendering markdown or code blocks, naive token-by-token output can produce visual glitches. Code blocks delimited by triple backticks may be split across tokens, causing partial fences to render incorrectly. The MarkdownStreamRenderer buffers content when inside a code block and outputs the complete block once it's finished, while streaming regular text immediately for maximum responsiveness.
Best Practices for Production Streaming
1. Always Set Reasonable Max Tokens
Never leave max_tokens unset in production. A runaway generation can consume significant tokens and cost. Set limits based on your use case — 1024 for short responses, 4096 for medium content, and higher only when explicitly needed.
2. Implement Proper Timeout Handling
import anthropic
client = anthropic.Anthropic(
timeout=anthropic.Timeout(
connect=10.0, # Connection timeout
read=300.0, # Read timeout (5 minutes for long streams)
write=10.0, # Write timeout
pool=5.0 # Connection pool timeout
),
max_retries=2 # SDK-level retries for transient errors
)
Streaming responses can take minutes for long generations. Your read timeout must accommodate this, or you'll see premature timeouts on legitimate long responses. However, keep the connect timeout short so you fail fast when the API is unreachable.
3. Log Token Usage for Cost Monitoring
import anthropic
import logging
from datetime import datetime
logger = logging.getLogger("claude_usage")
handler = logging.FileHandler("claude_usage.log")
handler.setFormatter(logging.Formatter(
'%(asctime)s - %(message)s'
))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
client = anthropic.Anthropic()
def stream_with_usage_logging(user_id: str, messages: list):
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=messages
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
final = stream.get_final_message()
logger.info(
f"user={user_id} "
f"model={final.model} "
f"input_tokens={final.usage.input_tokens} "
f"output_tokens={final.usage.output_tokens} "
f"stop_reason={final.stop_reason}"
)
Token usage logging is essential for cost control and capacity planning. Log every request with user identifiers, token counts, and stop reasons. This data helps you identify heavy users, detect anomalous usage patterns, and forecast API costs.
4. Use Streaming for Long-Running Operations
Streaming isn't just for text generation. Use it for any operation that might take more than a few seconds, including tool-heavy workflows, multi-step reasoning, and code generation. The real-time feedback keeps users engaged and informed.
5. Handle Backpressure Properly
import anthropic
import queue
import threading
client = anthropic.Anthropic()
def stream_to_queue(messages: list, output_queue: queue.Queue, max_queue_size: int = 100):
"""Stream tokens into a queue with backpressure handling."""
try:
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=4096,
messages=messages
) as stream:
for text in stream.text_stream:
# Block if queue is full, providing backpressure
while output_queue.qsize() >= max_queue_size:
threading.Event().wait(0.01)
output_queue.put(text)
output_queue.put(None) # Sentinel to signal completion
except Exception as e:
output_queue.put(("error", str(e)))
# Usage: consumer processes at its own pace
if __name__ == "__main__":
q = queue.Queue(maxsize=100)
messages = [{"role": "user", "content": "Write a long story about space exploration."}]
# Start streaming in a background thread
producer = threading.Thread(
target=stream_to_queue,
args=(messages, q)
)
producer.start()
# Consume tokens at our own pace
while True:
item = q.get()
if item is None:
break
elif isinstance(item, tuple) and item[0] == "error":
print(f"\nError: {item[1]}")
break
else:
print(item, end="", flush=True)
producer.join()
Backpressure matters when the consumer can't keep up with the producer. If your UI rendering or downstream processing is slower than token generation, unbounded buffering can lead to memory issues. The queue-based approach with size limits ensures the producer slows down when the consumer falls behind.
6. Validate and Sanitize Streamed Content
Never assume streamed content is safe for direct rendering. If you're streaming into a web UI, sanitize HTML to prevent XSS attacks. If you're streaming into a database, validate content before insertion. Streaming doesn't change security requirements — it just changes the timing.
7. Implement Graceful Degradation
import anthropic
client = anthropic.Anthropic()
def stream_with_fallback(messages: list):
"""Try streaming first, fall back to non-streaming if it fails."""
try:
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=messages
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print()
return
except Exception as e:
print(f"\n[Streaming failed, falling back: {e}]")
# Fallback to non-streaming
try:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=messages
)
print(response.content[0].text)
except Exception as e:
print(f"Fallback also failed: {e}")
Some environments (certain proxies, corporate firewalls, older HTTP clients) don't handle SSE properly. Implementing a fallback to non-streaming requests ensures your application remains functional even when streaming isn't available.
Monitoring and Observability
Production streaming requires visibility into what's happening in real time. Here's a comprehensive monitoring setup.
Metrics Collection
import anthropic
import time
from dataclasses import dataclass, field
from typing import Optional
client = anthropic.Anthropic()
@dataclass
class StreamMetrics:
request_id: str = ""
model: str = ""
start_time: float = 0.0
first_token_time: Optional[float] = None
end_time: Optional[float] = None
input_tokens: int = 0
output_tokens: int = 0
stop_reason: str = ""
error: Optional[str] = None
@property
def time_to_first_token(self) -> Optional[float]:
if self.first_token_time and self.start_time:
return self.first_token_time - self.start_time
return None
@property
def total_duration(self) -> Optional[float]:
if self.end_time and self.start_time:
return self.end_time - self.start_time
return None
@property
def tokens_per_second(self) -> Optional[float]:
if self.total_duration and self.output_tokens and self.total_duration > 0:
return self.output_tokens / self.total_duration
return None
def report(self) -> str:
return (
f"request_id={self.request_id} "
f"model={self.model} "
f"ttft={self.time_to_first_token:.3f}s "
f"duration={self.total_duration:.3f}s "
f"input_tokens={self.input_tokens} "
f"output_tokens={self.output_tokens} "
f"tps={self.tokens_per_second:.1f} "
f"stop_reason={self.stop_reason}"
)
def stream_with_metrics(messages: list, request_id: str = "req-001") -> StreamMetrics:
metrics = StreamMetrics(
request_id=request_id,
start_time=time.time()
)
try:
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=messages
) as stream:
for event in stream:
if event.type == "message_start":
metrics.model = event.message.model
if event.message.usage:
metrics.input_tokens = event.message.usage.input_tokens
elif event.type == "content_block_delta":
if event.delta.type == "text_delta" and metrics.first_token_time is None:
metrics.first_token_time = time.time()
print(event.delta.text, end="", flush=True)