Skip to content

Extensions

Provider extensions. Importing an extension module auto-registers its input adapter with the adapter registry. See the providers guide and the AG-UI guide for usage.

Gemini

hother.streamblocks.extensions.gemini

Gemini extension for StreamBlocks.

This extension provides input adapters for Google GenAI streams.

Importing this module registers the GeminiInputAdapter for auto-detection.

Example

Import to enable auto-detection

import hother.streamblocks.extensions.gemini

Auto-detect from Gemini stream

processor = ProtocolStreamProcessor(registry) async for event in processor.process_stream(gemini_stream): ... print(event)

Or use convenience factory

from hother.streamblocks.extensions.gemini import create_gemini_processor processor = create_gemini_processor(registry)

GeminiInputAdapter

Input adapter for Google GenAI streams.

Handles chunks from google.genai.models.generate_content_stream() and google.ai.generativelanguage clients.

Extracts: - Text from chunk.text attribute - Usage metadata (token counts) - Model version information

Example

from google import genai adapter = GeminiInputAdapter()

async for chunk in client.aio.models.generate_content_stream(...): ... text = adapter.extract_text(chunk) ... metadata = adapter.get_metadata(chunk) ... if metadata: ... print(f"Tokens: {metadata['usage']}")

Source code in src/hother/streamblocks/extensions/gemini/__init__.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@InputAdapterRegistry.register(
    module_prefix="google.genai",
    attributes=["text", "candidates"],  # Fallback attribute detection
)
class GeminiInputAdapter:
    """Input adapter for Google GenAI streams.

    Handles chunks from google.genai.models.generate_content_stream()
    and google.ai.generativelanguage clients.

    Extracts:
    - Text from chunk.text attribute
    - Usage metadata (token counts)
    - Model version information

    Example:
        >>> from google import genai
        >>> adapter = GeminiInputAdapter()
        >>>
        >>> async for chunk in client.aio.models.generate_content_stream(...):
        ...     text = adapter.extract_text(chunk)
        ...     metadata = adapter.get_metadata(chunk)
        ...     if metadata:
        ...         print(f"Tokens: {metadata['usage']}")
    """

    native_module_prefix: ClassVar[str] = "google.genai"

    def categorize(self, event: GenerateContentResponse) -> EventCategory:
        """Categorize event - all Gemini chunks are text content.

        Args:
            event: Gemini GenerateContentResponse chunk

        Returns:
            TEXT_CONTENT for all chunks
        """
        return EventCategory.TEXT_CONTENT

    def extract_text(self, event: GenerateContentResponse) -> str | None:
        """Extract text from chunk.text attribute.

        Args:
            event: Gemini GenerateContentResponse chunk

        Returns:
            Text content, or None if not present
        """
        return event.text

    def is_complete(self, event: GenerateContentResponse) -> bool:
        """Gemini doesn't have explicit finish markers in each chunk.

        Completion is typically detected by the stream ending.

        Args:
            event: Gemini GenerateContentResponse chunk

        Returns:
            Always False - Gemini streams end naturally
        """
        return False

    def get_metadata(self, event: GenerateContentResponse) -> dict[str, Any] | None:
        """Extract usage metadata and model information.

        Args:
            event: Gemini GenerateContentResponse chunk

        Returns:
            Dictionary with usage and/or model if present
        """
        metadata: dict[str, Any] = {}

        # Extract usage metadata if available
        if event.usage_metadata:
            metadata["usage"] = event.usage_metadata

        # Extract model version
        if event.model_version:
            metadata["model"] = event.model_version

        return metadata if metadata else None

native_module_prefix class-attribute 

native_module_prefix: str = 'google.genai'

categorize 

categorize(event: GenerateContentResponse) -> EventCategory

Categorize event - all Gemini chunks are text content.

Parameters:

Name Type Description Default
event GenerateContentResponse

Gemini GenerateContentResponse chunk

 required

Returns:

Type Description
EventCategory

TEXT_CONTENT for all chunks
Source code in src/hother/streamblocks/extensions/gemini/__init__.py 
69
70
71
72
73
74
75
76
77
78
def categorize(self, event: GenerateContentResponse) -> EventCategory:
    """Categorize event - all Gemini chunks are text content.

    Args:
        event: Gemini GenerateContentResponse chunk

    Returns:
        TEXT_CONTENT for all chunks
    """
    return EventCategory.TEXT_CONTENT

extract_text 

extract_text(event: GenerateContentResponse) -> str | None

Extract text from chunk.text attribute.

Parameters:

Name Type Description Default
event GenerateContentResponse

Gemini GenerateContentResponse chunk

 required

Returns:

Type Description
str | None

Text content, or None if not present
Source code in src/hother/streamblocks/extensions/gemini/__init__.py 
80
81
82
83
84
85
86
87
88
89
def extract_text(self, event: GenerateContentResponse) -> str | None:
    """Extract text from chunk.text attribute.

    Args:
        event: Gemini GenerateContentResponse chunk

    Returns:
        Text content, or None if not present
    """
    return event.text

get_metadata 

get_metadata(
    event: GenerateContentResponse,
) -> dict[str, Any] | None

Extract usage metadata and model information.

Parameters:

Name Type Description Default
event GenerateContentResponse

Gemini GenerateContentResponse chunk

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary with usage and/or model if present
Source code in src/hother/streamblocks/extensions/gemini/__init__.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_metadata(self, event: GenerateContentResponse) -> dict[str, Any] | None:
    """Extract usage metadata and model information.

    Args:
        event: Gemini GenerateContentResponse chunk

    Returns:
        Dictionary with usage and/or model if present
    """
    metadata: dict[str, Any] = {}

    # Extract usage metadata if available
    if event.usage_metadata:
        metadata["usage"] = event.usage_metadata

    # Extract model version
    if event.model_version:
        metadata["model"] = event.model_version

    return metadata if metadata else None

is_complete 

is_complete(event: GenerateContentResponse) -> bool

Gemini doesn't have explicit finish markers in each chunk.

Completion is typically detected by the stream ending.

Parameters:

Name Type Description Default
event GenerateContentResponse

Gemini GenerateContentResponse chunk

 required

Returns:

Type Description
bool

Always False - Gemini streams end naturally
Source code in src/hother/streamblocks/extensions/gemini/__init__.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def is_complete(self, event: GenerateContentResponse) -> bool:
    """Gemini doesn't have explicit finish markers in each chunk.

    Completion is typically detected by the stream ending.

    Args:
        event: Gemini GenerateContentResponse chunk

    Returns:
        Always False - Gemini streams end naturally
    """
    return False

create_gemini_processor 

create_gemini_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]

Create processor pre-configured for Gemini streams.

This is a convenience factory that creates a ProtocolStreamProcessor with GeminiInputAdapter and StreamBlocksOutputAdapter.

Parameters:

Name Type Description Default
registry Registry

Registry with syntax and block definitions

 required

Returns:

Type Description
ProtocolStreamProcessor[Any, BaseEvent]

Pre-configured processor for Gemini streams
Example

from hother.streamblocks.extensions.gemini import create_gemini_processor processor = create_gemini_processor(registry) async for event in processor.process_stream(gemini_stream): ... if isinstance(event, BlockExtractedEvent): ... print(f"Block: {event.block.metadata.id}")

Source code in src/hother/streamblocks/extensions/gemini/__init__.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def create_gemini_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]:
    """Create processor pre-configured for Gemini streams.

    This is a convenience factory that creates a ProtocolStreamProcessor
    with GeminiInputAdapter and StreamBlocksOutputAdapter.

    Args:
        registry: Registry with syntax and block definitions

    Returns:
        Pre-configured processor for Gemini streams

    Example:
        >>> from hother.streamblocks.extensions.gemini import create_gemini_processor
        >>> processor = create_gemini_processor(registry)
        >>> async for event in processor.process_stream(gemini_stream):
        ...     if isinstance(event, BlockExtractedEvent):
        ...         print(f"Block: {event.block.metadata.id}")
    """
    from hother.streamblocks.adapters.output import StreamBlocksOutputAdapter
    from hother.streamblocks.core.protocol_processor import ProtocolStreamProcessor

    return ProtocolStreamProcessor(
        registry,
        input_adapter=GeminiInputAdapter(),
        output_adapter=StreamBlocksOutputAdapter(),
    )

OpenAI

hother.streamblocks.extensions.openai

OpenAI extension for StreamBlocks.

This extension provides input adapters for OpenAI ChatCompletionChunk streams.

Importing this module registers the OpenAIInputAdapter for auto-detection.

Example

Import to enable auto-detection

import hother.streamblocks.extensions.openai

Auto-detect from OpenAI stream

processor = ProtocolStreamProcessor(registry) async for event in processor.process_stream(openai_stream): ... print(event)

Or use convenience factory

from hother.streamblocks.extensions.openai import create_openai_processor processor = create_openai_processor(registry)

OpenAIInputAdapter

Input adapter for OpenAI ChatCompletionChunk streams.

Handles streams from openai.AsyncStream[ChatCompletionChunk].

Extracts: - Delta content from choices[0].delta.content - Finish reasons - Model information

Example

from openai import AsyncOpenAI adapter = OpenAIInputAdapter()

client = AsyncOpenAI() stream = await client.chat.completions.create( ... model="gpt-4", ... messages=[...], ... stream=True ... )

async for chunk in stream: ... text = adapter.extract_text(chunk) ... if adapter.is_complete(chunk): ... print("Stream complete!")

Source code in src/hother/streamblocks/extensions/openai/__init__.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@InputAdapterRegistry.register(module_prefix="openai.types")
class OpenAIInputAdapter:
    """Input adapter for OpenAI ChatCompletionChunk streams.

    Handles streams from openai.AsyncStream[ChatCompletionChunk].

    Extracts:
    - Delta content from choices[0].delta.content
    - Finish reasons
    - Model information

    Example:
        >>> from openai import AsyncOpenAI
        >>> adapter = OpenAIInputAdapter()
        >>>
        >>> client = AsyncOpenAI()
        >>> stream = await client.chat.completions.create(
        ...     model="gpt-4",
        ...     messages=[...],
        ...     stream=True
        ... )
        >>>
        >>> async for chunk in stream:
        ...     text = adapter.extract_text(chunk)
        ...     if adapter.is_complete(chunk):
        ...         print("Stream complete!")
    """

    native_module_prefix: ClassVar[str] = "openai.types"

    def categorize(self, event: ChatCompletionChunk) -> EventCategory:
        """Categorize event - all OpenAI chunks are text content.

        Args:
            event: OpenAI ChatCompletionChunk

        Returns:
            TEXT_CONTENT for all chunks
        """
        return EventCategory.TEXT_CONTENT

    def extract_text(self, event: ChatCompletionChunk) -> str | None:
        """Extract text from choices[0].delta.content.

        Args:
            event: OpenAI ChatCompletionChunk

        Returns:
            Delta content text, or None if not present
        """
        if event.choices:
            delta = event.choices[0].delta
            return delta.content if delta else None
        return None

    def is_complete(self, event: ChatCompletionChunk) -> bool:
        """Check if finish_reason is set.

        Args:
            event: OpenAI ChatCompletionChunk

        Returns:
            True if this is the final chunk
        """
        if event.choices:
            return event.choices[0].finish_reason is not None
        return False

    def get_metadata(self, event: ChatCompletionChunk) -> dict[str, Any] | None:
        """Extract model and finish reason.

        Args:
            event: OpenAI ChatCompletionChunk

        Returns:
            Dictionary with model and/or finish_reason if present
        """
        metadata: dict[str, Any] = {}

        # Extract model name
        if event.model:
            metadata["model"] = event.model

        # Extract finish reason if present
        if event.choices:
            choice = event.choices[0]
            if choice.finish_reason:
                metadata["finish_reason"] = choice.finish_reason

        return metadata if metadata else None

native_module_prefix class-attribute 

native_module_prefix: str = 'openai.types'

categorize 

categorize(event: ChatCompletionChunk) -> EventCategory

Categorize event - all OpenAI chunks are text content.

Parameters:

Name Type Description Default
event ChatCompletionChunk

OpenAI ChatCompletionChunk

 required

Returns:

Type Description
EventCategory

TEXT_CONTENT for all chunks
Source code in src/hother/streamblocks/extensions/openai/__init__.py 
71
72
73
74
75
76
77
78
79
80
def categorize(self, event: ChatCompletionChunk) -> EventCategory:
    """Categorize event - all OpenAI chunks are text content.

    Args:
        event: OpenAI ChatCompletionChunk

    Returns:
        TEXT_CONTENT for all chunks
    """
    return EventCategory.TEXT_CONTENT

extract_text 

extract_text(event: ChatCompletionChunk) -> str | None

Extract text from choices[0].delta.content.

Parameters:

Name Type Description Default
event ChatCompletionChunk

OpenAI ChatCompletionChunk

 required

Returns:

Type Description
str | None

Delta content text, or None if not present
Source code in src/hother/streamblocks/extensions/openai/__init__.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
def extract_text(self, event: ChatCompletionChunk) -> str | None:
    """Extract text from choices[0].delta.content.

    Args:
        event: OpenAI ChatCompletionChunk

    Returns:
        Delta content text, or None if not present
    """
    if event.choices:
        delta = event.choices[0].delta
        return delta.content if delta else None
    return None

get_metadata 

get_metadata(
    event: ChatCompletionChunk,
) -> dict[str, Any] | None

Extract model and finish reason.

Parameters:

Name Type Description Default
event ChatCompletionChunk

OpenAI ChatCompletionChunk

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary with model and/or finish_reason if present
Source code in src/hother/streamblocks/extensions/openai/__init__.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_metadata(self, event: ChatCompletionChunk) -> dict[str, Any] | None:
    """Extract model and finish reason.

    Args:
        event: OpenAI ChatCompletionChunk

    Returns:
        Dictionary with model and/or finish_reason if present
    """
    metadata: dict[str, Any] = {}

    # Extract model name
    if event.model:
        metadata["model"] = event.model

    # Extract finish reason if present
    if event.choices:
        choice = event.choices[0]
        if choice.finish_reason:
            metadata["finish_reason"] = choice.finish_reason

    return metadata if metadata else None

is_complete 

is_complete(event: ChatCompletionChunk) -> bool

Check if finish_reason is set.

Parameters:

Name Type Description Default
event ChatCompletionChunk

OpenAI ChatCompletionChunk

 required

Returns:

Type Description
bool

True if this is the final chunk
Source code in src/hother/streamblocks/extensions/openai/__init__.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def is_complete(self, event: ChatCompletionChunk) -> bool:
    """Check if finish_reason is set.

    Args:
        event: OpenAI ChatCompletionChunk

    Returns:
        True if this is the final chunk
    """
    if event.choices:
        return event.choices[0].finish_reason is not None
    return False

create_openai_processor 

create_openai_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]

Create processor pre-configured for OpenAI streams.

This is a convenience factory that creates a ProtocolStreamProcessor with OpenAIInputAdapter and StreamBlocksOutputAdapter.

Parameters:

Name Type Description Default
registry Registry

Registry with syntax and block definitions

 required

Returns:

Type Description
ProtocolStreamProcessor[Any, BaseEvent]

Pre-configured processor for OpenAI streams
Example

from hother.streamblocks.extensions.openai import create_openai_processor processor = create_openai_processor(registry) async for event in processor.process_stream(openai_stream): ... if isinstance(event, BlockExtractedEvent): ... print(f"Block: {event.block.metadata.id}")

Source code in src/hother/streamblocks/extensions/openai/__init__.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def create_openai_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]:
    """Create processor pre-configured for OpenAI streams.

    This is a convenience factory that creates a ProtocolStreamProcessor
    with OpenAIInputAdapter and StreamBlocksOutputAdapter.

    Args:
        registry: Registry with syntax and block definitions

    Returns:
        Pre-configured processor for OpenAI streams

    Example:
        >>> from hother.streamblocks.extensions.openai import create_openai_processor
        >>> processor = create_openai_processor(registry)
        >>> async for event in processor.process_stream(openai_stream):
        ...     if isinstance(event, BlockExtractedEvent):
        ...         print(f"Block: {event.block.metadata.id}")
    """
    from hother.streamblocks.adapters.output import StreamBlocksOutputAdapter
    from hother.streamblocks.core.protocol_processor import ProtocolStreamProcessor

    return ProtocolStreamProcessor(
        registry,
        input_adapter=OpenAIInputAdapter(),
        output_adapter=StreamBlocksOutputAdapter(),
    )

Anthropic

hother.streamblocks.extensions.anthropic

Anthropic extension for StreamBlocks.

This extension provides input adapters for Anthropic message streams.

Importing this module registers the AnthropicInputAdapter for auto-detection.

Example

Import to enable auto-detection

import hother.streamblocks.extensions.anthropic

Auto-detect from Anthropic stream

processor = ProtocolStreamProcessor(registry) async for event in processor.process_stream(anthropic_stream): ... print(event)

Or use convenience factory

from hother.streamblocks.extensions.anthropic import create_anthropic_processor processor = create_anthropic_processor(registry)

AnthropicEvent module-attribute 

AnthropicEvent = (
    ContentBlockDeltaEvent
    | MessageDeltaEvent
    | MessageStopEvent
)

AnthropicInputAdapter

Input adapter for Anthropic message streams.

Handles event-based streaming from anthropic.MessageStream.

Anthropic uses different event types: - content_block_delta: Contains text deltas (TEXT_CONTENT) - message_delta: Contains usage information (PASSTHROUGH) - message_stop: Signals stream completion (PASSTHROUGH) - Other events: PASSTHROUGH

Example

from anthropic import AsyncAnthropic adapter = AnthropicInputAdapter()

client = AsyncAnthropic() async with client.messages.stream(...) as stream: ... async for event in stream: ... text = adapter.extract_text(event) ... if text: ... print(text, end='', flush=True) ... if adapter.is_complete(event): ... print("\nDone!")

Source code in src/hother/streamblocks/extensions/anthropic/__init__.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@InputAdapterRegistry.register(module_prefix="anthropic.")
class AnthropicInputAdapter:
    """Input adapter for Anthropic message streams.

    Handles event-based streaming from anthropic.MessageStream.

    Anthropic uses different event types:
    - content_block_delta: Contains text deltas (TEXT_CONTENT)
    - message_delta: Contains usage information (PASSTHROUGH)
    - message_stop: Signals stream completion (PASSTHROUGH)
    - Other events: PASSTHROUGH

    Example:
        >>> from anthropic import AsyncAnthropic
        >>> adapter = AnthropicInputAdapter()
        >>>
        >>> client = AsyncAnthropic()
        >>> async with client.messages.stream(...) as stream:
        ...     async for event in stream:
        ...         text = adapter.extract_text(event)
        ...         if text:
        ...             print(text, end='', flush=True)
        ...         if adapter.is_complete(event):
        ...             print("\\nDone!")
    """

    native_module_prefix: ClassVar[str] = "anthropic."

    def categorize(self, event: AnthropicEvent) -> EventCategory:
        """Categorize event based on type.

        Args:
            event: Anthropic event

        Returns:
            TEXT_CONTENT for content_block_delta events, PASSTHROUGH for others
        """
        if isinstance(event, ContentBlockDeltaEvent):
            return EventCategory.TEXT_CONTENT
        # Other event types pass through
        return EventCategory.PASSTHROUGH

    def extract_text(self, event: AnthropicEvent) -> str | None:
        """Extract text from content_block_delta events.

        Args:
            event: Anthropic event

        Returns:
            Delta text if present, None otherwise
        """
        if isinstance(event, ContentBlockDeltaEvent):
            delta = event.delta
            if isinstance(delta, TextDelta):
                return delta.text
        return None

    def is_complete(self, event: AnthropicEvent) -> bool:
        """Check for message_stop event.

        Args:
            event: Anthropic event

        Returns:
            True if this is the message_stop event
        """
        return isinstance(event, MessageStopEvent)

    def get_metadata(self, event: AnthropicEvent) -> dict[str, Any] | None:
        """Extract stop reason and usage information.

        Args:
            event: Anthropic event

        Returns:
            Dictionary with stop_reason or usage if present
        """
        metadata: dict[str, Any] = {}

        if isinstance(event, MessageStopEvent):
            # MessageStopEvent doesn't have stop_reason, it's in MessageDeltaEvent
            return None

        if isinstance(event, MessageDeltaEvent):
            if event.usage:
                metadata["usage"] = event.usage
            return metadata if metadata else None

        return None

native_module_prefix class-attribute 

native_module_prefix: str = 'anthropic.'

categorize 

categorize(event: AnthropicEvent) -> EventCategory

Categorize event based on type.

Parameters:

Name Type Description Default
event AnthropicEvent

Anthropic event

 required

Returns:

Type Description
EventCategory

TEXT_CONTENT for content_block_delta events, PASSTHROUGH for others
Source code in src/hother/streamblocks/extensions/anthropic/__init__.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
def categorize(self, event: AnthropicEvent) -> EventCategory:
    """Categorize event based on type.

    Args:
        event: Anthropic event

    Returns:
        TEXT_CONTENT for content_block_delta events, PASSTHROUGH for others
    """
    if isinstance(event, ContentBlockDeltaEvent):
        return EventCategory.TEXT_CONTENT
    # Other event types pass through
    return EventCategory.PASSTHROUGH

extract_text 

extract_text(event: AnthropicEvent) -> str | None

Extract text from content_block_delta events.

Parameters:

Name Type Description Default
event AnthropicEvent

Anthropic event

 required

Returns:

Type Description
str | None

Delta text if present, None otherwise
Source code in src/hother/streamblocks/extensions/anthropic/__init__.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def extract_text(self, event: AnthropicEvent) -> str | None:
    """Extract text from content_block_delta events.

    Args:
        event: Anthropic event

    Returns:
        Delta text if present, None otherwise
    """
    if isinstance(event, ContentBlockDeltaEvent):
        delta = event.delta
        if isinstance(delta, TextDelta):
            return delta.text
    return None

get_metadata 

get_metadata(
    event: AnthropicEvent,
) -> dict[str, Any] | None

Extract stop reason and usage information.

Parameters:

Name Type Description Default
event AnthropicEvent

Anthropic event

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary with stop_reason or usage if present
Source code in src/hother/streamblocks/extensions/anthropic/__init__.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def get_metadata(self, event: AnthropicEvent) -> dict[str, Any] | None:
    """Extract stop reason and usage information.

    Args:
        event: Anthropic event

    Returns:
        Dictionary with stop_reason or usage if present
    """
    metadata: dict[str, Any] = {}

    if isinstance(event, MessageStopEvent):
        # MessageStopEvent doesn't have stop_reason, it's in MessageDeltaEvent
        return None

    if isinstance(event, MessageDeltaEvent):
        if event.usage:
            metadata["usage"] = event.usage
        return metadata if metadata else None

    return None

is_complete 

is_complete(event: AnthropicEvent) -> bool

Check for message_stop event.

Parameters:

Name Type Description Default
event AnthropicEvent

Anthropic event

 required

Returns:

Type Description
bool

True if this is the message_stop event
Source code in src/hother/streamblocks/extensions/anthropic/__init__.py 
111
112
113
114
115
116
117
118
119
120
def is_complete(self, event: AnthropicEvent) -> bool:
    """Check for message_stop event.

    Args:
        event: Anthropic event

    Returns:
        True if this is the message_stop event
    """
    return isinstance(event, MessageStopEvent)

create_anthropic_processor 

create_anthropic_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]

Create processor pre-configured for Anthropic streams.

This is a convenience factory that creates a ProtocolStreamProcessor with AnthropicInputAdapter and StreamBlocksOutputAdapter.

Parameters:

Name Type Description Default
registry Registry

Registry with syntax and block definitions

 required

Returns:

Type Description
ProtocolStreamProcessor[Any, BaseEvent]

Pre-configured processor for Anthropic streams
Example

from hother.streamblocks.extensions.anthropic import create_anthropic_processor processor = create_anthropic_processor(registry) async for event in processor.process_stream(anthropic_stream): ... if isinstance(event, BlockExtractedEvent): ... print(f"Block: {event.block.metadata.id}")

Source code in src/hother/streamblocks/extensions/anthropic/__init__.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def create_anthropic_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]:
    """Create processor pre-configured for Anthropic streams.

    This is a convenience factory that creates a ProtocolStreamProcessor
    with AnthropicInputAdapter and StreamBlocksOutputAdapter.

    Args:
        registry: Registry with syntax and block definitions

    Returns:
        Pre-configured processor for Anthropic streams

    Example:
        >>> from hother.streamblocks.extensions.anthropic import create_anthropic_processor
        >>> processor = create_anthropic_processor(registry)
        >>> async for event in processor.process_stream(anthropic_stream):
        ...     if isinstance(event, BlockExtractedEvent):
        ...         print(f"Block: {event.block.metadata.id}")
    """
    from hother.streamblocks.adapters.output import StreamBlocksOutputAdapter
    from hother.streamblocks.core.protocol_processor import ProtocolStreamProcessor

    return ProtocolStreamProcessor(
        registry,
        input_adapter=AnthropicInputAdapter(),
        output_adapter=StreamBlocksOutputAdapter(),
    )

AG-UI

hother.streamblocks.extensions.agui

AG-UI extension for StreamBlocks.

This extension provides bidirectional adapters for the AG-UI protocol.

AG-UI is an event-based protocol for agent-to-frontend communication with event types for lifecycle, text messages, tool calls, and state management.

Importing this module registers the AGUIInputAdapter for auto-detection.

Example

Import to enable auto-detection

import hother.streamblocks.extensions.agui

Auto-detect from AG-UI stream

processor = ProtocolStreamProcessor(registry) async for event in processor.process_stream(agui_stream): ... print(event)

Bidirectional: AG-UI in, AG-UI out

from hother.streamblocks.extensions.agui import ( ... create_agui_bidirectional_processor, ... AGUIEventFilter, ... ) processor = create_agui_bidirectional_processor( ... registry, ... event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS, ... ) async for event in processor.process_stream(agui_stream): ... # event is an AG-UI event (dict format) ... if event["type"] == "CUSTOM": ... handle_streamblocks_event(event) ... else: ... forward_to_frontend(event)

AGUIEventFilter

Bases: Flag

Configurable event filtering for AG-UI output adapter.

Use these flags to control which StreamBlocks events are emitted when using AGUIOutputAdapter.

Example

Only emit block-related events

filter = AGUIEventFilter.BLOCKS_ONLY

Emit blocks with progress updates

filter = AGUIEventFilter.BLOCKS_WITH_PROGRESS

Custom combination

filter = AGUIEventFilter.TEXT_DELTA | AGUIEventFilter.BLOCK_EXTRACTED

