Implementing Streaming Responses — Real-time Responses with the Claude API

I still remember the first time I wired streaming into one of my own apps. Until then the screen sat silent for half a minute while the response was generated, and one tester told me they closed the tab because they assumed it had frozen. The moment tokens started flowing one by one, the experience felt completely different — even though the total processing time hadn't changed. As an indie developer I can't afford those silent seconds, so this switch mattered more than any UI polish. Below is the path for wiring up streaming in Python or TypeScript, in the order I verified it myself.

What is Streaming?

Streaming allows you to receive Claude's response token-by-token in real time, rather than waiting for the complete response. This provides immediate feedback to users and is essential for building responsive applications.

ℹ️

Streaming is particularly valuable for: - Chat applications (show response as it's being generated) - Long-form content (news articles, reports) - Real-time dashboards - Mobile applications with limited bandwidth - Improving perceived response time

Understanding Server-Sent Events (SSE)

Streaming responses use the Server-Sent Events (SSE) protocol, which sends data as a stream of events from the server to the client.

SSE Format

data: {"type":"content_block_start","index":0,"content_block":{"type":"text"}}

data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}

data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" world"}}

data: {"type":"message_stop"}

Each event is a JSON object with a type field indicating what kind of event it is.

Python Streaming Implementation

Basic Streaming Example

import anthropic
 
client = anthropic.Anthropic()
 
messages = [
    {
        "role": "user",
        "content": "Write a short story about a robot learning to paint."
    }
]
 
# Use stream=True to enable streaming
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=messages
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()  # Newline after complete response

The examples use claude-sonnet-5, the default model as of July 2026. Sonnet 5 launched on June 30, 2026 with introductory pricing ($2 per million input tokens / $10 per million output tokens through August 31, 2026), roughly 40% cheaper than Opus 4.8 on both sides — a sensible choice for chat-style workloads where streaming is heavy. For long-running, complex agent work, consider claude-opus-4-8 instead.

Detailed Event Handling

For more control, you can handle individual events:

import anthropic
 
client = anthropic.Anthropic()
 
messages = [
    {
        "role": "user",
        "content": "Explain quantum computing in simple terms."
    }
]
 
# Process events individually
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=messages
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            print(f"Starting content block {event.index}")
 
        elif event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
 
        elif event.type == "content_block_stop":
            print(f"\nFinished content block {event.index}")
 
        elif event.type == "message_stop":
            print("\nStream finished")

Collecting Full Response During Streaming

import anthropic
 
client = anthropic.Anthropic()
 
def stream_message(user_input: str) -> str:
    """Stream a message and return the full response"""
    full_response = ""
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_input}]
    ) as stream:
        for text in stream.text_stream:
            full_response += text
            print(text, end="", flush=True)
 
    print()  # Final newline
    return full_response
 
# Use the function
result = stream_message("What are the benefits of renewable energy?")
print(f"\nFull response ({len(result)} characters)")

TypeScript/JavaScript Streaming

Using the Official SDK

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
async function streamResponse() {
  const stream = client.messages.stream({
    model: "claude-sonnet-5",
    max_tokens: 1024,
    messages: [
      {
        role: "user",
        content: "Write a haiku about the moon."
      }
    ]
  });
 
  // Handle text events
  stream.on("text", (text: string) => {
    process.stdout.write(text);
  });
 
  // Get final message
  const finalMessage = await stream.finalMessage();
  console.log("\n\nFinal message:", finalMessage);
}
 
streamResponse();

Manual Event Processing

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
async function processStreamEvents() {
  const stream = await client.messages.create({
    model: "claude-sonnet-5",
    max_tokens: 1024,
    stream: true,
    messages: [
      {
        role: "user",
        content: "Describe the water cycle."
      }
    ]
  });
 
  for await (const event of stream) {
    switch (event.type) {
      case "content_block_start":
        console.log(`Starting block ${event.index}`);
        break;
 
      case "content_block_delta":
        if (event.delta.type === "text_delta") {
          process.stdout.write(event.delta.text);
        }
        break;
 
      case "content_block_stop":
        console.log(`\nFinished block ${event.index}`);
        break;
 
      case "message_stop":
        console.log("Stream complete");
        break;
    }
  }
}
 
processStreamEvents();

Web-Based Streaming (Frontend)

Fetch API with Streaming

