Skip to content

Adapters

LLM SDKs don't stream plain strings; they stream provider-specific event objects (OpenAI ChatCompletionChunk, Anthropic ContentBlockDeltaEvent, Gemini GenerateContentResponse, …). Adapters bridge that gap: an input adapter extracts text from incoming events, an output adapter converts StreamBlocks events into whatever format your application emits.

Plain text needs nothing

Strings are handled by the built-in IdentityInputAdapter, selected automatically; any AsyncIterator[str] just works:

# Setup
registry = Registry()
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)

# Process stream
print("Processing plain text stream...")
print()

async for event in processor.process_stream(plain_text_stream()):
    # Text deltas - emitted in real-time
    if isinstance(event, TextDeltaEvent):
        print(f"📝 Text Delta: {repr(event.delta)[:50]}", flush=True)

    # Raw text outside blocks
    elif isinstance(event, TextContentEvent):
        print(f"💬 Raw Text: {event.content}")

    # Extracted blocks
    elif isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block is None:
            continue
        print("\n✅ Block Extracted:")
        print(block.model_dump_json(indent=2))
        print()

View source on GitHub

The InputProtocolAdapter protocol

An input adapter is any object satisfying the InputProtocolAdapter protocol; no inheritance required:

Method Required Purpose
categorize(event) -> EventCategory yes Route the event: process its text, pass it through, or drop it
extract_text(event) -> str \| None yes Pull the text out of a TEXT_CONTENT event
get_metadata(event) -> dict \| None no Extract protocol-specific metadata (defaults to None)
is_complete(event) -> bool no Signal stream completion (defaults to False)

EventCategory is exhaustive; every event falls into one of three buckets:

Category Meaning
TEXT_CONTENT Event contains text; extract it and run block detection
PASSTHROUGH No text; forward the event unchanged to the output
SKIP Drop the event entirely

A minimal adapter for dict-shaped chunks:

# Simple inline adapter class
class DictInputAdapter:
    """Simple adapter for dict-based chunks."""

    def categorize(self, event: dict[str, Any]) -> EventCategory:
        """All dicts contain text content."""
        return EventCategory.TEXT_CONTENT

    def extract_text(self, chunk: dict[str, Any]) -> str | None:
        """Extract text from 'content' key."""
        return chunk.get("content")

    def is_complete(self, chunk: dict[str, Any]) -> bool:
        """Check 'done' flag for completion."""
        return chunk.get("done", False)

    def get_metadata(self, chunk: dict[str, Any]) -> dict[str, Any] | None:
        """Extract ID as metadata."""
        if "id" in chunk:
            return {"chunk_id": chunk.get("id")}
        return None

Pass it explicitly to bypass detection:

# Create adapter instance
adapter = DictInputAdapter()

# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)

print("Processing dict stream with inline adapter...")
print()

async for event in processor.process_stream(dict_stream(), adapter=adapter):
    # Original dicts
    if isinstance(event, dict):
        print(f"Dict Chunk: id={event['id']}, content={repr(event['content'])[:30]}")
        if event.get("done"):
            print("   Final chunk!")

    # Blocks
    elif isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block is None:
            continue
        print("\nBlock Extracted:")
        print(block.model_dump_json(indent=2))
        print()

View source on GitHub

For objects where the text simply lives in one attribute, skip the custom class and use the built-in AttributeInputAdapter:

# Create adapter for 'message' attribute
adapter = AttributeInputAdapter(text_attr="message")

# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)

Auto-detection

When no adapter is passed (and auto_detect_adapter=True, the default), InputAdapterRegistry.detect() inspects the first chunk and picks an adapter in this order:

First chunk arrivesIs it a str?Module prefix match?Attribute pattern match?Has text or content attr?IdentityInputAdapterRegistered adapter (OpenAI, Gemini, ...)Pattern-registered adapterAttributeInputAdapterNone: no adapter yesnoyesnoyesnoyesno
First chunk arrivesIs it a str?Module prefix match?Attribute pattern match?Has text or content attr?IdentityInputAdapterRegistered adapter (OpenAI, Gemini, ...)Pattern-registered adapterAttributeInputAdapterNone: no adapter yesnoyesnoyesnoyesno
  1. str → IdentityInputAdapter
  2. Module prefix match: the chunk class's __module__ is matched against registered prefixes (e.g. "openai.types", "google.genai", "anthropic.")
  3. Attribute pattern match: registered duck-typing patterns (e.g. ["text", "candidates"] for Gemini)
  4. Fallback: objects with a text or content attribute get an AttributeInputAdapter
  5. No match → None (detect_input_adapter() raises ValueError with the list of registered prefixes)

Provider extensions self-register on import; import hother.streamblocks.extensions.openai is enough to make auto-detection work for OpenAI streams.

Registering a custom adapter

Define the adapter for your format:

# Custom adapter implementing the InputProtocolAdapter protocol
class ProprietaryInputAdapter:
    """Input adapter for proprietary streaming format."""

    def categorize(self, event: ProprietaryChunk) -> EventCategory:
        """All events contain text content."""
        return EventCategory.TEXT_CONTENT

    def extract_text(self, chunk: ProprietaryChunk) -> str | None:
        """Extract text from payload field."""
        return chunk.payload

    def is_complete(self, chunk: ProprietaryChunk) -> bool:
        """Check if stream is complete."""
        return chunk.meta.get("final", False)

    def get_metadata(self, chunk: ProprietaryChunk) -> dict[str, Any] | None:
        """Extract metadata."""
        if chunk.meta:
            return {
                "request_id": chunk.meta.get("request_id"),
                "timestamp": chunk.meta.get("timestamp"),
            }
        return None

Then register it by module prefix so auto-detection picks it up:

# Register custom adapter for auto-detection
print("Registering custom adapter...")
InputAdapterRegistry.register_module(
    "mycompany.streaming",
    ProprietaryInputAdapter,
)
print("Registered for module: mycompany.streaming.*")

You can also register with the @InputAdapterRegistry.register(module_prefix=..., attributes=...) decorator, or InputAdapterRegistry.register_pattern() for attribute-based detection.

View source on GitHub

Output adapters

An OutputProtocolAdapter transforms StreamBlocks events into your target format:

Method Purpose
to_protocol_event(event) -> TOutput \| list[TOutput] \| None Convert a StreamBlocks event; return a list to fan out, None to filter
passthrough(original_event) -> TOutput \| None Handle events the input adapter categorized as PASSTHROUGH 
class JsonEventAdapter:
    """Output adapter that emits simplified JSON-like dicts.

    Features:
    - Return None to filter out events
    - Return a list to emit multiple events
    - Transform events to any output format
    """

    def to_protocol_event(self, event: BaseEvent) -> dict[str, Any] | list[dict[str, Any]] | None:
        """Convert StreamBlocks events to simplified dicts."""
        # Filter out text content events (we only care about blocks)
        if isinstance(event, TextContentEvent):
            return None

        # Emit block start and end as separate events
        if isinstance(event, BlockStartEvent):
            return {"event": "block_start", "block_id": event.block_id}

        if isinstance(event, BlockEndEvent):
            block = event.get_block()
            if block:
                # Emit multiple events: block info + operations
                events: list[dict[str, Any]] = [
                    {
                        "event": "block_end",
                        "block_id": block.metadata.id,
                        "block_type": block.metadata.block_type,
                    },
                ]
                # Add operation details
                if hasattr(block.content, "operations"):
                    for op in block.content.operations:
                        events.append(
                            {
                                "event": "operation",
                                "action": op.action,
                                "path": op.path,
                            }
                        )
                return events

        return None

    def passthrough(self, original_event: Any) -> dict[str, Any] | None:
        """Handle passthrough events."""
        return {"event": "passthrough", "data": str(original_event)}

View source on GitHub

The default is StreamBlocksOutputAdapter, which passes native StreamBlocks events through unchanged. The BidirectionalAdapter protocol bundles an input and an output adapter behind input_adapter / output_adapter properties when you want to ship both as one object.

Two processors

Processor Input Output Use when
StreamBlockProcessor Any stream (adapter auto-detected or passed via process_stream(stream, adapter=...)) Native StreamBlocks events, interleaved with original chunks if emit_original_events=True You consume StreamBlocks events directly, the common case
ProtocolStreamProcessor[TInput, TOutput] Any stream Your protocol's event type via an output adapter You translate between protocols end to end (e.g. AG-UI)

ProtocolStreamProcessor wires both directions together:

syntax = DelimiterFrontmatterSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)

processor = ProtocolStreamProcessor[CustomInputEvent, CustomOutputEvent](
    registry,
    input_adapter=CustomInputAdapter(),
    output_adapter=CustomOutputAdapter(),
)

PASSTHROUGH input events reach the output adapter's passthrough(); SKIP events vanish; text flows through block detection and comes out as transformed events:

text = dedent("""
    !!start
    ---
    id: ops001
    block_type: files_operations
    ---
    src/main.py:C
    !!end
""").strip()

async def input_stream():
    # Mix of different event types
    yield CustomInputEvent("metadata", "session-123")
    yield CustomInputEvent("text", text)
    yield CustomInputEvent("control", "ping")  # Will be skipped

print("=== Bidirectional Protocol ===")
async for output_event in processor.process_stream(input_stream()):
    print(f"[{output_event.kind}] {output_event.data}")

View source on GitHub

Next steps