Source code in src/hother/streamblocks/extensions/agui/filters.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class AGUIEventFilter(Flag):
    """Configurable event filtering for AG-UI output adapter.

    Use these flags to control which StreamBlocks events are emitted
    when using AGUIOutputAdapter.

    Example:
        >>> # Only emit block-related events
        >>> filter = AGUIEventFilter.BLOCKS_ONLY
        >>>
        >>> # Emit blocks with progress updates
        >>> filter = AGUIEventFilter.BLOCKS_WITH_PROGRESS
        >>>
        >>> # Custom combination
        >>> filter = AGUIEventFilter.TEXT_DELTA | AGUIEventFilter.BLOCK_EXTRACTED
    """

    NONE = 0
    """Emit no StreamBlocks events."""

    RAW_TEXT = auto()
    """Emit RawTextEvent as TextMessageContentEvent."""

    TEXT_DELTA = auto()
    """Emit TextDeltaEvent as TextMessageContentEvent."""

    BLOCK_OPENED = auto()
    """Emit BlockOpenedEvent as CustomEvent."""

    BLOCK_DELTA = auto()
    """Emit section delta events (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent) as CustomEvent."""

    BLOCK_EXTRACTED = auto()
    """Emit BlockExtractedEvent as CustomEvent."""

    BLOCK_REJECTED = auto()
    """Emit BlockRejectedEvent as CustomEvent."""

    # Presets
    ALL = RAW_TEXT | TEXT_DELTA | BLOCK_OPENED | BLOCK_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit all StreamBlocks events."""

    BLOCKS_ONLY = BLOCK_OPENED | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit only block lifecycle events (opened, extracted, rejected)."""

    BLOCKS_WITH_PROGRESS = BLOCK_OPENED | BLOCK_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit block lifecycle events plus progress updates."""

    TEXT_AND_FINAL = TEXT_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit text streaming plus final block results."""

ALL class-attribute instance-attribute 

ALL = (
    RAW_TEXT
    | TEXT_DELTA
    | BLOCK_OPENED
    | BLOCK_DELTA
    | BLOCK_EXTRACTED
    | BLOCK_REJECTED
)

Emit all StreamBlocks events.

BLOCKS_ONLY class-attribute instance-attribute 

BLOCKS_ONLY = (
    BLOCK_OPENED | BLOCK_EXTRACTED | BLOCK_REJECTED
)

Emit only block lifecycle events (opened, extracted, rejected).

BLOCKS_WITH_PROGRESS class-attribute instance-attribute 

BLOCKS_WITH_PROGRESS = (
    BLOCK_OPENED
    | BLOCK_DELTA
    | BLOCK_EXTRACTED
    | BLOCK_REJECTED
)

Emit block lifecycle events plus progress updates.

BLOCK_DELTA class-attribute instance-attribute 

BLOCK_DELTA = auto()

Emit section delta events (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent) as CustomEvent.

BLOCK_EXTRACTED class-attribute instance-attribute 

BLOCK_EXTRACTED = auto()

Emit BlockExtractedEvent as CustomEvent.

BLOCK_OPENED class-attribute instance-attribute 

BLOCK_OPENED = auto()

Emit BlockOpenedEvent as CustomEvent.

BLOCK_REJECTED class-attribute instance-attribute 

BLOCK_REJECTED = auto()

Emit BlockRejectedEvent as CustomEvent.

NONE class-attribute instance-attribute 

NONE = 0

Emit no StreamBlocks events.

RAW_TEXT class-attribute instance-attribute 

RAW_TEXT = auto()

Emit RawTextEvent as TextMessageContentEvent.

TEXT_AND_FINAL class-attribute instance-attribute 

TEXT_AND_FINAL = (
    TEXT_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
)

Emit text streaming plus final block results.

TEXT_DELTA class-attribute instance-attribute 

TEXT_DELTA = auto()

Emit TextDeltaEvent as TextMessageContentEvent.

AGUIInputAdapter

Input adapter for AG-UI protocol events.

Handles event-based streaming from AG-UI protocol.

AG-UI Event Categories: - TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_CHUNK: TEXT_CONTENT (has text) - All other events: PASSTHROUGH (lifecycle, tool calls, state)

Example

adapter = AGUIInputAdapter()

async for event in agui_stream: ... category = adapter.categorize(event) ... if category == EventCategory.TEXT_CONTENT: ... text = adapter.extract_text(event) ... print(text, end='', flush=True)

Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@InputAdapterRegistry.register(module_prefix="ag_ui.")
class AGUIInputAdapter:
    """Input adapter for AG-UI protocol events.

    Handles event-based streaming from AG-UI protocol.

    AG-UI Event Categories:
    - TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_CHUNK: TEXT_CONTENT (has text)
    - All other events: PASSTHROUGH (lifecycle, tool calls, state)

    Example:
        >>> adapter = AGUIInputAdapter()
        >>>
        >>> async for event in agui_stream:
        ...     category = adapter.categorize(event)
        ...     if category == EventCategory.TEXT_CONTENT:
        ...         text = adapter.extract_text(event)
        ...         print(text, end='', flush=True)
    """

    native_module_prefix: ClassVar[str] = "ag_ui."

    def categorize(self, event: BaseEvent) -> EventCategory:
        """Categorize event based on type.

        Args:
            event: AG-UI BaseEvent

        Returns:
            TEXT_CONTENT for text message events, PASSTHROUGH for others
        """
        if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
            return EventCategory.TEXT_CONTENT

        # All other events (lifecycle, tool calls, state) pass through
        return EventCategory.PASSTHROUGH

    def extract_text(self, event: BaseEvent) -> str | None:
        """Extract text from TEXT_CONTENT events.

        Args:
            event: AG-UI BaseEvent

        Returns:
            Delta text if TEXT_MESSAGE_CONTENT or TEXT_MESSAGE_CHUNK
        """
        if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
            return event.delta

        return None

    def is_complete(self, event: BaseEvent) -> bool:
        """Check for RUN_FINISHED event.

        Args:
            event: AG-UI BaseEvent

        Returns:
            True if this is the RUN_FINISHED event
        """
        return isinstance(event, RunFinishedEvent)

    def get_metadata(self, event: BaseEvent) -> dict[str, Any] | None:
        """Extract protocol metadata.

        Args:
            event: AG-UI BaseEvent

        Returns:
            Dictionary with event_type and timestamp if available
        """
        # Get event type as string
        event_type_str = event.type.value if event.type else None

        metadata: dict[str, Any] = {"event_type": event_type_str}

        if event.timestamp is not None:
            metadata["timestamp"] = event.timestamp

        return metadata

native_module_prefix class-attribute 

native_module_prefix: str = 'ag_ui.'

categorize 

categorize(event: BaseEvent) -> EventCategory

Categorize event based on type.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
EventCategory

TEXT_CONTENT for text message events, PASSTHROUGH for others
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def categorize(self, event: BaseEvent) -> EventCategory:
    """Categorize event based on type.

    Args:
        event: AG-UI BaseEvent

    Returns:
        TEXT_CONTENT for text message events, PASSTHROUGH for others
    """
    if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
        return EventCategory.TEXT_CONTENT

    # All other events (lifecycle, tool calls, state) pass through
    return EventCategory.PASSTHROUGH

extract_text 

extract_text(event: BaseEvent) -> str | None

Extract text from TEXT_CONTENT events.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
str | None

Delta text if TEXT_MESSAGE_CONTENT or TEXT_MESSAGE_CHUNK
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def extract_text(self, event: BaseEvent) -> str | None:
    """Extract text from TEXT_CONTENT events.

    Args:
        event: AG-UI BaseEvent

    Returns:
        Delta text if TEXT_MESSAGE_CONTENT or TEXT_MESSAGE_CHUNK
    """
    if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
        return event.delta

    return None

get_metadata 

get_metadata(event: BaseEvent) -> dict[str, Any] | None

Extract protocol metadata.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary with event_type and timestamp if available
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def get_metadata(self, event: BaseEvent) -> dict[str, Any] | None:
    """Extract protocol metadata.

    Args:
        event: AG-UI BaseEvent

    Returns:
        Dictionary with event_type and timestamp if available
    """
    # Get event type as string
    event_type_str = event.type.value if event.type else None

    metadata: dict[str, Any] = {"event_type": event_type_str}

    if event.timestamp is not None:
        metadata["timestamp"] = event.timestamp

    return metadata

is_complete 

is_complete(event: BaseEvent) -> bool

Check for RUN_FINISHED event.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
bool

True if this is the RUN_FINISHED event
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
77
78
79
80
81
82
83
84
85
86
def is_complete(self, event: BaseEvent) -> bool:
    """Check for RUN_FINISHED event.

    Args:
        event: AG-UI BaseEvent

    Returns:
        True if this is the RUN_FINISHED event
    """
    return isinstance(event, RunFinishedEvent)

AGUIOutputAdapter

Output adapter for AG-UI protocol events.

Transforms StreamBlocks events into AG-UI CustomEvent format.

StreamBlocks events are mapped to AG-UI as follows: - TextDeltaEvent, TextContentEvent → TextMessageContentEvent - BlockStartEvent → CustomEvent(name="streamblocks.block_start") - BlockHeaderDeltaEvent → CustomEvent(name="streamblocks.block_delta") - BlockMetadataDeltaEvent → CustomEvent(name="streamblocks.block_delta") - BlockContentDeltaEvent → CustomEvent(name="streamblocks.block_delta") - BlockEndEvent → CustomEvent(name="streamblocks.block_end") - BlockErrorEvent → CustomEvent(name="streamblocks.block_error")

