Skip to content

AG-UI Protocol

Consume and emit AG-UI protocol events with bidirectional adapters.

AG-UI is an event-based protocol for agent-to-frontend communication. The hother.streamblocks.extensions.agui extension lets you run an AG-UI event stream through StreamBlocks: text events are scanned for blocks, every other event (lifecycle, tool calls, state) passes through untouched.

The extension requires the ag-ui-protocol package:

uv add ag-ui-protocol

pip install ag-ui-protocol

import asyncio
from textwrap import dedent

from hother.streamblocks import DelimiterFrontmatterSyntax, Registry
from hother.streamblocks.core.types import BlockEndEvent
from hother.streamblocks_examples.blocks.agent.files import FileOperations

Choose a direction

Factory Input Output
create_agui_processor(registry) AG-UI events Native StreamBlocks events
create_agui_bidirectional_processor(registry, event_filter=...) AG-UI events AG-UI events (dicts)

Both factories return a pre-configured ProtocolStreamProcessor wired with AGUIInputAdapter on the input side.

Unidirectional: AG-UI in, StreamBlocks events out

Use this when your own code sits at the end of the pipeline and wants typed StreamBlocks events:

# Mode 1: Unidirectional - AG-UI in, StreamBlocks events out
print("=== Unidirectional (AG-UI → StreamBlocks) ===")
processor = create_agui_processor(registry)

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

async def agui_stream():
    yield RunStartedEvent(thread_id="thread-1", run_id="run-1")
    yield TextMessageContentEvent(delta=text, message_id="msg-1")
    yield RunFinishedEvent(thread_id="thread-1", run_id="run-1")

async for event in processor.process_stream(agui_stream()):
    if isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block:
            print("Block extracted:")
            print(block.model_dump_json(indent=2))

The input adapter extracts text from TEXT_MESSAGE_CONTENT and TEXT_MESSAGE_CHUNK events and treats RUN_FINISHED as end of stream.

Bidirectional: AG-UI in, AG-UI out

Use this when the output goes back to an AG-UI frontend. StreamBlocks events are converted to AG-UI CUSTOM events (emitted as dicts, so ag-ui-protocol is not needed at runtime on the consumer side), while original AG-UI events pass through unchanged:

AG-UI streamAGUIInputAdapterStreamBlockProcessorAGUIOutputAdapterAG-UI events TEXT_MESSAGE_CONTENTextracted text passthrough (non-text events)StreamBlocks eventsCUSTOM streamblocks.*
AG-UI streamAGUIInputAdapterStreamBlockProcessorAGUIOutputAdapterAG-UI events TEXT_MESSAGE_CONTENTextracted text passthrough (non-text events)StreamBlocks eventsCUSTOM streamblocks.* 
# Mode 2: Bidirectional - AG-UI in, AG-UI out
print("\n=== Bidirectional (AG-UI → AG-UI) ===")
bidir_processor = create_agui_bidirectional_processor(
    registry,
    event_filter=AGUIEventFilter.BLOCKS_ONLY,
)

async def agui_stream2():
    yield RunStartedEvent(thread_id="thread-2", run_id="run-2")
    yield TextMessageContentEvent(delta=text, message_id="msg-2")
    yield RunFinishedEvent(thread_id="thread-2", run_id="run-2")

async for event in bidir_processor.process_stream(agui_stream2()):
    if isinstance(event, dict):
        event_type = event.get("type", "unknown")
        if event_type == "CUSTOM":
            print(f"Custom event: {event.get('name')}")
        else:
            print(f"Passthrough: {event_type}")

View source on GitHub

The AGUIOutputAdapter maps StreamBlocks events to AG-UI as follows:

StreamBlocks event AG-UI event
TextDeltaEvent, TextContentEvent TEXT_MESSAGE_CONTENT
BlockStartEvent CUSTOM (streamblocks.block_start)
BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent CUSTOM (streamblocks.block_delta)
BlockEndEvent CUSTOM (streamblocks.block_end)
BlockErrorEvent CUSTOM (streamblocks.block_error)

Block payloads travel in the event's value field (block id, syntax, metadata, content, line range, and so on).

Filtering with AGUIEventFilter

AGUIEventFilter is a Flag enum controlling which StreamBlocks events the bidirectional processor emits. Presets:

Preset Emits
ALL (default) Every StreamBlocks event.
BLOCKS_ONLY Block lifecycle only: start, end, error.
BLOCKS_WITH_PROGRESS Block lifecycle plus per-section delta events.
TEXT_AND_FINAL Text deltas plus final block results (end, error).
NONE No StreamBlocks events; passthrough only.

Because it is a flag enum, you can also combine the fine-grained members (such as RAW_TEXT, TEXT_DELTA, and BLOCK_DELTA) with | to build a custom filter; see the extensions reference for the full list:

processor = create_agui_bidirectional_processor(
    registry,
    event_filter=AGUIEventFilter.BLOCKS_WITH_PROGRESS,
)

Filtering applies to StreamBlocks events only; passthrough AG-UI events are always forwarded.

Auto-detection

Importing hother.streamblocks.extensions.agui registers AGUIInputAdapter for adapter auto-detection (it claims events from the ag_ui. module namespace). A plain ProtocolStreamProcessor with auto_detect_adapter enabled will then recognize an AG-UI stream from its first event without explicit wiring.

Next steps