Display generated text incrementally and distinguish stream events from final output.

Download the runnable notebook

This complete lesson was converted from the matching notebook stored locally in the TechClub repository. Outputs are cleared; never add API keys or private data.

Goal, prerequisites, and setup

Understand why streaming improves perceived latency without changing the final answer. Complete the requests lesson first; know Python loops, context managers, and iterators. Install anthropic and python-dotenv, set ANTHROPIC_API_KEY and ANTHROPIC_MODEL, and test with a short prompt before using a long response.

Run the raw-event example to inspect event types, then run the text-stream example. Save the final message separately from the displayed chunks and note what should happen when the network fails mid-stream.

"""
RESPONSE STREAMING
-> Enables receiving the model's response in smaller chunks as they are generated.

COMMON EVENTS:
-> 'RawMessageStartEvent': A new message is being sent.
-> 'RawContentBlockStartEvent': Start of a new block, which contains a text message, tool use, or other.
-> 'RawContentBlockDeltaEvent': Chunk related to the latest content block that was started.
-> 'RawContentBlockStopEvent': The current content block has been completed.
-> 'RawMessageDeltaEvent': The current message is complete.
-> 'RawMessageStopEvent': End of information about the current message.
"""

# Load environment variables
from dotenv import load_dotenv

load_dotenv()

# Create an API client
import os
from anthropic import Anthropic

client = Anthropic()

model = os.getenv("ANTHROPIC_MODEL")

# Helpers
def add_message(messages, content, role):
    message = {"role": role, "content": content}
    messages.append(message)

Understanding Stream Events

When you enable streaming, Claude sends back several types of events:

  • MessageStart - A new message is being sent
  • ContentBlockStart - Start of a new block containing text, tool use, or other content
  • ContentBlockDelta - Chunks of the actual generated text
  • ContentBlockStop - The current content block has been completed
  • MessageDelta - The current message is complete
  • MessageStop - End of information about the current message

Basic Streaming Implementation

To enable streaming, add ‘stream=True’ to your messages.create call:

messages = []

add_message(messages,
            "Write a 1 sentence description of a fake database",
            role="user")
steam = client.messages.create(
    model=model,
    max_tokens=1000,
    messages=messages,
    stream=True,
)

for event in steam:
    print(event)

messages = []

add_message(messages,
            "Write a 1 sentence description of a fake database",
            role="user")

with client.messages.stream(
    model=model,
    max_tokens=1000,
    messages=messages,
) as stream:
    for text in stream.text_stream:
        # print(text, end="") # end="" to avoid double newlines
        pass

stream.get_final_message()

Evidence and references