Passthrough events (AG-UI events from input) are passed through unchanged.

Example

adapter = AGUIOutputAdapter(event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS)

Convert StreamBlocks event to AG-UI event

agui_event = adapter.to_protocol_event(block_extracted_event)

Pass through AG-UI events

original = adapter.passthrough(run_started_event)

Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class AGUIOutputAdapter:
    """Output adapter for AG-UI protocol events.

    Transforms StreamBlocks events into AG-UI CustomEvent format.

    StreamBlocks events are mapped to AG-UI as follows:
    - TextDeltaEvent, TextContentEvent → TextMessageContentEvent
    - BlockStartEvent → CustomEvent(name="streamblocks.block_start")
    - BlockHeaderDeltaEvent → CustomEvent(name="streamblocks.block_delta")
    - BlockMetadataDeltaEvent → CustomEvent(name="streamblocks.block_delta")
    - BlockContentDeltaEvent → CustomEvent(name="streamblocks.block_delta")
    - BlockEndEvent → CustomEvent(name="streamblocks.block_end")
    - BlockErrorEvent → CustomEvent(name="streamblocks.block_error")

    Passthrough events (AG-UI events from input) are passed through unchanged.

    Example:
        >>> adapter = AGUIOutputAdapter(event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS)
        >>>
        >>> # Convert StreamBlocks event to AG-UI event
        >>> agui_event = adapter.to_protocol_event(block_extracted_event)
        >>>
        >>> # Pass through AG-UI events
        >>> original = adapter.passthrough(run_started_event)
    """

    def __init__(
        self,
        event_filter: AGUIEventFilter = AGUIEventFilter.ALL,
    ) -> None:
        """Initialize AG-UI output adapter.

        Args:
            event_filter: Filter to control which StreamBlocks events are emitted
        """
        self.event_filter = event_filter
        self._message_id: str | None = None

    def to_protocol_event(
        self,
        event: BaseEvent,
    ) -> dict[str, Any] | None:
        """Convert StreamBlocks event to AG-UI event format.

        Returns a dictionary representation of the AG-UI event that can be
        serialized or converted to the actual AG-UI event type.

        Args:
            event: StreamBlocks event

        Returns:
            Dictionary representing AG-UI event, or None if filtered out

        Note:
            Returns dict rather than actual AG-UI types to avoid requiring
            ag-ui-protocol as a runtime dependency.
        """
        from hother.streamblocks.core.types import (
            BlockContentDeltaEvent,
            BlockEndEvent,
            BlockErrorEvent,
            BlockHeaderDeltaEvent,
            BlockMetadataDeltaEvent,
            BlockStartEvent,
            TextContentEvent,
            TextDeltaEvent,
        )

        # Check filter first
        if not self._should_emit(event):
            return None

        if isinstance(event, TextDeltaEvent):
            return {
                "type": "TEXT_MESSAGE_CONTENT",
                "message_id": self._ensure_message_id(),
                "delta": event.delta,
            }

        if isinstance(event, TextContentEvent):
            return {
                "type": "TEXT_MESSAGE_CONTENT",
                "message_id": self._ensure_message_id(),
                "delta": event.content,
            }

        if isinstance(event, BlockStartEvent):
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_start",
                "value": {
                    "block_id": event.block_id,
                    "syntax": event.syntax,
                    "start_line": event.start_line,
                    "inline_metadata": event.inline_metadata,
                },
            }

        if isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
            section = event.type.value.replace("BLOCK_", "").replace("_DELTA", "").lower()
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_delta",
                "value": {
                    "block_id": event.block_id,
                    "syntax": event.syntax,
                    "section": section,
                    "delta": event.delta,
                    "current_line": event.current_line,
                },
            }

        if isinstance(event, BlockEndEvent):
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_end",
                "value": {
                    "block_id": event.block_id,
                    "block_type": event.block_type,
                    "syntax": event.syntax,
                    "metadata": event.metadata,
                    "content": event.content,
                    "start_line": event.start_line,
                    "end_line": event.end_line,
                    "hash_id": event.hash_id,
                },
            }

        if isinstance(event, BlockErrorEvent):
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_error",
                "value": {
                    "block_id": event.block_id,
                    "reason": event.reason,
                    "error_code": event.error_code,
                    "syntax": event.syntax,
                    "start_line": event.start_line,
                    "end_line": event.end_line,
                },
            }

        return None

    def passthrough(self, original_event: Any) -> Any:
        """Handle passthrough events.

        For AG-UI events, passes them through unchanged.

        Args:
            original_event: Original input event

        Returns:
            The original event unchanged
        """
        # If it's an AG-UI event (dict or object with type), pass through
        if isinstance(original_event, dict) and "type" in original_event:
            return original_event

        if isinstance(original_event, HasEventType):
            return original_event

        # For other events, wrap in a raw event format
        return {
            "type": "RAW",
            "event": original_event,
        }

    def _should_emit(
        self,
        event: BaseEvent,
    ) -> bool:
        """Check if event passes the filter.

        Args:
            event: StreamBlocks event

        Returns:
            True if event should be emitted
        """
        from hother.streamblocks.core.types import (
            BlockContentDeltaEvent,
            BlockEndEvent,
            BlockErrorEvent,
            BlockHeaderDeltaEvent,
            BlockMetadataDeltaEvent,
            BlockStartEvent,
            TextContentEvent,
            TextDeltaEvent,
        )

        if isinstance(event, TextContentEvent):
            return AGUIEventFilter.RAW_TEXT in self.event_filter

        if isinstance(event, TextDeltaEvent):
            return AGUIEventFilter.TEXT_DELTA in self.event_filter

        if isinstance(event, BlockStartEvent):
            return AGUIEventFilter.BLOCK_OPENED in self.event_filter

        if isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
            return AGUIEventFilter.BLOCK_DELTA in self.event_filter

        if isinstance(event, BlockEndEvent):
            return AGUIEventFilter.BLOCK_EXTRACTED in self.event_filter

        if isinstance(event, BlockErrorEvent):
            return AGUIEventFilter.BLOCK_REJECTED in self.event_filter

        return True

    def _ensure_message_id(self) -> str:
        """Generate or reuse message ID for text content events.

        Returns:
            Message ID string
        """
        if self._message_id is None:
            self._message_id = str(uuid.uuid4())
        return self._message_id

    def reset_message_id(self) -> None:
        """Reset message ID for new conversation turn."""
        self._message_id = None

event_filter instance-attribute 

event_filter = event_filter

passthrough 

passthrough(original_event: Any) -> Any

Handle passthrough events.

For AG-UI events, passes them through unchanged.

Parameters:

Name Type Description Default
original_event Any

Original input event

 required

Returns:

Type Description
Any

The original event unchanged
Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def passthrough(self, original_event: Any) -> Any:
    """Handle passthrough events.

    For AG-UI events, passes them through unchanged.

    Args:
        original_event: Original input event

    Returns:
        The original event unchanged
    """
    # If it's an AG-UI event (dict or object with type), pass through
    if isinstance(original_event, dict) and "type" in original_event:
        return original_event

    if isinstance(original_event, HasEventType):
        return original_event

    # For other events, wrap in a raw event format
    return {
        "type": "RAW",
        "event": original_event,
    }

reset_message_id 

reset_message_id() -> None

Reset message ID for new conversation turn.

Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
242
243
244
def reset_message_id(self) -> None:
    """Reset message ID for new conversation turn."""
    self._message_id = None

to_protocol_event 

to_protocol_event(
    event: BaseEvent,
) -> dict[str, Any] | None

Convert StreamBlocks event to AG-UI event format.

Returns a dictionary representation of the AG-UI event that can be serialized or converted to the actual AG-UI event type.

Parameters:

Name Type Description Default
event BaseEvent

StreamBlocks event

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary representing AG-UI event, or None if filtered out
Note

