Quickstart Examples

Three ultra-minimal examples to get a feel for StreamBlocks in a few minutes. Each runs offline against a simulated stream; no API keys required. Continue with the basics examples afterwards.

Hello World

The simplest working example: register a block type, process a text stream, and react to BlockEndEvent when a complete block is extracted.

src/hother/streamblocks_examples/00_quickstart/01_hello_world.py

#!/usr/bin/env python3
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



async def main() -> None:
    """Extract a single block from a 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))




if __name__ == "__main__":
    asyncio.run(main())

View source on GitHub

Basic Stream

Processes text arriving in chunks, distinguishing extracted blocks from surrounding free text via TextContentEvent. This is the core streaming loop you will use everywhere.

src/hother/streamblocks_examples/00_quickstart/02_basic_stream.py

#!/usr/bin/env python3
import asyncio

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



async def main() -> None:
    """Process a chunked text stream."""
    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))




if __name__ == "__main__":
    asyncio.run(main())

View source on GitHub

Custom Block

Defines a custom block type with its own metadata and content models, then extracts it with DelimiterFrontmatterSyntax. The key takeaway: a block is just a pair of Pydantic models registered under a type name.

src/hother/streamblocks_examples/00_quickstart/03_custom_block.py

#!/usr/bin/env python3
import asyncio

from hother.streamblocks import DelimiterFrontmatterSyntax, Registry, StreamBlockProcessor
from hother.streamblocks.core.models import Block
from hother.streamblocks.core.types import BaseContent, BaseMetadata, BlockEndEvent
from hother.streamblocks_examples.helpers.simulator import simple_text_stream



class TaskMetadata(BaseMetadata):
    """Custom metadata for task blocks."""

    id: str
    block_type: str
    title: str = "Untitled"
    priority: str = "normal"


class TaskContent(BaseContent):
    """Custom content for task blocks."""

    description: str = ""

    @classmethod
    def parse(cls, raw_text: str) -> "TaskContent":
        return cls(raw_content=raw_text, description=raw_text.strip())


TaskBlock = Block[TaskMetadata, TaskContent]


async def main() -> None:
    """Use a custom block type."""
    registry = Registry(syntax=DelimiterFrontmatterSyntax())
    registry.register("task", TaskBlock)
    processor = StreamBlockProcessor(registry)

    text = "!!start\n---\nid: task-1\nblock_type: task\ntitle: Fix bug\npriority: high\n---\nFix the login issue\n!!end"
    stream = simple_text_stream(text)

    async for event in processor.process_stream(stream):
        if isinstance(event, BlockEndEvent):
            block = event.get_block()
            if block:
                print("\nExtracted Task Block:")
                print(block.model_dump_json(indent=2))




if __name__ == "__main__":
    asyncio.run(main())

View source on GitHub