async function streamFromAPI(userMessage) {
  const response = await fetch("https://api.anthropic.com/v1/messages", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "x-api-key": process.env.ANTHROPIC_API_KEY,
      "anthropic-version": "2023-06-01"
    },
    body: JSON.stringify({
      model: "claude-sonnet-5",
      max_tokens: 1024,
      stream: true,
      messages: [
        { role: "user", content: userMessage }
      ]
    })
  });
 
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let fullResponse = "";
 
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
 
    const chunk = decoder.decode(value);
    const lines = chunk.split("\n");
 
    for (const line of lines) {
      if (line.startsWith("data: ")) {
        const eventData = line.slice(6);
        try {
          const event = JSON.parse(eventData);
 
          if (event.type === "content_block_delta") {
            if (event.delta.type === "text_delta") {
              const text = event.delta.text;
              fullResponse += text;
              // Update UI in real time
              document.getElementById("response").textContent = fullResponse;
            }
          }
        } catch (e) {
          // Skip invalid JSON
        }
      }
    }
  }
 
  return fullResponse;
}

React Streaming Component

import { useState } from "react";
 
export function StreamingChat() {
  const [input, setInput] = useState("");
  const [response, setResponse] = useState("");
  const [loading, setLoading] = useState(false);
 
  const handleSubmit = async (e) => {
    e.preventDefault();
    setLoading(true);
    setResponse("");
 
    try {
      const fetchResponse = await fetch("/api/chat", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ message: input })
      });
 
      const reader = fetchResponse.body.getReader();
      const decoder = new TextDecoder();
      let fullText = "";
 
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
 
        const chunk = decoder.decode(value);
        const lines = chunk.split("\n");
 
        for (const line of lines) {
          if (line.startsWith("data: ")) {
            try {
              const event = JSON.parse(line.slice(6));
              if (event.type === "content_block_delta" &&
                  event.delta.type === "text_delta") {
                fullText += event.delta.text;
                setResponse(fullText);
              }
            } catch (e) {
              // Skip invalid JSON
            }
          }
        }
      }
    } finally {
      setLoading(false);
    }
  };
 
  return (
    <div>
      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder="Enter your message..."
          disabled={loading}
        />
        <button type="submit" disabled={loading}>
          {loading ? "Streaming..." : "Send"}
        </button>
      </form>
      <div id="response">{response}</div>
    </div>
  );
}

Stream Event Types

Supported Events

# Message start event
{
  "type": "message_start",
  "message": {
    "id": "msg_123",
    "type": "message",
    "role": "assistant",
    "stop_reason": null
  }
}
 
# Content block start
{
  "type": "content_block_start",
  "index": 0,
  "content_block": {
    "type": "text"
  }
}
 
# Text delta (token received)
{
  "type": "content_block_delta",
  "index": 0,
  "delta": {
    "type": "text_delta",
    "text": "Hello"
  }
}
 
# Content block stop
{
  "type": "content_block_stop",
  "index": 0
}
 
# Message delta (final statistics)
{
  "type": "message_delta",
  "delta": {
    "stop_reason": "end_turn"
  },
  "usage": {
    "output_tokens": 42
  }
}
 
# Message stop
{
  "type": "message_stop"
}

Error Handling in Streams

Handling Stream Errors

import anthropic
 
client = anthropic.Anthropic()
 
try:
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello"}]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
 
except anthropic.APIError as e:
    print(f"API Error: {e}")
    # Handle rate limits, authentication errors, etc.
 
except anthropic.APIConnectionError as e:
    print(f"Connection Error: {e}")
    # Handle network errors
 
except anthropic.RateLimitError as e:
    print(f"Rate Limited: {e}")
    # Implement exponential backoff

Timeout Handling

async function streamWithTimeout(
  userMessage: string,
  timeoutMs: number = 30000
) {
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
 
  try {
    const stream = client.messages.stream({
      model: "claude-sonnet-5",
      max_tokens: 1024,
      messages: [{ role: "user", content: userMessage }],
      signal: controller.signal
    });
 
    for await (const event of stream) {
      if (event.type === "content_block_delta" &&
          event.delta.type === "text_delta") {
        console.log(event.delta.text);
      }
    }
  } catch (error) {
    if (error.name === "AbortError") {
      console.error("Stream timeout");
    } else {
      throw error;
    }
  } finally {
    clearTimeout(timeoutId);
  }
}

UI/UX Best Practices

Progressive Display

Show response as it arrives:

<div id="chat">
  <div class="message user">Your question</div>
  <div class="message assistant" id="response">
    <!-- Response appears here, token by token -->
  </div>
</div>

Loading Indicators

function showLoadingIndicator() {
  const response = document.getElementById("response");
  response.innerHTML = '<span class="loading">Thinking...</span>';
}
 