Returns dict rather than actual AG-UI types to avoid requiring ag-ui-protocol as a runtime dependency.

Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def to_protocol_event(
    self,
    event: BaseEvent,
) -> dict[str, Any] | None:
    """Convert StreamBlocks event to AG-UI event format.

    Returns a dictionary representation of the AG-UI event that can be
    serialized or converted to the actual AG-UI event type.

    Args:
        event: StreamBlocks event

    Returns:
        Dictionary representing AG-UI event, or None if filtered out

    Note:
        Returns dict rather than actual AG-UI types to avoid requiring
        ag-ui-protocol as a runtime dependency.
    """
    from hother.streamblocks.core.types import (
        BlockContentDeltaEvent,
        BlockEndEvent,
        BlockErrorEvent,
        BlockHeaderDeltaEvent,
        BlockMetadataDeltaEvent,
        BlockStartEvent,
        TextContentEvent,
        TextDeltaEvent,
    )

    # Check filter first
    if not self._should_emit(event):
        return None

    if isinstance(event, TextDeltaEvent):
        return {
            "type": "TEXT_MESSAGE_CONTENT",
            "message_id": self._ensure_message_id(),
            "delta": event.delta,
        }

    if isinstance(event, TextContentEvent):
        return {
            "type": "TEXT_MESSAGE_CONTENT",
            "message_id": self._ensure_message_id(),
            "delta": event.content,
        }

    if isinstance(event, BlockStartEvent):
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_start",
            "value": {
                "block_id": event.block_id,
                "syntax": event.syntax,
                "start_line": event.start_line,
                "inline_metadata": event.inline_metadata,
            },
        }

    if isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
        section = event.type.value.replace("BLOCK_", "").replace("_DELTA", "").lower()
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_delta",
            "value": {
                "block_id": event.block_id,
                "syntax": event.syntax,
                "section": section,
                "delta": event.delta,
                "current_line": event.current_line,
            },
        }

    if isinstance(event, BlockEndEvent):
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_end",
            "value": {
                "block_id": event.block_id,
                "block_type": event.block_type,
                "syntax": event.syntax,
                "metadata": event.metadata,
                "content": event.content,
                "start_line": event.start_line,
                "end_line": event.end_line,
                "hash_id": event.hash_id,
            },
        }

    if isinstance(event, BlockErrorEvent):
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_error",
            "value": {
                "block_id": event.block_id,
                "reason": event.reason,
                "error_code": event.error_code,
                "syntax": event.syntax,
                "start_line": event.start_line,
                "end_line": event.end_line,
            },
        }

    return None

create_agui_bidirectional_processor 

create_agui_bidirectional_processor(
    registry: Registry, event_filter: AGUIEventFilter = ALL
) -> ProtocolStreamProcessor[Any, dict[str, Any]]

Create processor for AG-UI → AG-UI (bidirectional).

This processor takes AG-UI events as input and emits AG-UI events as output. StreamBlocks events are converted to AG-UI CustomEvent format.

Parameters:

Name Type Description Default
registry Registry

Registry with syntax and block definitions

 required
event_filter AGUIEventFilter

Filter to control which StreamBlocks events are emitted

 ALL

Returns:

Type Description
ProtocolStreamProcessor[Any, dict[str, Any]]

Pre-configured processor for AG-UI input and output
Example

from hother.streamblocks.extensions.agui import ( ... create_agui_bidirectional_processor, ... AGUIEventFilter, ... ) processor = create_agui_bidirectional_processor( ... registry, ... event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS, ... ) async for event in processor.process_stream(agui_stream): ... if event["type"] == "CUSTOM": ... if event["name"] == "streamblocks.block_extracted": ... handle_block(event["value"]) ... else: ... # Passthrough: lifecycle, tool calls, state ... forward_to_frontend(event)

Source code in src/hother/streamblocks/extensions/agui/__init__.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def create_agui_bidirectional_processor(
    registry: Registry,
    event_filter: AGUIEventFilter = AGUIEventFilter.ALL,
) -> ProtocolStreamProcessor[Any, dict[str, Any]]:
    """Create processor for AG-UI → AG-UI (bidirectional).

    This processor takes AG-UI events as input and emits AG-UI events
    as output. StreamBlocks events are converted to AG-UI CustomEvent
    format.

    Args:
        registry: Registry with syntax and block definitions
        event_filter: Filter to control which StreamBlocks events are emitted

    Returns:
        Pre-configured processor for AG-UI input and output

    Example:
        >>> from hother.streamblocks.extensions.agui import (
        ...     create_agui_bidirectional_processor,
        ...     AGUIEventFilter,
        ... )
        >>> processor = create_agui_bidirectional_processor(
        ...     registry,
        ...     event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS,
        ... )
        >>> async for event in processor.process_stream(agui_stream):
        ...     if event["type"] == "CUSTOM":
        ...         if event["name"] == "streamblocks.block_extracted":
        ...             handle_block(event["value"])
        ...     else:
        ...         # Passthrough: lifecycle, tool calls, state
        ...         forward_to_frontend(event)
    """
    from hother.streamblocks.core.protocol_processor import ProtocolStreamProcessor

    return ProtocolStreamProcessor(
        registry,
        input_adapter=AGUIInputAdapter(),
        output_adapter=AGUIOutputAdapter(event_filter=event_filter),
    )

create_agui_processor 

create_agui_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]

Create processor for AG-UI → StreamBlocks (unidirectional).

This processor takes AG-UI events as input and emits native StreamBlocks events.

Parameters:

Name Type Description Default
registry Registry

Registry with syntax and block definitions

 required

Returns:

Type Description
ProtocolStreamProcessor[Any, BaseEvent]

Pre-configured processor for AG-UI input, StreamBlocks output
Example

from hother.streamblocks.extensions.agui import create_agui_processor processor = create_agui_processor(registry) async for event in processor.process_stream(agui_stream): ... if isinstance(event, BlockExtractedEvent): ... print(f"Block: {event.block.metadata.id}")

Source code in src/hother/streamblocks/extensions/agui/__init__.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def create_agui_processor(
    registry: Registry,
) -> ProtocolStreamProcessor[Any, BaseEvent]:
    """Create processor for AG-UI → StreamBlocks (unidirectional).

    This processor takes AG-UI events as input and emits native
    StreamBlocks events.

    Args:
        registry: Registry with syntax and block definitions

    Returns:
        Pre-configured processor for AG-UI input, StreamBlocks output

    Example:
        >>> from hother.streamblocks.extensions.agui import create_agui_processor
        >>> processor = create_agui_processor(registry)
        >>> async for event in processor.process_stream(agui_stream):
        ...     if isinstance(event, BlockExtractedEvent):
        ...         print(f"Block: {event.block.metadata.id}")
    """
    from hother.streamblocks.adapters.output import StreamBlocksOutputAdapter
    from hother.streamblocks.core.protocol_processor import ProtocolStreamProcessor

    return ProtocolStreamProcessor(
        registry,
        input_adapter=AGUIInputAdapter(),
        output_adapter=StreamBlocksOutputAdapter(),
    )

hother.streamblocks.extensions.agui.filters

Event filter for AG-UI output adapter.

AGUIEventFilter

Bases: Flag

Configurable event filtering for AG-UI output adapter.

Use these flags to control which StreamBlocks events are emitted when using AGUIOutputAdapter.

Example

Only emit block-related events

filter = AGUIEventFilter.BLOCKS_ONLY

Emit blocks with progress updates

filter = AGUIEventFilter.BLOCKS_WITH_PROGRESS

Custom combination

filter = AGUIEventFilter.TEXT_DELTA | AGUIEventFilter.BLOCK_EXTRACTED

Source code in src/hother/streamblocks/extensions/agui/filters.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class AGUIEventFilter(Flag):
    """Configurable event filtering for AG-UI output adapter.

    Use these flags to control which StreamBlocks events are emitted
    when using AGUIOutputAdapter.

    Example:
        >>> # Only emit block-related events
        >>> filter = AGUIEventFilter.BLOCKS_ONLY
        >>>
        >>> # Emit blocks with progress updates
        >>> filter = AGUIEventFilter.BLOCKS_WITH_PROGRESS
        >>>
        >>> # Custom combination
        >>> filter = AGUIEventFilter.TEXT_DELTA | AGUIEventFilter.BLOCK_EXTRACTED
    """

    NONE = 0
    """Emit no StreamBlocks events."""

    RAW_TEXT = auto()
    """Emit RawTextEvent as TextMessageContentEvent."""

    TEXT_DELTA = auto()
    """Emit TextDeltaEvent as TextMessageContentEvent."""

    BLOCK_OPENED = auto()
    """Emit BlockOpenedEvent as CustomEvent."""

    BLOCK_DELTA = auto()
    """Emit section delta events (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent) as CustomEvent."""

    BLOCK_EXTRACTED = auto()
    """Emit BlockExtractedEvent as CustomEvent."""

    BLOCK_REJECTED = auto()
    """Emit BlockRejectedEvent as CustomEvent."""

    # Presets
    ALL = RAW_TEXT | TEXT_DELTA | BLOCK_OPENED | BLOCK_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit all StreamBlocks events."""

    BLOCKS_ONLY = BLOCK_OPENED | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit only block lifecycle events (opened, extracted, rejected)."""

    BLOCKS_WITH_PROGRESS = BLOCK_OPENED | BLOCK_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit block lifecycle events plus progress updates."""

    TEXT_AND_FINAL = TEXT_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
    """Emit text streaming plus final block results."""

ALL class-attribute instance-attribute 

ALL = (
    RAW_TEXT
    | TEXT_DELTA
    | BLOCK_OPENED
    | BLOCK_DELTA
    | BLOCK_EXTRACTED
    | BLOCK_REJECTED
)

Emit all StreamBlocks events.

BLOCKS_ONLY class-attribute instance-attribute 

BLOCKS_ONLY = (
    BLOCK_OPENED | BLOCK_EXTRACTED | BLOCK_REJECTED
)

Emit only block lifecycle events (opened, extracted, rejected).

BLOCKS_WITH_PROGRESS class-attribute instance-attribute 

BLOCKS_WITH_PROGRESS = (
    BLOCK_OPENED
    | BLOCK_DELTA
    | BLOCK_EXTRACTED
    | BLOCK_REJECTED
)

Emit block lifecycle events plus progress updates.

BLOCK_DELTA class-attribute instance-attribute 

BLOCK_DELTA = auto()

Emit section delta events (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent) as CustomEvent.

BLOCK_EXTRACTED class-attribute instance-attribute 

BLOCK_EXTRACTED = auto()

Emit BlockExtractedEvent as CustomEvent.

BLOCK_OPENED class-attribute instance-attribute 

BLOCK_OPENED = auto()

Emit BlockOpenedEvent as CustomEvent.

BLOCK_REJECTED class-attribute instance-attribute 

BLOCK_REJECTED = auto()

Emit BlockRejectedEvent as CustomEvent.

NONE class-attribute instance-attribute 

NONE = 0

Emit no StreamBlocks events.

RAW_TEXT class-attribute instance-attribute 

RAW_TEXT = auto()

Emit RawTextEvent as TextMessageContentEvent.

TEXT_AND_FINAL class-attribute instance-attribute 

TEXT_AND_FINAL = (
    TEXT_DELTA | BLOCK_EXTRACTED | BLOCK_REJECTED
)

Emit text streaming plus final block results.

TEXT_DELTA class-attribute instance-attribute 

TEXT_DELTA = auto()

Emit TextDeltaEvent as TextMessageContentEvent.

hother.streamblocks.extensions.agui.input_adapter

AG-UI input adapter for StreamBlocks.

AGUITextEvent module-attribute 

AGUITextEvent = (
    TextMessageContentEvent | TextMessageChunkEvent
)

AGUIInputAdapter

Input adapter for AG-UI protocol events.

Handles event-based streaming from AG-UI protocol.

AG-UI Event Categories: - TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_CHUNK: TEXT_CONTENT (has text) - All other events: PASSTHROUGH (lifecycle, tool calls, state)

Example

adapter = AGUIInputAdapter()

async for event in agui_stream: ... category = adapter.categorize(event) ... if category == EventCategory.TEXT_CONTENT: ... text = adapter.extract_text(event) ... print(text, end='', flush=True)

Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@InputAdapterRegistry.register(module_prefix="ag_ui.")
class AGUIInputAdapter:
    """Input adapter for AG-UI protocol events.

    Handles event-based streaming from AG-UI protocol.

    AG-UI Event Categories:
    - TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_CHUNK: TEXT_CONTENT (has text)
    - All other events: PASSTHROUGH (lifecycle, tool calls, state)

    Example:
        >>> adapter = AGUIInputAdapter()
        >>>
        >>> async for event in agui_stream:
        ...     category = adapter.categorize(event)
        ...     if category == EventCategory.TEXT_CONTENT:
        ...         text = adapter.extract_text(event)
        ...         print(text, end='', flush=True)
    """

    native_module_prefix: ClassVar[str] = "ag_ui."

    def categorize(self, event: BaseEvent) -> EventCategory:
        """Categorize event based on type.

        Args:
            event: AG-UI BaseEvent

        Returns:
            TEXT_CONTENT for text message events, PASSTHROUGH for others
        """
        if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
            return EventCategory.TEXT_CONTENT

        # All other events (lifecycle, tool calls, state) pass through
        return EventCategory.PASSTHROUGH

    def extract_text(self, event: BaseEvent) -> str | None:
        """Extract text from TEXT_CONTENT events.

        Args:
            event: AG-UI BaseEvent

        Returns:
            Delta text if TEXT_MESSAGE_CONTENT or TEXT_MESSAGE_CHUNK
        """
        if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
            return event.delta

        return None

    def is_complete(self, event: BaseEvent) -> bool:
        """Check for RUN_FINISHED event.

        Args:
            event: AG-UI BaseEvent

        Returns:
            True if this is the RUN_FINISHED event
        """
        return isinstance(event, RunFinishedEvent)

    def get_metadata(self, event: BaseEvent) -> dict[str, Any] | None:
        """Extract protocol metadata.

        Args:
            event: AG-UI BaseEvent

        Returns:
            Dictionary with event_type and timestamp if available
        """
        # Get event type as string
        event_type_str = event.type.value if event.type else None

        metadata: dict[str, Any] = {"event_type": event_type_str}

        if event.timestamp is not None:
            metadata["timestamp"] = event.timestamp

        return metadata

native_module_prefix class-attribute 

native_module_prefix: str = 'ag_ui.'

categorize 

categorize(event: BaseEvent) -> EventCategory

Categorize event based on type.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
EventCategory

TEXT_CONTENT for text message events, PASSTHROUGH for others
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def categorize(self, event: BaseEvent) -> EventCategory:
    """Categorize event based on type.

    Args:
        event: AG-UI BaseEvent

    Returns:
        TEXT_CONTENT for text message events, PASSTHROUGH for others
    """
    if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
        return EventCategory.TEXT_CONTENT

    # All other events (lifecycle, tool calls, state) pass through
    return EventCategory.PASSTHROUGH

extract_text 

extract_text(event: BaseEvent) -> str | None

Extract text from TEXT_CONTENT events.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
str | None

Delta text if TEXT_MESSAGE_CONTENT or TEXT_MESSAGE_CHUNK
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def extract_text(self, event: BaseEvent) -> str | None:
    """Extract text from TEXT_CONTENT events.

    Args:
        event: AG-UI BaseEvent

    Returns:
        Delta text if TEXT_MESSAGE_CONTENT or TEXT_MESSAGE_CHUNK
    """
    if isinstance(event, (TextMessageContentEvent, TextMessageChunkEvent)):
        return event.delta

    return None

get_metadata 

get_metadata(event: BaseEvent) -> dict[str, Any] | None

Extract protocol metadata.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary with event_type and timestamp if available
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def get_metadata(self, event: BaseEvent) -> dict[str, Any] | None:
    """Extract protocol metadata.

    Args:
        event: AG-UI BaseEvent

    Returns:
        Dictionary with event_type and timestamp if available
    """
    # Get event type as string
    event_type_str = event.type.value if event.type else None

    metadata: dict[str, Any] = {"event_type": event_type_str}

    if event.timestamp is not None:
        metadata["timestamp"] = event.timestamp

    return metadata

is_complete 

is_complete(event: BaseEvent) -> bool

Check for RUN_FINISHED event.

Parameters:

Name Type Description Default
event BaseEvent

AG-UI BaseEvent

 required

Returns:

Type Description
bool

True if this is the RUN_FINISHED event
Source code in src/hother/streamblocks/extensions/agui/input_adapter.py 
77
78
79
80
81
82
83
84
85
86
def is_complete(self, event: BaseEvent) -> bool:
    """Check for RUN_FINISHED event.

    Args:
        event: AG-UI BaseEvent

    Returns:
        True if this is the RUN_FINISHED event
    """
    return isinstance(event, RunFinishedEvent)

hother.streamblocks.extensions.agui.output_adapter

AG-UI output adapter for StreamBlocks.

AGUIOutputAdapter

Output adapter for AG-UI protocol events.

Transforms StreamBlocks events into AG-UI CustomEvent format.

StreamBlocks events are mapped to AG-UI as follows: - TextDeltaEvent, TextContentEvent → TextMessageContentEvent - BlockStartEvent → CustomEvent(name="streamblocks.block_start") - BlockHeaderDeltaEvent → CustomEvent(name="streamblocks.block_delta") - BlockMetadataDeltaEvent → CustomEvent(name="streamblocks.block_delta") - BlockContentDeltaEvent → CustomEvent(name="streamblocks.block_delta") - BlockEndEvent → CustomEvent(name="streamblocks.block_end") - BlockErrorEvent → CustomEvent(name="streamblocks.block_error")

Passthrough events (AG-UI events from input) are passed through unchanged.

Example

adapter = AGUIOutputAdapter(event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS)

Convert StreamBlocks event to AG-UI event

agui_event = adapter.to_protocol_event(block_extracted_event)

Pass through AG-UI events

original = adapter.passthrough(run_started_event)

Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class AGUIOutputAdapter:
    """Output adapter for AG-UI protocol events.

    Transforms StreamBlocks events into AG-UI CustomEvent format.

    StreamBlocks events are mapped to AG-UI as follows:
    - TextDeltaEvent, TextContentEvent → TextMessageContentEvent
    - BlockStartEvent → CustomEvent(name="streamblocks.block_start")
    - BlockHeaderDeltaEvent → CustomEvent(name="streamblocks.block_delta")
    - BlockMetadataDeltaEvent → CustomEvent(name="streamblocks.block_delta")
    - BlockContentDeltaEvent → CustomEvent(name="streamblocks.block_delta")
    - BlockEndEvent → CustomEvent(name="streamblocks.block_end")
    - BlockErrorEvent → CustomEvent(name="streamblocks.block_error")

    Passthrough events (AG-UI events from input) are passed through unchanged.

    Example:
        >>> adapter = AGUIOutputAdapter(event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS)
        >>>
        >>> # Convert StreamBlocks event to AG-UI event
        >>> agui_event = adapter.to_protocol_event(block_extracted_event)
        >>>
        >>> # Pass through AG-UI events
        >>> original = adapter.passthrough(run_started_event)
    """

    def __init__(
        self,
        event_filter: AGUIEventFilter = AGUIEventFilter.ALL,
    ) -> None:
        """Initialize AG-UI output adapter.

        Args:
            event_filter: Filter to control which StreamBlocks events are emitted
        """
        self.event_filter = event_filter
        self._message_id: str | None = None

    def to_protocol_event(
        self,
        event: BaseEvent,
    ) -> dict[str, Any] | None:
        """Convert StreamBlocks event to AG-UI event format.

        Returns a dictionary representation of the AG-UI event that can be
        serialized or converted to the actual AG-UI event type.

        Args:
            event: StreamBlocks event

        Returns:
            Dictionary representing AG-UI event, or None if filtered out

        Note:
            Returns dict rather than actual AG-UI types to avoid requiring
            ag-ui-protocol as a runtime dependency.
        """
        from hother.streamblocks.core.types import (
            BlockContentDeltaEvent,
            BlockEndEvent,
            BlockErrorEvent,
            BlockHeaderDeltaEvent,
            BlockMetadataDeltaEvent,
            BlockStartEvent,
            TextContentEvent,
            TextDeltaEvent,
        )

        # Check filter first
        if not self._should_emit(event):
            return None

        if isinstance(event, TextDeltaEvent):
            return {
                "type": "TEXT_MESSAGE_CONTENT",
                "message_id": self._ensure_message_id(),
                "delta": event.delta,
            }

        if isinstance(event, TextContentEvent):
            return {
                "type": "TEXT_MESSAGE_CONTENT",
                "message_id": self._ensure_message_id(),
                "delta": event.content,
            }

        if isinstance(event, BlockStartEvent):
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_start",
                "value": {
                    "block_id": event.block_id,
                    "syntax": event.syntax,
                    "start_line": event.start_line,
                    "inline_metadata": event.inline_metadata,
                },
            }

        if isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
            section = event.type.value.replace("BLOCK_", "").replace("_DELTA", "").lower()
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_delta",
                "value": {
                    "block_id": event.block_id,
                    "syntax": event.syntax,
                    "section": section,
                    "delta": event.delta,
                    "current_line": event.current_line,
                },
            }

        if isinstance(event, BlockEndEvent):
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_end",
                "value": {
                    "block_id": event.block_id,
                    "block_type": event.block_type,
                    "syntax": event.syntax,
                    "metadata": event.metadata,
                    "content": event.content,
                    "start_line": event.start_line,
                    "end_line": event.end_line,
                    "hash_id": event.hash_id,
                },
            }

        if isinstance(event, BlockErrorEvent):
            return {
                "type": "CUSTOM",
                "name": "streamblocks.block_error",
                "value": {
                    "block_id": event.block_id,
                    "reason": event.reason,
                    "error_code": event.error_code,
                    "syntax": event.syntax,
                    "start_line": event.start_line,
                    "end_line": event.end_line,
                },
            }

        return None

    def passthrough(self, original_event: Any) -> Any:
        """Handle passthrough events.

        For AG-UI events, passes them through unchanged.

        Args:
            original_event: Original input event

        Returns:
            The original event unchanged
        """
        # If it's an AG-UI event (dict or object with type), pass through
        if isinstance(original_event, dict) and "type" in original_event:
            return original_event

        if isinstance(original_event, HasEventType):
            return original_event

        # For other events, wrap in a raw event format
        return {
            "type": "RAW",
            "event": original_event,
        }

    def _should_emit(
        self,
        event: BaseEvent,
    ) -> bool:
        """Check if event passes the filter.

        Args:
            event: StreamBlocks event

        Returns:
            True if event should be emitted
        """
        from hother.streamblocks.core.types import (
            BlockContentDeltaEvent,
            BlockEndEvent,
            BlockErrorEvent,
            BlockHeaderDeltaEvent,
            BlockMetadataDeltaEvent,
            BlockStartEvent,
            TextContentEvent,
            TextDeltaEvent,
        )

        if isinstance(event, TextContentEvent):
            return AGUIEventFilter.RAW_TEXT in self.event_filter

        if isinstance(event, TextDeltaEvent):
            return AGUIEventFilter.TEXT_DELTA in self.event_filter

        if isinstance(event, BlockStartEvent):
            return AGUIEventFilter.BLOCK_OPENED in self.event_filter

        if isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
            return AGUIEventFilter.BLOCK_DELTA in self.event_filter

        if isinstance(event, BlockEndEvent):
            return AGUIEventFilter.BLOCK_EXTRACTED in self.event_filter

        if isinstance(event, BlockErrorEvent):
            return AGUIEventFilter.BLOCK_REJECTED in self.event_filter

        return True

    def _ensure_message_id(self) -> str:
        """Generate or reuse message ID for text content events.

        Returns:
            Message ID string
        """
        if self._message_id is None:
            self._message_id = str(uuid.uuid4())
        return self._message_id

    def reset_message_id(self) -> None:
        """Reset message ID for new conversation turn."""
        self._message_id = None

event_filter instance-attribute 

event_filter = event_filter

passthrough 

passthrough(original_event: Any) -> Any

Handle passthrough events.

For AG-UI events, passes them through unchanged.

Parameters:

Name Type Description Default
original_event Any

Original input event

 required

Returns:

Type Description
Any

The original event unchanged
Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def passthrough(self, original_event: Any) -> Any:
    """Handle passthrough events.

    For AG-UI events, passes them through unchanged.

    Args:
        original_event: Original input event

    Returns:
        The original event unchanged
    """
    # If it's an AG-UI event (dict or object with type), pass through
    if isinstance(original_event, dict) and "type" in original_event:
        return original_event

    if isinstance(original_event, HasEventType):
        return original_event

    # For other events, wrap in a raw event format
    return {
        "type": "RAW",
        "event": original_event,
    }

reset_message_id 

reset_message_id() -> None

Reset message ID for new conversation turn.

Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
242
243
244
def reset_message_id(self) -> None:
    """Reset message ID for new conversation turn."""
    self._message_id = None

to_protocol_event 

to_protocol_event(
    event: BaseEvent,
) -> dict[str, Any] | None

Convert StreamBlocks event to AG-UI event format.

Returns a dictionary representation of the AG-UI event that can be serialized or converted to the actual AG-UI event type.

Parameters:

Name Type Description Default
event BaseEvent

StreamBlocks event

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary representing AG-UI event, or None if filtered out
Note

Returns dict rather than actual AG-UI types to avoid requiring ag-ui-protocol as a runtime dependency.

Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def to_protocol_event(
    self,
    event: BaseEvent,
) -> dict[str, Any] | None:
    """Convert StreamBlocks event to AG-UI event format.

    Returns a dictionary representation of the AG-UI event that can be
    serialized or converted to the actual AG-UI event type.

    Args:
        event: StreamBlocks event

    Returns:
        Dictionary representing AG-UI event, or None if filtered out

    Note:
        Returns dict rather than actual AG-UI types to avoid requiring
        ag-ui-protocol as a runtime dependency.
    """
    from hother.streamblocks.core.types import (
        BlockContentDeltaEvent,
        BlockEndEvent,
        BlockErrorEvent,
        BlockHeaderDeltaEvent,
        BlockMetadataDeltaEvent,
        BlockStartEvent,
        TextContentEvent,
        TextDeltaEvent,
    )

    # Check filter first
    if not self._should_emit(event):
        return None

    if isinstance(event, TextDeltaEvent):
        return {
            "type": "TEXT_MESSAGE_CONTENT",
            "message_id": self._ensure_message_id(),
            "delta": event.delta,
        }

    if isinstance(event, TextContentEvent):
        return {
            "type": "TEXT_MESSAGE_CONTENT",
            "message_id": self._ensure_message_id(),
            "delta": event.content,
        }

    if isinstance(event, BlockStartEvent):
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_start",
            "value": {
                "block_id": event.block_id,
                "syntax": event.syntax,
                "start_line": event.start_line,
                "inline_metadata": event.inline_metadata,
            },
        }

    if isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
        section = event.type.value.replace("BLOCK_", "").replace("_DELTA", "").lower()
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_delta",
            "value": {
                "block_id": event.block_id,
                "syntax": event.syntax,
                "section": section,
                "delta": event.delta,
                "current_line": event.current_line,
            },
        }

    if isinstance(event, BlockEndEvent):
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_end",
            "value": {
                "block_id": event.block_id,
                "block_type": event.block_type,
                "syntax": event.syntax,
                "metadata": event.metadata,
                "content": event.content,
                "start_line": event.start_line,
                "end_line": event.end_line,
                "hash_id": event.hash_id,
            },
        }

    if isinstance(event, BlockErrorEvent):
        return {
            "type": "CUSTOM",
            "name": "streamblocks.block_error",
            "value": {
                "block_id": event.block_id,
                "reason": event.reason,
                "error_code": event.error_code,
                "syntax": event.syntax,
                "start_line": event.start_line,
                "end_line": event.end_line,
            },
        }

    return None

HasEventType

Bases: Protocol

Protocol for events with a type attribute.

Source code in src/hother/streamblocks/extensions/agui/output_adapter.py 
14
15
16
17
18
@runtime_checkable
class HasEventType(Protocol):
    """Protocol for events with a type attribute."""

    type: Any

type instance-attribute 

type: Any