Skip to content

Adapters

Input/output adapter protocols, auto-detection, and event categories. See Adapters for the conceptual guide.

hother.streamblocks.adapters.protocols

Protocol definitions for bidirectional stream adapters.

TInput module-attribute 

TInput = TypeVar('TInput', contravariant=True)

TOutput module-attribute 

TOutput = TypeVar('TOutput', covariant=True)

BidirectionalAdapter

Bases: Protocol[TInput, TOutput]

Combined bidirectional adapter for full protocol transformation.

This is a convenience pattern - users can also use separate adapters.

Example

class MyBidirectionalAdapter: ... def init(self): ... self._input = MyInputAdapter() ... self._output = MyOutputAdapter() ... ... @property ... def input_adapter(self) -> MyInputAdapter: ... return self._input ... ... @property ... def output_adapter(self) -> MyOutputAdapter: ... return self._output

Source code in src/hother/streamblocks/adapters/protocols.py 
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
@runtime_checkable
class BidirectionalAdapter(Protocol[TInput, TOutput]):
    """Combined bidirectional adapter for full protocol transformation.

    This is a convenience pattern - users can also use separate adapters.

    Example:
        >>> class MyBidirectionalAdapter:
        ...     def __init__(self):
        ...         self._input = MyInputAdapter()
        ...         self._output = MyOutputAdapter()
        ...
        ...     @property
        ...     def input_adapter(self) -> MyInputAdapter:
        ...         return self._input
        ...
        ...     @property
        ...     def output_adapter(self) -> MyOutputAdapter:
        ...         return self._output
    """

    @property
    def input_adapter(self) -> InputProtocolAdapter[TInput]:
        """Input protocol adapter."""
        ...

    @property
    def output_adapter(self) -> OutputProtocolAdapter[TOutput]:
        """Output protocol adapter."""
        ...

input_adapter property 

input_adapter: InputProtocolAdapter[TInput]

Input protocol adapter.

output_adapter property 

output_adapter: OutputProtocolAdapter[TOutput]

Output protocol adapter.

HasNativeModulePrefix

Bases: Protocol

Protocol for adapters that can identify native events by module prefix.

Adapters implementing this protocol can help the processor determine which events are "native" to their protocol and should be passed through.

Example

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

Source code in src/hother/streamblocks/adapters/protocols.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@runtime_checkable
class HasNativeModulePrefix(Protocol):
    """Protocol for adapters that can identify native events by module prefix.

    Adapters implementing this protocol can help the processor determine
    which events are "native" to their protocol and should be passed through.

    Example:
        >>> class OpenAIInputAdapter:
        ...     native_module_prefix: ClassVar[str] = "openai.types"
        ...     ...
    """

    native_module_prefix: str

native_module_prefix instance-attribute 

native_module_prefix: str

InputProtocolAdapter

Bases: Protocol[TInput]

Protocol for transforming input events for StreamBlocks processing.

Handles any input format: chunks (OpenAI), events (AG-UI), or custom.

Example

class MyInputAdapter: ... def categorize(self, event: MyEvent) -> EventCategory: ... if event.has_text: ... return EventCategory.TEXT_CONTENT ... return EventCategory.PASSTHROUGH ... ... def extract_text(self, event: MyEvent) -> str | None: ... return event.text

Source code in src/hother/streamblocks/adapters/protocols.py 
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
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
@runtime_checkable
class InputProtocolAdapter(Protocol[TInput]):
    """Protocol for transforming input events for StreamBlocks processing.

    Handles any input format: chunks (OpenAI), events (AG-UI), or custom.

    Example:
        >>> class MyInputAdapter:
        ...     def categorize(self, event: MyEvent) -> EventCategory:
        ...         if event.has_text:
        ...             return EventCategory.TEXT_CONTENT
        ...         return EventCategory.PASSTHROUGH
        ...
        ...     def extract_text(self, event: MyEvent) -> str | None:
        ...         return event.text
    """

    def categorize(self, event: TInput) -> EventCategory:
        """Categorize event for routing.

        Args:
            event: Input event to categorize

        Returns:
            - TEXT_CONTENT: Event contains text, process with StreamBlocks
            - PASSTHROUGH: Pass through to output unchanged
            - SKIP: Don't include in output
        """
        ...

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

        Only called for events categorized as TEXT_CONTENT.

        This method can perform any complex extraction/computation:
        - Simple field access: return event.text
        - Nested extraction: return event.data.content.text
        - Multiple fields: return f"{event.prefix}{event.body}"
        - Decoding: return base64.decode(event.encoded_text)
        - JSON extraction: return json.loads(event.payload)["message"]

        Args:
            event: Input event to extract text from

        Returns:
            Extracted text, or None if no text available
        """
        ...

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

        Args:
            event: Input event to extract metadata from

        Returns:
            Dictionary of metadata, or None
        """
        return None

    def is_complete(self, event: TInput) -> bool:
        """Check if this event signals stream completion (optional).

        Args:
            event: Input event to check

        Returns:
            True if this is the final event in the stream
        """
        return False

categorize 

categorize(event: TInput) -> EventCategory

Categorize event for routing.

Parameters:

Name Type Description Default
event TInput

Input event to categorize

 required

Returns:

Type Description
EventCategory
  • TEXT_CONTENT: Event contains text, process with StreamBlocks
EventCategory
  • PASSTHROUGH: Pass through to output unchanged
EventCategory
  • SKIP: Don't include in output
Source code in src/hother/streamblocks/adapters/protocols.py 
32
33
34
35
36
37
38
39
40
41
42
43
def categorize(self, event: TInput) -> EventCategory:
    """Categorize event for routing.

    Args:
        event: Input event to categorize

    Returns:
        - TEXT_CONTENT: Event contains text, process with StreamBlocks
        - PASSTHROUGH: Pass through to output unchanged
        - SKIP: Don't include in output
    """
    ...

extract_text 

extract_text(event: TInput) -> str | None

Extract text content from TEXT_CONTENT events.

Only called for events categorized as TEXT_CONTENT.

This method can perform any complex extraction/computation: - Simple field access: return event.text - Nested extraction: return event.data.content.text - Multiple fields: return f"{event.prefix}{event.body}" - Decoding: return base64.decode(event.encoded_text) - JSON extraction: return json.loads(event.payload)["message"]

Parameters:

Name Type Description Default
event TInput

Input event to extract text from

 required

Returns:

Type Description
str | None

Extracted text, or None if no text available
Source code in src/hother/streamblocks/adapters/protocols.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def extract_text(self, event: TInput) -> str | None:
    """Extract text content from TEXT_CONTENT events.

    Only called for events categorized as TEXT_CONTENT.

    This method can perform any complex extraction/computation:
    - Simple field access: return event.text
    - Nested extraction: return event.data.content.text
    - Multiple fields: return f"{event.prefix}{event.body}"
    - Decoding: return base64.decode(event.encoded_text)
    - JSON extraction: return json.loads(event.payload)["message"]

    Args:
        event: Input event to extract text from

    Returns:
        Extracted text, or None if no text available
    """
    ...

get_metadata 

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

Extract protocol-specific metadata from event (optional).

Parameters:

Name Type Description Default
event TInput

Input event to extract metadata from

 required

Returns:

Type Description
dict[str, Any] | None

Dictionary of metadata, or None
Source code in src/hother/streamblocks/adapters/protocols.py 
65
66
67
68
69
70
71
72
73
74
def get_metadata(self, event: TInput) -> dict[str, Any] | None:
    """Extract protocol-specific metadata from event (optional).

    Args:
        event: Input event to extract metadata from

    Returns:
        Dictionary of metadata, or None
    """
    return None

is_complete 

is_complete(event: TInput) -> bool

Check if this event signals stream completion (optional).

Parameters:

Name Type Description Default
event TInput

Input event to check

 required

Returns:

Type Description
bool

