StreamManager

View module source on GitHub

​

TextStreamWrapper

class TextStreamWrapper()

Wrapper for text stream events with callback support.

This class wraps a text stream to provide callback functionality for processing text events. It implements the iterator protocol to allow for seamless iteration over text content.

Attributes:

​

__init__

def __init__(wrapper, callback)

Initialize the text stream wrapper.

Arguments:

​

__iter__

def __iter__()

Return the iterator object.

Returns:

​

__next__

def __next__()

Get the next text content from the stream.

This method processes the next event from the stream, calls the callback function, and returns the text content if available.

Returns:

Raises:

• StopIteration - When the stream is exhausted.

​

StreamWrapper

class StreamWrapper(MessageStreamManager)

Wrapper for Anthropic MessageStreamManager with callback support.

This class wraps the Anthropic MessageStreamManager to provide callback functionality for processing stream events. It maintains compatibility with the original API while adding the ability to execute custom callbacks for each stream event.

The wrapper implements the context manager protocol and iterator protocol to provide seamless integration with existing code.

Attributes:

​

__init__

def __init__(mgr: MessageStreamManager,
             callback: Callable[[MessageStreamEvent], None]) -> None

Initialize the stream wrapper.

Arguments:

to be called for each stream event.

Notes:

Does not call super().init() since we’re wrapping another manager.

​

__enter__

def __enter__() -> MessageStream

Enter the stream context and return a callback-enabled stream.

This method creates a custom MessageStream that wraps the original stream to add callback functionality for each event.

Returns:

​

__exit__

def __exit__(exc_type: Union[type[BaseException], None],
             exc: Union[BaseException, None], exc_tb: Union[TracebackType,
                                                            None]) -> None

Exit the stream context.

This method delegates the context exit to the underlying stream manager to ensure proper cleanup and resource management.

Arguments:

​

__iter__

def __iter__()

Return the iterator object.

If no stream is active, this method will automatically enter the context to create one.

Returns:

​

__next__

def __next__()

Get the next event from the stream.

If no stream is active, this method will automatically enter the context to create one.

Returns:

Raises:

• StopIteration - When the stream is exhausted.

​

__getattr__

def __getattr__(name: str)

Delegate attribute access to the wrapped manager.

This method ensures that any attributes or methods not explicitly defined in this wrapper are delegated to the underlying stream manager.

Arguments:

Returns: