Quickstart

Extract your first structured block from a text stream. A block in the default delimiter syntax looks like this:

!!block01:files_operations
src/main.py:C
!!end

The opening line carries the block id and block_type; everything until !!end is the content.

Hello, block

Three steps: create a Registry, register a block class for each block_type you expect, and feed any async text stream to the processor.

import asyncio

from hother.streamblocks import Registry, StreamBlockProcessor
from hother.streamblocks.core.types import BlockEndEvent
from hother.streamblocks_examples.blocks.agent.files import FileOperations
from hother.streamblocks_examples.helpers.simulator import simple_text_stream


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

# Text with one block
text = "!!block01:files_operations\nsrc/main.py:C\n!!end"
stream = simple_text_stream(text)

# Process and extract blocks
async for event in processor.process_stream(stream):
    if isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block:
            print("Extracted block:")
            print(block.model_dump_json(indent=2))

View source on GitHub

The processor emits an event for everything it sees. Here we only react to BlockEndEvent, emitted when a block closes successfully, and event.get_block() returns the parsed block with typed metadata and content.

Mixing text and blocks

Real streams interleave prose and blocks. Text outside blocks is emitted as TextContentEvent, so nothing is lost:

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

# Text to stream in chunks
text = "Some text before.\n!!block:files_operations\napp.py:C\n!!end\nSome text after."
stream = simulated_stream(text, preset="fast")

async for event in processor.process_stream(stream):
    if isinstance(event, TextContentEvent):
        print(f"[TEXT] {event.content.strip()}")
    elif isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block:
            print("\n[BLOCK] Extracted:")
            print(block.model_dump_json(indent=2))

View source on GitHub

What just happened?

1. Registry() defaults to the delimiter preamble syntax (!!<id>:<type> … !!end).
2. registry.register("files_operations", FileOperations) maps the block type string to a Block class with typed metadata and content models.
3. processor.process_stream(stream) consumes any AsyncIterator[str] and yields events as lines complete.
4. When the closing delimiter arrives, the block's content is parsed and validated, then BlockEndEvent delivers it.

Next steps

• Your First Custom Block: define your own metadata and content models.
• Events: the full event model, including per-section streaming deltas.
• Providers guide: plug in OpenAI, Anthropic, or Gemini streams instead of the simulator.