True if this is the final event in the stream
Source code in src/hother/streamblocks/adapters/protocols.py 
76
77
78
79
80
81
82
83
84
85
def is_complete(self, event: TInput) -> bool:
    """Check if this event signals stream completion (optional).

    Args:
        event: Input event to check

    Returns:
        True if this is the final event in the stream
    """
    return False

OutputProtocolAdapter

Bases: Protocol[TOutput]

Protocol for transforming StreamBlocks events to output format.

Can emit to any protocol: AG-UI, custom events, plain text, etc.

Example

class MyOutputAdapter: ... def to_protocol_event(self, event: BaseEvent) -> MyEvent | None: ... if isinstance(event, BlockEndEvent): ... return MyEvent(type="block", data=event.get_block()) ... return None ... ... def passthrough(self, original_event: Any) -> MyEvent | None: ... return MyEvent(type="passthrough", data=original_event)

Source code in src/hother/streamblocks/adapters/protocols.py 
 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
@runtime_checkable
class OutputProtocolAdapter(Protocol[TOutput]):
    """Protocol for transforming StreamBlocks events to output format.

    Can emit to any protocol: AG-UI, custom events, plain text, etc.

    Example:
        >>> class MyOutputAdapter:
        ...     def to_protocol_event(self, event: BaseEvent) -> MyEvent | None:
        ...         if isinstance(event, BlockEndEvent):
        ...             return MyEvent(type="block", data=event.get_block())
        ...         return None
        ...
        ...     def passthrough(self, original_event: Any) -> MyEvent | None:
        ...         return MyEvent(type="passthrough", data=original_event)
    """

    def to_protocol_event(
        self,
        event: BaseEvent,
    ) -> TOutput | list[TOutput] | None:
        """Convert a StreamBlocks event to output protocol event(s).

        Args:
            event: StreamBlocks event to convert

        Returns:
            - Single event
            - List of events (for protocols requiring start/content/end pattern)
            - None to skip emission
        """
        ...

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

        Called for events categorized as PASSTHROUGH by the input adapter.

        Args:
            original_event: Original input event to pass through

        Returns:
            - The event in output protocol format
            - None if this adapter can't handle the event type
        """
        ...

passthrough 

passthrough(original_event: Any) -> TOutput | None

Handle passthrough events.

Called for events categorized as PASSTHROUGH by the input adapter.

Parameters:

Name Type Description Default
original_event Any

Original input event to pass through

 required

Returns:

Type Description
TOutput | None
  • The event in output protocol format
TOutput | None
  • None if this adapter can't handle the event type
Source code in src/hother/streamblocks/adapters/protocols.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def passthrough(
    self,
    original_event: Any,
) -> TOutput | None:
    """Handle passthrough events.

    Called for events categorized as PASSTHROUGH by the input adapter.

    Args:
        original_event: Original input event to pass through

    Returns:
        - The event in output protocol format
        - None if this adapter can't handle the event type
    """
    ...

to_protocol_event 

to_protocol_event(
    event: BaseEvent,
) -> TOutput | list[TOutput] | None

Convert a StreamBlocks event to output protocol event(s).

Parameters:

Name Type Description Default
event BaseEvent

StreamBlocks event to convert

 required

Returns:

Type Description
TOutput | list[TOutput] | None
  • Single event
TOutput | list[TOutput] | None
  • List of events (for protocols requiring start/content/end pattern)
TOutput | list[TOutput] | None
  • None to skip emission
Source code in src/hother/streamblocks/adapters/protocols.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def to_protocol_event(
    self,
    event: BaseEvent,
) -> TOutput | list[TOutput] | None:
    """Convert a StreamBlocks event to output protocol event(s).

    Args:
        event: StreamBlocks event to convert

    Returns:
        - Single event
        - List of events (for protocols requiring start/content/end pattern)
        - None to skip emission
    """
    ...

hother.streamblocks.adapters.detection

Automatic adapter detection for stream chunks.

HasContent

Bases: Protocol

Protocol for objects with a content attribute.

Source code in src/hother/streamblocks/adapters/detection.py 
22
23
24
25
26
@runtime_checkable
class HasContent(Protocol):
    """Protocol for objects with a content attribute."""

    content: str | None

content instance-attribute 

content: str | None

HasText

Bases: Protocol

Protocol for objects with a text attribute.

Source code in src/hother/streamblocks/adapters/detection.py 
15
16
17
18
19
@runtime_checkable
class HasText(Protocol):
    """Protocol for objects with a text attribute."""

    text: str | None

text instance-attribute 

text: str | None

InputAdapterRegistry

Registry for input adapter auto-detection.

Uses module prefix matching and attribute-based fallback detection. Extensions register themselves when imported.

Example

Register via decorator

@InputAdapterRegistry.register(module_prefix="openai.") ... class OpenAIInputAdapter: ... def categorize(self, event) -> EventCategory: ... return EventCategory.TEXT_CONTENT ... def extract_text(self, event) -> str | None: ... return event.choices[0].delta.content

Register via method

InputAdapterRegistry.register_module("mycompany.api", MyCustomAdapter)

Detect adapter from sample

adapter = InputAdapterRegistry.detect(sample_chunk)

Source code in src/hother/streamblocks/adapters/detection.py 
 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
class InputAdapterRegistry:
    """Registry for input adapter auto-detection.

    Uses module prefix matching and attribute-based fallback detection.
    Extensions register themselves when imported.

    Example:
        >>> # Register via decorator
        >>> @InputAdapterRegistry.register(module_prefix="openai.")
        ... class OpenAIInputAdapter:
        ...     def categorize(self, event) -> EventCategory:
        ...         return EventCategory.TEXT_CONTENT
        ...     def extract_text(self, event) -> str | None:
        ...         return event.choices[0].delta.content
        >>>
        >>> # Register via method
        >>> InputAdapterRegistry.register_module("mycompany.api", MyCustomAdapter)
        >>>
        >>> # Detect adapter from sample
        >>> adapter = InputAdapterRegistry.detect(sample_chunk)
    """

    # Module prefix → Adapter class (e.g., "openai." → OpenAIInputAdapter)
    _type_registry: ClassVar[dict[str, type[InputProtocolAdapter[Any]]]] = {}

    # Attribute patterns → Adapter class (for duck-typing detection)
    _pattern_registry: ClassVar[list[tuple[list[str], type[InputProtocolAdapter[Any]]]]] = []

    @classmethod
    def register(
        cls,
        *,
        module_prefix: str | None = None,
        attributes: list[str] | None = None,
    ) -> Callable[[type[InputProtocolAdapter[Any]]], type[InputProtocolAdapter[Any]]]:
        """Decorator to register an adapter for auto-detection.

        Args:
            module_prefix: Module path prefix to match (e.g., "openai.types")
            attributes: Required attributes for attribute-based detection

        Returns:
            Decorator function

        Example:
            >>> @InputAdapterRegistry.register(module_prefix="openai.")
            ... class OpenAIInputAdapter:
            ...     ...
            >>>
            >>> @InputAdapterRegistry.register(attributes=["text", "candidates"])
            ... class GeminiInputAdapter:
            ...     ...
        """

        def decorator(
            adapter_class: type[InputProtocolAdapter[Any]],
        ) -> type[InputProtocolAdapter[Any]]:
            if module_prefix:
                cls._type_registry[module_prefix] = adapter_class
            if attributes:
                cls._pattern_registry.insert(0, (attributes, adapter_class))
            return adapter_class

        return decorator

    @classmethod
    def register_module(
        cls,
        prefix: str,
        adapter_class: type[InputProtocolAdapter[Any]],
    ) -> None:
        """Register adapter by module prefix (non-decorator form).

        Args:
            prefix: Module path prefix to match
            adapter_class: Adapter class to instantiate when matched
        """
        cls._type_registry[prefix] = adapter_class

    @classmethod
    def register_pattern(
        cls,
        attrs: list[str],
        adapter_class: type[InputProtocolAdapter[Any]],
    ) -> None:
        """Register adapter by attribute pattern (non-decorator form).

        Args:
            attrs: Required attributes for detection
            adapter_class: Adapter class to instantiate when matched
        """
        cls._pattern_registry.insert(0, (attrs, adapter_class))

    @classmethod
    def detect(cls, chunk: Any) -> InputProtocolAdapter[Any] | None:
        """Detect and instantiate appropriate adapter from chunk.

        Detection order:
        1. String → IdentityInputAdapter
        2. Module prefix match
        3. Attribute pattern match
        4. Fallback to text/content attribute
        5. None if no match

        Args:
            chunk: Sample chunk from stream

        Returns:
            Adapter instance if detected, None otherwise
        """
        from hother.streamblocks.adapters.input import (
            AttributeInputAdapter,
            IdentityInputAdapter,
        )

        # Plain text
        if isinstance(chunk, str):
            return IdentityInputAdapter()

        # Module-based detection
        chunk_module = type(chunk).__module__
        for prefix, adapter_class in cls._type_registry.items():
            if chunk_module.startswith(prefix):
                return adapter_class()

        # Attribute-based detection
        for required_attrs, adapter_class in cls._pattern_registry:
            if all(hasattr(chunk, attr) for attr in required_attrs):
                return adapter_class()

        # Fallback: Protocol-based detection
        if isinstance(chunk, HasText):
            return AttributeInputAdapter("text")
        if isinstance(chunk, HasContent):
            return AttributeInputAdapter("content")

        return None

    @classmethod
    def get_registered_modules(cls) -> dict[str, type[InputProtocolAdapter[Any]]]:
        """Get all registered module prefixes.

        Returns:
            Copy of module prefix registry
        """
        return cls._type_registry.copy()

    @classmethod
    def get_registered_patterns(
        cls,
    ) -> list[tuple[list[str], type[InputProtocolAdapter[Any]]]]:
        """Get all registered attribute patterns.

        Returns:
            Copy of attribute pattern registry
        """
        return cls._pattern_registry.copy()

    @classmethod
    def clear(cls) -> None:
        """Clear all registered adapters (useful for testing)."""
        cls._type_registry.clear()
        cls._pattern_registry.clear()

clear classmethod 

clear() -> None

Clear all registered adapters (useful for testing).

Source code in src/hother/streamblocks/adapters/detection.py 
187
188
189
190
191
@classmethod
def clear(cls) -> None:
    """Clear all registered adapters (useful for testing)."""
    cls._type_registry.clear()
    cls._pattern_registry.clear()

detect classmethod 

detect(chunk: Any) -> InputProtocolAdapter[Any] | None

Detect and instantiate appropriate adapter from chunk.

Detection order: 1. String → IdentityInputAdapter 2. Module prefix match 3. Attribute pattern match 4. Fallback to text/content attribute 5. None if no match

Parameters:

Name Type Description Default
chunk Any

Sample chunk from stream

 required

Returns:

Type Description
InputProtocolAdapter[Any] | None

Adapter instance if detected, None otherwise
Source code in src/hother/streamblocks/adapters/detection.py 
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
@classmethod
def detect(cls, chunk: Any) -> InputProtocolAdapter[Any] | None:
    """Detect and instantiate appropriate adapter from chunk.

    Detection order:
    1. String → IdentityInputAdapter
    2. Module prefix match
    3. Attribute pattern match
    4. Fallback to text/content attribute
    5. None if no match

    Args:
        chunk: Sample chunk from stream

    Returns:
        Adapter instance if detected, None otherwise
    """
    from hother.streamblocks.adapters.input import (
        AttributeInputAdapter,
        IdentityInputAdapter,
    )

    # Plain text
    if isinstance(chunk, str):
        return IdentityInputAdapter()

    # Module-based detection
    chunk_module = type(chunk).__module__
    for prefix, adapter_class in cls._type_registry.items():
        if chunk_module.startswith(prefix):
            return adapter_class()

    # Attribute-based detection
    for required_attrs, adapter_class in cls._pattern_registry:
        if all(hasattr(chunk, attr) for attr in required_attrs):
            return adapter_class()

    # Fallback: Protocol-based detection
    if isinstance(chunk, HasText):
        return AttributeInputAdapter("text")
    if isinstance(chunk, HasContent):
        return AttributeInputAdapter("content")

    return None

get_registered_modules classmethod 

get_registered_modules() -> dict[
    str, type[InputProtocolAdapter[Any]]
]

Get all registered module prefixes.

Returns:

Type Description
dict[str, type[InputProtocolAdapter[Any]]]

Copy of module prefix registry
Source code in src/hother/streamblocks/adapters/detection.py 
167
168
169
170
171
172
173
174
@classmethod
def get_registered_modules(cls) -> dict[str, type[InputProtocolAdapter[Any]]]:
    """Get all registered module prefixes.

    Returns:
        Copy of module prefix registry
    """
    return cls._type_registry.copy()

get_registered_patterns classmethod 

get_registered_patterns() -> list[
    tuple[list[str], type[InputProtocolAdapter[Any]]]
]

Get all registered attribute patterns.

Returns:

Type Description
list[tuple[list[str], type[InputProtocolAdapter[Any]]]]

Copy of attribute pattern registry
Source code in src/hother/streamblocks/adapters/detection.py 
176
177
178
179
180
181
182
183
184
185
@classmethod
def get_registered_patterns(
    cls,
) -> list[tuple[list[str], type[InputProtocolAdapter[Any]]]]:
    """Get all registered attribute patterns.

    Returns:
        Copy of attribute pattern registry
    """
    return cls._pattern_registry.copy()

register classmethod 

register(
    *,
    module_prefix: str | None = None,
    attributes: list[str] | None = None,
) -> Callable[
    [type[InputProtocolAdapter[Any]]],
    type[InputProtocolAdapter[Any]],
]

Decorator to register an adapter for auto-detection.

Parameters:

Name Type Description Default
module_prefix str | None

Module path prefix to match (e.g., "openai.types")

 None
attributes list[str] | None

Required attributes for attribute-based detection

 None

Returns:

Type Description
Callable[[type[InputProtocolAdapter[Any]]], type[InputProtocolAdapter[Any]]]

Decorator function
Example

@InputAdapterRegistry.register(module_prefix="openai.") ... class OpenAIInputAdapter: ... ...

@InputAdapterRegistry.register(attributes=["text", "candidates"]) ... class GeminiInputAdapter: ... ...

Source code in src/hother/streamblocks/adapters/detection.py 
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
@classmethod
def register(
    cls,
    *,
    module_prefix: str | None = None,
    attributes: list[str] | None = None,
) -> Callable[[type[InputProtocolAdapter[Any]]], type[InputProtocolAdapter[Any]]]:
    """Decorator to register an adapter for auto-detection.

    Args:
        module_prefix: Module path prefix to match (e.g., "openai.types")
        attributes: Required attributes for attribute-based detection

    Returns:
        Decorator function

    Example:
        >>> @InputAdapterRegistry.register(module_prefix="openai.")
        ... class OpenAIInputAdapter:
        ...     ...
        >>>
        >>> @InputAdapterRegistry.register(attributes=["text", "candidates"])
        ... class GeminiInputAdapter:
        ...     ...
    """

    def decorator(
        adapter_class: type[InputProtocolAdapter[Any]],
    ) -> type[InputProtocolAdapter[Any]]:
        if module_prefix:
            cls._type_registry[module_prefix] = adapter_class
        if attributes:
            cls._pattern_registry.insert(0, (attributes, adapter_class))
        return adapter_class

    return decorator

register_module classmethod 

register_module(
    prefix: str,
    adapter_class: type[InputProtocolAdapter[Any]],
) -> None

Register adapter by module prefix (non-decorator form).

Parameters:

Name Type Description Default
prefix str

Module path prefix to match

 required
adapter_class type[InputProtocolAdapter[Any]]

Adapter class to instantiate when matched

 required
Source code in src/hother/streamblocks/adapters/detection.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@classmethod
def register_module(
    cls,
    prefix: str,
    adapter_class: type[InputProtocolAdapter[Any]],
) -> None:
    """Register adapter by module prefix (non-decorator form).

    Args:
        prefix: Module path prefix to match
        adapter_class: Adapter class to instantiate when matched
    """
    cls._type_registry[prefix] = adapter_class

register_pattern classmethod 

register_pattern(
    attrs: list[str],
    adapter_class: type[InputProtocolAdapter[Any]],
) -> None

Register adapter by attribute pattern (non-decorator form).

Parameters:

Name Type Description Default
attrs list[str]

Required attributes for detection

 required
adapter_class type[InputProtocolAdapter[Any]]

Adapter class to instantiate when matched

 required
Source code in src/hother/streamblocks/adapters/detection.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def register_pattern(
    cls,
    attrs: list[str],
    adapter_class: type[InputProtocolAdapter[Any]],
) -> None:
    """Register adapter by attribute pattern (non-decorator form).

    Args:
        attrs: Required attributes for detection
        adapter_class: Adapter class to instantiate when matched
    """
    cls._pattern_registry.insert(0, (attrs, adapter_class))

detect_input_adapter 

detect_input_adapter(
    sample: Any,
) -> InputProtocolAdapter[Any]

Detect input adapter from sample event.

Parameters:

Name Type Description Default
sample Any

A sample event from the stream

 required

Returns:

Type Description
InputProtocolAdapter[Any]

Detected adapter instance

Raises:

Type Description
AdapterDetectionError

If no adapter matches the sample
Source code in src/hother/streamblocks/adapters/detection.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def detect_input_adapter(sample: Any) -> InputProtocolAdapter[Any]:
    """Detect input adapter from sample event.

    Args:
        sample: A sample event from the stream

    Returns:
        Detected adapter instance

    Raises:
        AdapterDetectionError: If no adapter matches the sample
    """
    adapter = InputAdapterRegistry.detect(sample)
    if adapter is None:
        # sample is Any, so type(sample) is type[Unknown]; pin it to a known type.
        chunk_type = cast("type[object]", type(sample))
        registered_modules = tuple(InputAdapterRegistry.get_registered_modules().keys())
        raise AdapterDetectionError(
            chunk_type=f"{chunk_type.__module__}.{chunk_type.__name__}",
            registered_prefixes=registered_modules,
        )
    return adapter

hother.streamblocks.adapters.categories

Event categories for protocol adapter routing.

EventCategory

Bases: StrEnum

Semantic categorization of protocol events.

These three categories are EXHAUSTIVE - every protocol event falls into one:

  • TEXT_CONTENT: Event contains text that should be processed by StreamBlocks
  • PASSTHROUGH: Event should pass through unchanged to output
  • SKIP: Event should not be emitted in output at all
Source code in src/hother/streamblocks/adapters/categories.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class EventCategory(StrEnum):
    """Semantic categorization of protocol events.

    These three categories are EXHAUSTIVE - every protocol event falls into one:

    - TEXT_CONTENT: Event contains text that should be processed by StreamBlocks
    - PASSTHROUGH: Event should pass through unchanged to output
    - SKIP: Event should not be emitted in output at all
    """

    TEXT_CONTENT = "text_content"
    """Event contains text content that should be processed by StreamBlocks."""

    PASSTHROUGH = "passthrough"
    """Event has no text content and should pass through unchanged to output."""

    SKIP = "skip"
    """Event should not be included in output at all."""

PASSTHROUGH class-attribute instance-attribute 

PASSTHROUGH = 'passthrough'

Event has no text content and should pass through unchanged to output.

SKIP class-attribute instance-attribute 

SKIP = 'skip'

Event should not be included in output at all.

TEXT_CONTENT class-attribute instance-attribute 

TEXT_CONTENT = 'text_content'

Event contains text content that should be processed by StreamBlocks.