Skip to main content
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:
NameDescription
_wrapperThe underlying stream wrapper.
_callbackCallback function to process stream events.

​
__init__

def __init__(wrapper, callback)
Initialize the text stream wrapper. Arguments:
NameDescription
wrapperThe underlying stream wrapper to delegate to.
callbackCallback function to be called for each event.

​
__iter__

def __iter__()
Return the iterator object. Returns:
NameDescription
[TextStreamWrapper](/sdk/python/references/logger/anthropic/stream_manager)Self as the iterator.

​
__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:
NameDescription
strThe text content from the event, or empty string if no text.
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:
NameTypeDescription
_mgrMessageStreamManagerThe underlying stream manager.
_callbackCallableCallback function for processing events.
_streamMessageStreamThe active stream instance.

​
__init__

def __init__(mgr: MessageStreamManager,
             callback: Callable[[MessageStreamEvent], None]) -> None
Initialize the stream wrapper. Arguments:
NameTypeDescription
mgrMessageStreamManagerThe underlying stream manager to wrap.
callbackCallable[[MessageStreamEvent], None]Callback function
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:
NameDescription
MessageStreamA stream instance with callback support.

​
__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:
NameDescription
exc_typeThe exception type if an exception occurred.
excThe exception instance if an exception occurred.
exc_tbThe traceback if an exception occurred.

​
__iter__

def __iter__()
Return the iterator object. If no stream is active, this method will automatically enter the context to create one. Returns:
NameDescription
[StreamWrapper](/sdk/python/references/logger/anthropic/stream_manager)Self as the iterator.

​
__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:
NameDescription
MessageStreamEventThe next event from the stream.
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:
NameTypeDescription
namestrThe attribute name to access.
Returns:
NameDescription
AnyThe attribute value from the underlying manager.
⌘I