Streaming responses

Reading a stream in Python

The academy's proxy doesn't pass streams through. A request with "stream": true gets a 400 error from lesson code. So this lesson works with a recording: the exact text a stream sends. The code that reads it is the same code you'd run on a live stream, one line at a time.

import json

RECORDED = """event: message_start
data: {"type":"message_start","message":{"id":"msg_01Example","model":"claude-opus-5-5","content":[],"usage":{"input_tokens":31,"output_tokens":1}}}

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

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Rain jackets"}}

event: ping
data: {"type":"ping"}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" 30% off,"}}

event: message_stop
data: {"type":"message_stop"}"""

for line in RECORDED.splitlines():
    if line.startswith("data: "):
        event = json.loads(line[6:])
        print(event["type"])

Here's what each piece does:

• splitlines() breaks a string into a list of its lines.
• Only the data: lines matter. Each one repeats the event's name in its "type" field.
• line[6:] slices off the six characters of "data: ", counting the space. What's left is JSON that json.loads can read.

Rebuilding the reply

Printing event names is a start. To get the reply itself, you need a loop that adds each text delta to a string. Step through it and watch the string grow:

pieces = ["The ", "sky ", "is ", "blue"]
text = ""
for p in pieces:
    text += p
print(len(text.split()))

import json

RECORDED = """event: message_start
data: {"type":"message_start","message":{"id":"msg_01Example","model":"claude-opus-5-5","content":[],"usage":{"input_tokens":31,"output_tokens":1}}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}

event: content_block_stop
data: {"type":"content_block_stop","index":0}

event: content_block_start
data: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Rain jackets"}}

event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":" 30% off,"}}

event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":" through Sunday"}}

event: content_block_stop
data: {"type":"content_block_stop","index":1}

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":87}}

event: message_stop
data: {"type":"message_stop"}"""

text = ""
stop_reason = None
input_tokens = 0
output_tokens = 0

for line in RECORDED.splitlines():
    if not line.startswith("data: "):
        continue
    event = json.loads(line[6:])
    if event["type"] == "message_start":
        input_tokens = event["message"]["usage"]["input_tokens"]
    elif event["type"] == "content_block_delta" and event["delta"]["type"] == "text_delta":
        piece = event["delta"]["text"]
        print(piece, end="")
        text += piece
    elif event["type"] == "message_delta":
        stop_reason = event["delta"]["stop_reason"]
        output_tokens = event["usage"]["output_tokens"]

print()
print("stop_reason:", stop_reason)
print("tokens:", input_tokens, "in,", output_tokens, "out")

Streaming on your own machine

Outside the academy, the requests library can read a live stream line by line:

import json
import os
import requests

response = requests.post(
    "https://api.anthropic.com/v1/messages",
    headers={
        "x-api-key": os.environ["ANTHROPIC_API_KEY"],
        "anthropic-version": "2026-10-21",
        "content-type": "application/json",
    },
    json={
        "model": "claude-opus-5-5",
        "max_tokens": 2000,
        "stream": True,
        "messages": [{"role": "user", "content": "Write a short welcome email for new newsletter subscribers."}],
    },
    stream=True,
    timeout=60,
)
response.raise_for_status()
for line in response.iter_lines(decode_unicode=True):
    if line.startswith("data: "):
        event = json.loads(line[6:])
        if event["type"] == "content_block_delta" and event["delta"]["type"] == "text_delta":
            print(event["delta"]["text"], end="", flush=True)

There are two stream settings here, and you need both:

• "stream": True inside the JSON body asks the API to stream.
• stream=True as an argument to requests.post tells requests to hand you the response as it arrives. Without it, requests waits for all of it.

Then iter_lines gives you one line at a time. flush=True makes each piece show up on screen right away instead of waiting in a queue.

Letting the SDK do the work

An SDK (software development kit) is a ready-made library for calling an API. The official SDK handles the events for you:

import anthropic

client = anthropic.Anthropic()   # reads ANTHROPIC_API_KEY from the environment
with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=2000,
    messages=[{"role": "user", "content": "Write a short welcome email for new newsletter subscribers."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

print()
print(final.stop_reason, final.usage.output_tokens)

Reading it from the top:

• with ... as stream: opens the stream. It also makes sure the connection closes when the block ends, even if something fails.
• text_stream gives you only the text pieces.
• get_final_message() gives you the complete response, as if you hadn't streamed. It has content, stop_reason and usage.

If you only want streaming to avoid timeouts, that last call is all you need.

Errors in the middle of a stream

A streaming response sends its 200 OK status before the model has written anything. So a problem partway through can't change the status code. It arrives as an event instead:

event: error
data: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}

Think of a phone call that drops halfway through a sentence. The call connected fine, but you didn't hear the whole message. Your loop has to watch for this event. When one arrives, the text so far is incomplete. Throw it away or label it as cut off. Then retry, or tell the user.

The same goes for a stop_reason of "max_tokens". That means the reply hit your length limit, exactly as without streaming.

When to stream

Stream in three cases:

• A person is watching the reply being written. Think of a chat, a drafting tool or an assistant in a dashboard.
• The replies are long.
• max_tokens is large enough that a single waiting request could time out.

Don't bother for background jobs nobody watches, such as tagging 5,000 reviews overnight. Short replies don't need it either.

Structured output is a special case. Half a JSON object can't be parsed, so your code waits for the end either way. Stream it only to avoid timeouts, and parse the final message.

Streaming changes how a reply is delivered, not whether it's right. Check the finished result before anything acts on it.