Streamblocks

Real-time extraction of structured blocks from AI text streams

Streamblocks is a Python library for detecting and extracting structured content blocks from streaming text. Extract semantic blocks as they stream—not after completion—enabling reactive AI agents and real-time processing.

• 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 Guide

• Provider Adapters

  ---

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

  Adapters Guide

• Type Safe

  ---

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

  Block Types

Quick Example

import asyncio
from streamblocks import StreamBlockProcessor, BlockRegistry, Syntax

async def main():
    # Create registry and processor
    registry = BlockRegistry()
    processor = StreamBlockProcessor(registry, syntax=Syntax.DELIMITER_PREAMBLE)

    # Simulate a text stream
    async def text_stream():
        chunks = [
            "Here's the file operations:\n",
            "!!file01:files_operations\n",
            "src/main.py:C\n",
            "src/utils.py:E\n",
            "!!end\n",
            "Done!"
        ]
        for chunk in chunks:
            yield chunk

    # Process and react to blocks in real-time
    async for event in processor.process_stream(text_stream()):
        if event.type.name == "BLOCK_EXTRACTED":
            print(f"Extracted: {event.block.metadata.id}")

asyncio.run(main())

Why Streamblocks?

Feature	Benefit
LLM Agnostic	Works with any text stream—no vendor lock-in
Real-time Processing	React to blocks as they stream, don't wait for completion
Type Safety	Pydantic models for metadata and content validation
Extensible	Custom syntaxes, adapters, and block types
Framework Compatible	Integrates with LangGraph, Pydantic AI, and others

Installation

uvpip

uv add streamblocks

pip install streamblocks

With provider support:

uvpip

uv add streamblocks[gemini]      # Google Gemini
uv add streamblocks[openai]      # OpenAI
uv add streamblocks[anthropic]   # Anthropic Claude
uv add streamblocks[all-providers]  # All providers

pip install streamblocks[gemini]
pip install streamblocks[openai]
pip install streamblocks[anthropic]
pip install streamblocks[all-providers]

Get Started View Examples