Skip to content

StreamBlocks

StreamBlocks is a Python library that extracts structured, typed blocks from text streams while they stream, not after completion.

LLMs are good at embedding structured content (file operations, tool calls, plans) inside free-form text. StreamBlocks detects those blocks as the tokens arrive and hands you validated Pydantic objects in real time, enabling reactive agents and live UIs.

  • Stream Processing

    Extract blocks in real-time as text streams, enabling immediate reactions and feedback loops with LLMs.

    Getting Started

  • Multiple Syntaxes

    Choose from delimiter-based, Markdown frontmatter, or create custom syntaxes for your use case.

    Syntaxes

  • Provider Adapters

    Works with Gemini, OpenAI, Anthropic out of the box. Easy to add custom adapters.

    Adapters

  • Type Safe

    Full Pydantic model support with validation. Define your block metadata and content with type safety.

    Blocks & Registry

Quick Example

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))

The processor consumes any async text stream and emits events as it detects, accumulates, and completes blocks. When a block closes, BlockEndEvent.get_block() returns the fully parsed, validated, typed block.

Why StreamBlocks?

  • Any stream: plain strings or native OpenAI/Anthropic/Gemini chunks, no vendor lock-in.
  • Real time: react to a block the moment it completes, while the rest is still streaming.
  • Typed: metadata and content are Pydantic models, validated on extraction.
  • Extensible: custom syntaxes, adapters, block types, and framework integrations (Pydantic AI, AG-UI).

Installation

uv add streamblocks

pip install streamblocks

See Installation for provider extras.

Get Started View Examples