Streaming API - Quick Reference

def stream(
    self,
    *,
    model: str,
    messages: list[MessageParam],
    max_tokens: int,
    **kwargs  # All messages.create() parameters supported
) -> MessageStreamManager

class MessageStreamManager:
    def __enter__(self) -> MessageStream
    def __exit__(self, exc_type, exc_val, exc_tb) -> None

class MessageStream:
    def __iter__(self) -> Iterator[MessageStreamEvent]

    @property
    def text_stream(self) -> Iterator[str]:
        """Iterate over text deltas only"""
        ...

    @property
    def current_message_snapshot(self) -> Message:
        """Current accumulated message during streaming"""
        ...

    def get_final_message(self) -> Message:
        """Complete message after stream ends"""
        ...

    def get_final_text(self) -> str:
        """Accumulated text after stream ends"""
        ...

MessageStreamEvent = Union[
    MessageStartEvent,       # Stream started
    MessageDeltaEvent,       # Usage/stop_reason updated
    MessageStopEvent,        # Stream completed
    ContentBlockStartEvent,  # New content block
    ContentBlockDeltaEvent,  # Content delta
    ContentBlockStopEvent,   # Block completed
]

class MessageStartEvent(BaseModel):
    type: Literal["message_start"]
    message: Message  # Initial skeleton

class ContentBlockDeltaEvent(BaseModel):
    type: Literal["content_block_delta"]
    index: int
    delta: ContentBlockDelta  # TextDelta | InputJSONDelta

class TextDelta(BaseModel):
    type: Literal["text_delta"]
    text: str  # Incremental text to append

class MessageDeltaEvent(BaseModel):
    type: Literal["message_delta"]
    delta: MessageDelta  # stop_reason, stop_sequence
    usage: MessageDeltaUsage  # output_tokens

class MessageStopEvent(BaseModel):
    type: Literal["message_stop"]

# Text streaming
with client.messages.stream(...) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# Event processing
with client.messages.stream(...) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="")

# Get final message
with client.messages.stream(...) as stream:
    for text in stream.text_stream:
        print(text, end="")
message = stream.get_final_message()