Basics Examples
Foundational examples covering the core StreamBlocks workflow: setting up a processor, handling events and errors, producing structured output, and validating blocks. All run offline with no API keys.
Basic Usage
A complete tour of the standard workflow: registry setup, stream processing, and handling the main event types including section delta events. Good reference for the full event vocabulary.
src/hother/streamblocks_examples/01_basics/01_basic_usage.py
View source on GitHub
import asyncio
from textwrap import dedent
from typing import TYPE_CHECKING
from hother.streamblocks import Registry, StreamBlockProcessor
from hother.streamblocks.core.models import ExtractedBlock
from hother.streamblocks.core.types import (
BlockContentDeltaEvent,
BlockEndEvent,
BlockErrorEvent,
BlockHeaderDeltaEvent,
BlockMetadataDeltaEvent,
TextContentEvent,
)
from hother.streamblocks_examples.blocks.agent.files import (
FileOperations,
FileOperationsContent,
FileOperationsMetadata,
)
from hother.streamblocks_examples.helpers.simulator import simulated_stream
if TYPE_CHECKING:
from hother.streamblocks.core.types import BaseContent, BaseMetadata
async def main() -> None:
"""Main example function."""
# Create type-specific registry and register block
registry = Registry()
# Add a custom validator
def no_root_delete(block: ExtractedBlock[FileOperationsMetadata, FileOperationsContent]) -> bool:
"""Don't allow deleting files from root directory."""
return all(not (op.action == "delete" and op.path.startswith("/")) for op in block.content.operations)
registry.register("files_operations", FileOperations, validators=[no_root_delete])
# Create processor with config
from hother.streamblocks.core.processor import ProcessorConfig
config = ProcessorConfig(lines_buffer=5)
processor = StreamBlockProcessor(registry, config=config)
# Example text with multiple blocks
text = dedent("""
This is some introductory text that will be passed through as raw text.
!!file01:files_operations
src/main.py:C
src/utils.py:C
tests/test_main.py:C
!!end
Here's some text between blocks.
!!file02:files_operations:urgent
config.yaml:C
README.md:C
old_file.py:D
!!end
And some final text after all blocks.
""")
# Process stream
print("Processing stream...")
print("-" * 60)
blocks_extracted: list[ExtractedBlock[BaseMetadata, BaseContent]] = []
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, TextContentEvent):
# Raw text passed through
if event.content.strip(): # Skip empty lines for cleaner output
print(f"[TEXT] {event.content.strip()}")
elif isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
# Partial block update
section = event.type.value.replace("BLOCK_", "").replace("_DELTA", "").lower()
print(f"[DELTA] {section} - {event.delta.strip()}")
elif isinstance(event, BlockEndEvent):
# Complete block extracted
block = event.get_block()
if block is not None:
blocks_extracted.append(block)
print("\n[BLOCK] Extracted:")
print(block.model_dump_json(indent=2))
elif isinstance(event, BlockErrorEvent):
# Block rejected
print(f"[REJECT] {event.reason} - {event.syntax}")
print("-" * 60)
print(f"\nTotal blocks extracted: {len(blocks_extracted)}")
# Show all extracted blocks
print("\nExtracted blocks (full details):")
for i, block in enumerate(blocks_extracted, 1):
print(f"\n--- Block {i} ---")
print(block.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())
Minimal API
Shows the smallest possible setup using default models; no custom metadata or content classes. Useful as a quick-reference template when you just need block extraction without typing.
src/hother/streamblocks_examples/01_basics/02_minimal_api.py
View source on GitHub
import asyncio
from textwrap import dedent
from typing import TYPE_CHECKING
from hother.streamblocks import Registry, StreamBlockProcessor
from hother.streamblocks.core.processor import ProcessorConfig
from hother.streamblocks.core.types import BlockEndEvent, BlockErrorEvent, TextContentEvent
from hother.streamblocks_examples.helpers.simulator import simulated_stream
if TYPE_CHECKING:
from hother.streamblocks.core.models import ExtractedBlock
from hother.streamblocks.core.types import BaseContent, BaseMetadata
async def main() -> None:
"""Main example function."""
# Create syntax with NO custom models - uses BaseMetadata and BaseContent
registry = Registry()
# Create processor with the registry
config = ProcessorConfig(lines_buffer=5)
processor = StreamBlockProcessor(registry, config=config)
# Example text with simple blocks
text = dedent("""
This is a document with some blocks using the minimal API.
!!note01:notes
This is a simple note block.
No custom models needed!
The library handles everything.
!!end
Some text between blocks.
!!todo01:tasks
- Buy groceries
- Call mom
- Finish the report
!!end
!!code01:snippets
def hello():
print("Hello, world!")
!!end
That's all folks!
""")
# Process stream
print("Processing with minimal API...")
print("-" * 60)
blocks_extracted: list[ExtractedBlock[BaseMetadata, BaseContent]] = []
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, TextContentEvent):
# Raw text passed through
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
elif isinstance(event, BlockEndEvent):
# Complete block extracted
block = event.get_block()
if block is not None:
blocks_extracted.append(block)
print("\n[BLOCK] Extracted:")
print(block.model_dump_json(indent=2))
elif isinstance(event, BlockErrorEvent):
# Block rejected
print(f"\n[REJECT] {event.reason}")
print("\n" + "-" * 60)
print(f"Total blocks extracted: {len(blocks_extracted)}")
# Show all extracted blocks
print("\nExtracted blocks (full details):")
for i, block in enumerate(blocks_extracted, 1):
print(f"\n--- Block {i} ---")
print(block.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())
Error Handling
Demonstrates structured error handling: accessing detailed error information from BlockRejectedEvent, including the original exception objects, so failures can be diagnosed and recovered from.
src/hother/streamblocks_examples/01_basics/03_error_handling.py
View source on GitHub
import asyncio
import logging
from textwrap import dedent
from typing import TYPE_CHECKING
import yaml
from pydantic import ValidationError
from hother.streamblocks import Registry, StreamBlockProcessor
from hother.streamblocks.core.types import BlockEndEvent, BlockErrorEvent, TextContentEvent
from hother.streamblocks.syntaxes.models import Syntax
from hother.streamblocks_examples.helpers.simulator import simulated_stream
if TYPE_CHECKING:
from hother.streamblocks.core.models import ExtractedBlock
from hother.streamblocks.core.types import BaseContent, BaseMetadata
async def main() -> None:
"""Demonstrate structured error handling."""
# Suppress library logging to stderr (we handle errors programmatically)
logging.getLogger("hother.streamblocks").setLevel(logging.CRITICAL)
# Setup registry with basic syntax - no custom block types needed
registry = Registry(
syntax=Syntax.DELIMITER_FRONTMATTER,
)
# Create processor
processor = StreamBlockProcessor(registry)
# Test stream with various error scenarios
test_stream = dedent("""
Some normal text.
!!start
---
id: valid_block
block_type: task
status: complete
---
This is a valid block with proper YAML metadata.
Everything should parse correctly.
!!end
!!start
---
id: broken_yaml
block_type: config
# Malformed YAML below - unclosed bracket
settings: [option1, option2
priority: high
---
This block has invalid YAML in the metadata section.
The YAML parser will fail with a ScannerError.
!!end
!!start
---
# Missing required 'id' and 'block_type' fields
# These are required by BaseMetadata
description: This will fail validation
---
Content for block with missing metadata fields.
!!end
Some more text at the end.
""").strip()
# Process stream and handle errors with structured information
extracted_blocks: list[ExtractedBlock[BaseMetadata, BaseContent]] = []
rejected_blocks: list[BlockErrorEvent[BaseMetadata, BaseContent]] = []
async for event in processor.process_stream(simulated_stream(test_stream)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is not None:
extracted_blocks.append(block)
print("\nEXTRACTED Block:")
print(block.model_dump_json(indent=2))
elif isinstance(event, BlockErrorEvent):
rejected_blocks.append(event)
print(f"\nREJECTED: Block at lines {event.start_line}-{event.end_line}")
print(f" Syntax: {event.syntax}")
print(f" Reason: {event.reason}")
# Access structured exception information
if event.exception:
print(f" Exception Type: {type(event.exception).__name__}")
# Handle different exception types differently
if isinstance(event.exception, yaml.YAMLError):
print(" → YAML parsing error detected")
# Use getattr for attributes not in type stubs
problem = getattr(event.exception, "problem", "Unknown YAML error")
print(f" → Problem: {problem}")
problem_mark = getattr(event.exception, "problem_mark", None)
if problem_mark is not None:
line = getattr(problem_mark, "line", -1)
column = getattr(problem_mark, "column", -1)
print(f" → Location: line {line + 1}, column {column + 1}")
elif isinstance(event.exception, ValidationError):
print(" → Pydantic validation error detected")
print(event.exception)
print(" → Missing/invalid fields:")
for error in event.exception.errors():
field = ".".join(str(loc) for loc in error["loc"])
msg = error["msg"]
print(f" • {field}: {msg}")
elif isinstance(event.exception, TypeError):
print(" → Type error detected")
print(f" → Details: {event.exception}")
else:
print(f" → Other error: {event.exception}")
# Show block_id if available
if event.block_id:
print(f" Block ID: {event.block_id}")
elif isinstance(event, TextContentEvent):
# Normal text outside blocks
text = event.content.strip()
if text:
print(f"📄 TEXT: {text}")
# Summary
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
print(f"Extracted blocks: {len(extracted_blocks)}")
print(f"Rejected blocks: {len(rejected_blocks)}")
# Detailed rejection analysis
if rejected_blocks:
print("\nRejection Analysis:")
yaml_errors = sum(1 for e in rejected_blocks if isinstance(e.exception, yaml.YAMLError))
validation_errors = sum(1 for e in rejected_blocks if isinstance(e.exception, ValidationError))
other_errors = len(rejected_blocks) - yaml_errors - validation_errors
print(f" - YAML parsing errors: {yaml_errors}")
print(f" - Validation errors: {validation_errors}")
print(f" - Other errors: {other_errors}")
if __name__ == "__main__":
asyncio.run(main())
Structured Output
Uses the create_structured_output_block factory to build type-safe blocks from any Pydantic model, turning free-form LLM output into validated structured data.
src/hother/streamblocks_examples/01_basics/04_structured_output.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from datetime import date
from enum import StrEnum
from textwrap import dedent
from typing import Any
from pydantic import BaseModel, Field
from hother.streamblocks import DelimiterFrontmatterSyntax, Registry, StreamBlockProcessor
from hother.streamblocks.core.types import (
BlockContentDeltaEvent,
BlockEndEvent,
BlockHeaderDeltaEvent,
BlockMetadataDeltaEvent,
TextContentEvent,
)
from hother.streamblocks_examples.blocks.agent.structured_output import create_structured_output_block
from hother.streamblocks_examples.helpers.simulator import simulated_stream
# ============================================================================
# EXAMPLE 1: Basic Person Schema
# ============================================================================
class PersonSchema(BaseModel):
"""Simple person data schema."""
name: str
age: int
email: str
city: str
async def example_1_basic_person() -> None:
"""Basic example with a simple person schema."""
# Create the specialized block type
PersonBlock = create_structured_output_block( # noqa: N806
schema_model=PersonSchema,
schema_name="person",
format="json",
strict=True, # Strict validation
)
# Create syntax and registry
# The syntax will extract metadata and content classes from the block automatically
registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("person", PersonBlock)
# Create processor
processor = StreamBlockProcessor(registry)
# Example text with person data
text = dedent("""
Here's a person's profile:
!!start
---
id: person_001
block_type: person
schema_name: person
format: json
description: User profile from registration
---
{
"name": "Alice Johnson",
"age": 28,
"email": "alice@example.com",
"city": "San Francisco"
}
!!end
That's the profile data.
""")
# Process the stream
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\nExtracted Person Block:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# ============================================================================
# EXAMPLE 2: Task List with Validation
# ============================================================================
class Priority(StrEnum):
"""Task priority levels."""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
URGENT = "urgent"
class TaskSchema(BaseModel):
"""Task with validation."""
title: str = Field(..., min_length=1, max_length=100)
description: str = ""
priority: Priority = Priority.MEDIUM
due_date: date | None = None
completed: bool = False
tags: list[str] = Field(default_factory=list)
async def example_2_task_list() -> None:
"""Task list example with validation."""
# Create the task block
TaskBlock = create_structured_output_block( # noqa: N806
schema_model=TaskSchema,
schema_name="task",
format="json",
strict=False, # Permissive - falls back to raw_content on errors
)
# Setup
registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("task", TaskBlock)
processor = StreamBlockProcessor(registry)
# Text with multiple tasks
text = dedent("""
Here are your tasks for today:
!!start
---
id: task_001
block_type: task
schema_name: task
---
{
"title": "Fix critical bug in payment system",
"description": "Users are reporting failed transactions",
"priority": "urgent",
"due_date": "2024-12-15",
"tags": ["bug", "payments", "urgent"]
}
!!end
!!start
---
id: task_002
block_type: task
schema_name: task
---
{
"title": "Update documentation",
"description": "Add examples for new API endpoints",
"priority": "low",
"tags": ["docs", "api"]
}
!!end
!!start
---
id: task_003
block_type: task
schema_name: task
---
{
"title": "Implement dark mode",
"priority": "medium",
"due_date": "2024-12-20",
"completed": false,
"tags": ["feature", "ui"]
}
!!end
All tasks loaded!
""")
# Process
tasks: list[Any] = []
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
tasks.append(block)
print("\n📋 Extracted Task:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# Summary
print(f"\n📊 Total tasks: {len(tasks)}")
# ============================================================================
# EXAMPLE 3: Nested Schema
# ============================================================================
class Address(BaseModel):
"""Address schema."""
street: str
city: str
state: str
zip_code: str
country: str = "USA"
class Company(BaseModel):
"""Company schema."""
name: str
industry: str
employee_count: int | None = None
class DetailedPersonSchema(BaseModel):
"""Person with nested data."""
name: str
age: int
email: str
address: Address
company: Company | None = None
skills: list[str] = Field(default_factory=list)
async def example_3_nested_schema() -> None:
"""Example with nested Pydantic models."""
# Create the block
DetailedPersonBlock = create_structured_output_block( # noqa: N806
schema_model=DetailedPersonSchema,
schema_name="detailed_person",
format="json",
strict=True,
)
# Setup
registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("detailed_person", DetailedPersonBlock)
processor = StreamBlockProcessor(registry)
# Employee profile text
text = dedent("""
Employee profile:
!!start
---
id: emp_001
block_type: detailed_person
schema_name: detailed_person
---
{
"name": "Bob Smith",
"age": 35,
"email": "bob@techcorp.com",
"address": {
"street": "123 Tech Avenue",
"city": "Seattle",
"state": "WA",
"zip_code": "98101",
"country": "USA"
},
"company": {
"name": "TechCorp Inc",
"industry": "Software",
"employee_count": 500
},
"skills": ["Python", "Rust", "Go", "Kubernetes"]
}
!!end
Profile loaded.
""")
# Process
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n👤 Extracted Profile:")
print(block.model_dump_json(indent=2))
# ============================================================================
# EXAMPLE 4: YAML Format
# ============================================================================
class ConfigSchema(BaseModel):
"""Configuration schema."""
app_name: str
version: str
debug: bool = False
features: dict[str, bool] = Field(default_factory=dict)
allowed_hosts: list[str] = Field(default_factory=list)
async def example_4_yaml_format() -> None:
"""Example using YAML format instead of JSON."""
# Create the block with YAML parsing
ConfigBlock = create_structured_output_block( # noqa: N806
schema_model=ConfigSchema,
schema_name="config",
format="yaml", # Using YAML!
strict=True,
)
# Setup
registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("config", ConfigBlock)
processor = StreamBlockProcessor(registry)
# Text with YAML content
text = dedent("""
Application configuration:
!!start
---
id: config_001
block_type: config
schema_name: config
format: yaml
---
app_name: MyAwesomeApp
version: 2.5.0
debug: true
features:
authentication: true
dark_mode: true
api_v2: false
allowed_hosts:
- localhost
- example.com
- "*.myapp.com"
!!end
Configuration loaded successfully.
""")
# Process
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n⚙️ Extracted Configuration:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# ============================================================================
# EXAMPLE 5: Simulated LLM Structured Output
# ============================================================================
class AnalysisResult(BaseModel):
"""Analysis result from an LLM."""
summary: str
sentiment: str # positive, negative, neutral
confidence: float = Field(..., ge=0.0, le=1.0)
key_points: list[str]
entities: dict[str, list[str]] = Field(default_factory=dict)
async def example_5_llm_simulation() -> None:
"""Simulate LLM generating structured output."""
# Create the block
AnalysisBlock = create_structured_output_block( # noqa: N806
schema_model=AnalysisResult,
schema_name="analysis",
format="json",
strict=True,
)
# Setup
registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("analysis", AnalysisBlock)
processor = StreamBlockProcessor(registry)
# Simulate streaming LLM response
text = dedent("""
Let me analyze this text for you.
I'll provide a structured analysis:
!!start
---
id: analysis_001
block_type: analysis
schema_name: analysis
description: Sentiment analysis of customer feedback
---
{
"summary": "Overall positive customer feedback with minor concerns about pricing.",
"sentiment": "positive",
"confidence": 0.85,
"key_points": [
"Customers love the user interface",
"Performance improvements are well received",
"Some concerns about subscription costs",
"Excellent customer support mentioned multiple times"
],
"entities": {
"products": ["mobile app", "web dashboard", "API"],
"features": ["dark mode", "real-time sync", "offline mode"],
"concerns": ["pricing", "storage limits"]
}
}
!!end
The analysis is complete. The data shows strong positive sentiment overall.
""")
# Process with real-time feedback
print("\n[Streaming LLM response...]")
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, TextContentEvent):
# Stream text as it arrives
if event.content.strip():
print(f"{event.content.strip()}")
elif isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
# Show progress while block is being accumulated
print(".", end="", flush=True)
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n")
print("\n📊 Analysis Results:")
print(block.model_dump_json(indent=2))
# ============================================================================
# Main
# ============================================================================
async def main() -> None:
"""Run all examples."""
await example_1_basic_person()
await example_2_task_list()
await example_3_nested_schema()
await example_4_yaml_format()
await example_5_llm_simulation()
if __name__ == "__main__":
asyncio.run(main())
Metadata Validators
Validates block metadata early, before content streams in, and shows the available MetadataValidationFailureMode options for deciding what happens when validation fails.
src/hother/streamblocks_examples/01_basics/05_metadata_validators.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from textwrap import dedent
from typing import Any
from hother.streamblocks import (
DelimiterFrontmatterSyntax,
MetadataValidationFailureMode,
Registry,
StreamBlockProcessor,
ValidationResult,
)
from hother.streamblocks.core.types import BlockEndEvent, BlockErrorEvent
from hother.streamblocks_examples.blocks.agent.files import FileOperations
from hother.streamblocks_examples.helpers.simulator import simple_text_stream
def validate_required_fields(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
"""Validate that required metadata fields are present."""
if not parsed:
return ValidationResult.failure("No metadata parsed")
if "id" not in parsed:
return ValidationResult.failure("Missing required field: id")
return ValidationResult.success()
def validate_id_format(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
"""Validate that ID follows naming convention."""
if not parsed:
return ValidationResult.success() # Let other validator handle this
block_id = parsed.get("id", "")
if not block_id.startswith("ops-"):
return ValidationResult.failure(f"ID must start with 'ops-', got: {block_id}")
return ValidationResult.success()
async def main() -> None:
"""Demonstrate metadata validation with different failure modes."""
syntax = DelimiterFrontmatterSyntax()
# Mode 1: ABORT_BLOCK (default) - stop processing on validation failure
registry = Registry(
syntax=syntax,
metadata_failure_mode=MetadataValidationFailureMode.ABORT_BLOCK,
)
registry.register("files_operations", FileOperations)
registry.add_metadata_validator("files_operations", validate_id_format)
processor = StreamBlockProcessor(registry)
# This block has an invalid ID (doesn't start with 'ops-')
text = dedent("""
!!start
---
id: invalid-id
block_type: files_operations
---
src/main.py:C
!!end
""").strip()
print("=== ABORT_BLOCK mode ===")
async for event in processor.process_stream(simple_text_stream(text)):
if isinstance(event, BlockErrorEvent):
print(f"Error: {event.reason}")
elif isinstance(event, BlockEndEvent):
print(f"Block extracted: {event.get_block()}")
# Valid block with correct ID format
valid_text = dedent("""
!!start
---
id: ops-001
block_type: files_operations
---
src/main.py:C
!!end
""").strip()
print("\n=== Valid block ===")
async for event in processor.process_stream(simple_text_stream(valid_text)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block:
print("Block extracted:")
print(block.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())
Validator Composition
Chains multiple validators (metadata, content, and general block validators) on a single block type, showing how validation responsibilities compose.
src/hother/streamblocks_examples/01_basics/06_validator_composition.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from textwrap import dedent
from typing import Any
from hother.streamblocks import (
DelimiterFrontmatterSyntax,
Registry,
StreamBlockProcessor,
ValidationResult,
)
from hother.streamblocks.core.models import ExtractedBlock
from hother.streamblocks.core.types import BlockEndEvent
from hother.streamblocks_examples.blocks.agent.files import FileOperations
def validate_has_description(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
"""First metadata validator: check for description."""
if parsed and "description" not in parsed:
return ValidationResult.failure("Missing 'description' in metadata")
return ValidationResult.success()
def validate_id_length(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
"""Second metadata validator: check ID length."""
min_id_length = 3
if parsed:
block_id = str(parsed.get("id", ""))
if len(block_id) < min_id_length:
return ValidationResult.failure(f"ID too short: {len(block_id)} < {min_id_length}")
return ValidationResult.success()
def validate_has_operations(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
"""Content validator: ensure there are operations."""
if not raw.strip():
return ValidationResult.failure("Content is empty")
return ValidationResult.success()
def validate_no_delete(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
"""Content validator: prevent delete operations."""
if ":D" in raw.upper():
return ValidationResult.failure("Delete operations not allowed")
return ValidationResult.success()
def validate_max_operations(block: ExtractedBlock[Any, Any]) -> bool:
"""General validator: limit number of operations."""
max_ops = 5
if hasattr(block.content, "operations"):
return len(block.content.operations) <= max_ops
return True
async def main() -> None:
"""Demonstrate validator chaining."""
syntax = DelimiterFrontmatterSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
# Register metadata validators (run in order, first failure stops)
registry.add_metadata_validator("files_operations", validate_has_description)
registry.add_metadata_validator("files_operations", validate_id_length)
# Register content validators (run in order, first failure stops)
registry.add_content_validator("files_operations", validate_has_operations)
registry.add_content_validator("files_operations", validate_no_delete)
# Register general validator (runs after full block extraction)
registry.add_validator("files_operations", validate_max_operations)
processor = StreamBlockProcessor(registry)
# Valid block that passes all validators
text = dedent("""
!!start
---
id: ops001
block_type: files_operations
description: Create new source files
---
src/main.py:C
src/utils.py:C
!!end
""").strip()
from hother.streamblocks_examples.helpers.simulator import simple_text_stream
print("=== Validator chain test ===")
async for event in processor.process_stream(simple_text_stream(text)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block:
print("Block passed all validators!")
print(block.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())
Prompt Generation
Builds an LLM system prompt from a registry with Registry.to_prompt() and generate_block_prompt() — the syntax format, block docstrings, content-format hints, and serialized examples — entirely offline. See the Generating Prompts guide.
src/hother/streamblocks_examples/01_basics/07_prompt_generation.py
View source on GitHub
from typing import ClassVar, Literal
from hother.streamblocks import (
BaseContent,
BaseMetadata,
Block,
DelimiterFrontmatterSyntax,
Registry,
generate_block_prompt,
parse_as_yaml,
)
class SearchMetadata(BaseMetadata):
"""Metadata for a catalog search block."""
block_type: Literal["search"] = "search"
@parse_as_yaml()
class SearchContent(BaseContent):
"""Search parameters."""
query: str = ""
limit: int = 10
class Search(Block[SearchMetadata, SearchContent]):
"""Search the product catalog.
Usage: emit a search block to look up products before answering.
"""
# Declared examples are serialized into the prompt so the model sees a
# concrete, correctly-formatted block.
__examples__: ClassVar[list[dict[str, object]]] = [
{
"metadata": {"id": "s1", "block_type": "search"},
"content": {"query": "wireless headphones", "limit": 5},
},
]
def main() -> None:
"""Build prompts from a registry and print them."""
registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("search", Search)
# A full system prompt documenting every registered block: the syntax
# format, the block description + "Usage:" line, its metadata fields, the
# YAML content format (from @parse_as_yaml), and serialized examples.
prompt = registry.to_prompt()
print(prompt)
# A prompt for a single block type -- here without the examples section.
single = generate_block_prompt(Search, registry.syntax, include_examples=False)
print(single)
# Register a custom template and select it with template_version. Templates
# receive `syntax_name`, `syntax_format`, and `blocks` in their context.
registry.register_template("compact", "Available blocks: {{ blocks | length }}", mode="registry")
print(registry.to_prompt(template_version="compact"))
if __name__ == "__main__":
main()