function updateResponse(token) {
  const response = document.getElementById("response");
  // Remove loading indicator on first token
  if (response.textContent.includes("Thinking")) {
    response.textContent = token;
  } else {
    response.textContent += token;
  }
}

Handling Network Issues

import anthropic
import time
 
def stream_with_retry(prompt, max_retries=3):
    client = anthropic.Anthropic()
 
    for attempt in range(max_retries):
        try:
            with client.messages.stream(
                model="claude-sonnet-5",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            ) as stream:
                for text in stream.text_stream:
                    yield text
            return  # Success
 
        except anthropic.APIConnectionError as e:
            if attempt == max_retries - 1:
                raise  # Last attempt
            wait_time = 2 ** attempt  # Exponential backoff
            print(f"Retry in {wait_time}s...")
            time.sleep(wait_time)

Performance Tips

Always use streaming for long responses — Better UX and reduced perceived latency
Implement debouncing — Don't update UI on every token for very fast streams
Use connection pooling — Reuse HTTP connections for multiple streams
Monitor stream health — Track errors and retry failures
Set reasonable timeouts — Prevent hanging connections

⚠️

Don't buffer the entire response before showing it to users. Stream tokens as they arrive for the best user experience.

Worked Example: A Streaming Chat Application

import anthropic
import json
 
def chat_with_streaming(conversation_history: list[dict]):
    """Stream a chat conversation with memory"""
    client = anthropic.Anthropic()
 
    # Add system context
    messages = [
        {
            "role": "user",
            "content": "You are a helpful assistant."
        }
    ] + conversation_history
 
    full_response = ""
 
    print("Assistant: ", end="", flush=True)
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=messages
    ) as stream:
        for text in stream.text_stream:
            full_response += text
            print(text, end="", flush=True)
 
    print()  # New line
    return full_response
 
# Example usage
history = []
while True:
    user_input = input("You: ")
    if user_input.lower() == "quit":
        break
 
    history.append({"role": "user", "content": user_input})
    assistant_response = chat_with_streaming(history)
    history.append({"role": "assistant", "content": assistant_response})

Next Steps

Implement streaming in your chat application
Build a real-time dashboard using streaming
Add error recovery to your streaming implementation
Optimize UI updates for high-speed streams

If you want to keep going, Implementing Claude API SSE Streaming in Next.js App Router is a natural next read for wiring this into an App Router project. And for detecting streams that stop without raising an error — and resuming them mid-stream — I've written up the operational details in When Claude API Streaming Stops Without an Error.

Streaming looks impressive, but as an indie developer I've found the hard part is never the happy path — it's cleaning up after a connection that drops mid-stream. Across the Dolice Labs apps I now lock down reconnection and a fallback for partial output before polishing any UI. The shine you can add later; how it breaks is worth deciding first.

Implementing Streaming Responses — Real-time Responses with the Claude API

What is Streaming?

Understanding Server-Sent Events (SSE)

SSE Format

Python Streaming Implementation

Basic Streaming Example

Detailed Event Handling

Collecting Full Response During Streaming

TypeScript/JavaScript Streaming

Using the Official SDK

Manual Event Processing

Web-Based Streaming (Frontend)

Fetch API with Streaming

React Streaming Component

Stream Event Types

Supported Events

Error Handling in Streams

Handling Stream Errors

Timeout Handling

UI/UX Best Practices

Progressive Display

Loading Indicators

Handling Network Issues

Performance Tips

Worked Example: A Streaming Chat Application

Next Steps

Thank You for Reading

Related Articles

Related Articles

⬡ API & SDK2026-06-27
When Claude API Streaming Stops Without an Error: Detecting Silent Stalls and Resuming Mid-Stream
How to catch the 'silent stall' where Claude API streaming stops with no exception at all, using a content-level watchdog that times the gap between tokens, plus a resume path that carries received text forward as an assistant prefill, and a four-layer timeout budget for long-running automation.

⬡ API & SDK2026-05-06
Claude API × Python in Practice: Building an AI Assistant with Tool Calling and Streaming
A practical guide to combining Claude API's Tool Use and Streaming in Python. Build a working AI assistant with real tool execution, complete source code included, plus a breakdown of the tricky parts that trip up most developers.

⬡ API & SDK2026-05-04
Claude API stop_sequences Not Working — 5 Things to Check Before You Give Up
Diagnose why your Claude API stop_sequences parameter isn't halting generation as expected. Practical breakdown of token boundaries, whitespace mismatches, Tool Use interactions, and streaming pitfalls — with copy-paste code examples.