Streaming Responses 🌊

Why streaming matters

The model generates one token at a time (Lesson 53). Without streaming, your code waits for the whole reply, then prints it: a long, dead pause. With streaming, you print each token the instant it arrives, so the answer types itself out. The total time is the same; the felt time is transformed. Every chat interface you have ever enjoyed using does this.

The basic stream

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Tell me a two-line joke about Python."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)      # each chunk, immediately

print()      # a newline after the reply finishes

Three things make this work, and each matters:

Getting the full reply after streaming

You stream for the user's benefit, but you still need the complete text to append to your history (Lesson 55). The stream helper keeps it for you:

import anthropic

client = anthropic.Anthropic()

messages = [{"role": "user", "content": "Name three uses for Python."}]

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()

    # after the loop, get the assembled message for your history
    final = stream.get_final_message()

messages.append({"role": "assistant", "content": final.content[0].text})
print(f"\n[streamed reply was {final.usage.output_tokens} output tokens]")

get_final_message() hands back the complete response object, including the full text and the token usage. So you get the best of both: a responsive display for the human, and the whole reply for your program.

A streaming chat loop

import anthropic

client = anthropic.Anthropic()
SYSTEM = "You are Jarvis. Concise, warm, and quick."


def streaming_chat():
    messages = []
    while True:
        text = input("You: ")
        if text.strip().lower() in {"quit", "exit"}:
            break
        messages.append({"role": "user", "content": text})

        print("Jarvis: ", end="", flush=True)
        with client.messages.stream(
            model="claude-sonnet-5",
            max_tokens=1024,
            system=SYSTEM,
            messages=messages,
        ) as stream:
            for chunk in stream.text_stream:
                print(chunk, end="", flush=True)
            reply = stream.get_final_message().content[0].text
        print("\n")

        messages.append({"role": "assistant", "content": reply})


if __name__ == "__main__":
    streaming_chat()

That is the Lesson 55 loop with streaming spliced in. Nothing else changed, and it feels like a different, far better program. This is the version you actually want to use.

Streaming and thinking, and the async version

Some models can show their reasoning as a separate stream of "thinking" before the answer. If you want to surface that, you handle stream events rather than just text:

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Plan a simple weekly meal prep."}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
print()

For a web app serving many users at once, you want the async client (Lesson 40), so one waiting request does not block the others. It is the same shape with await and async for:

import asyncio
import anthropic


async def ask(question):
    client = anthropic.AsyncAnthropic()
    async with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": question}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
    print()


asyncio.run(ask("What is asyncio good for, in one line?"))

This is exactly the payoff promised back in Lesson 40: streaming a language model's reply is the textbook case for async, because your program spends almost all its time waiting for tokens to arrive over the network.

import time
import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-5", max_tokens=500,
    messages=[{"role": "user", "content": "Say hello in a friendly way."}],
) as stream:
    for text in stream.text_stream:
        for char in text:
            print(char, end="", flush=True)
            time.sleep(0.005)      # a whisper of delay per character
print()

with client.messages.stream(model="claude-sonnet-5", max_tokens=500,
        messages=messages) as stream:
    for text in stream.text_stream:
        print(text, end="")