Adapters Examples
Adapters extract text from provider-specific stream chunks (Gemini, OpenAI, Anthropic, or your own format) and shape what events the processor emits. See the Adapters concept page for background. Most examples here run offline; the ones hitting real provider APIs are flagged with the required key.
Identity Adapter (Plain Text)
The default behavior: plain text streams need no adapter at all; chunks pass straight through to block detection.
src/hother/streamblocks_examples/03_adapters/01_identity_adapter_plain_text.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from collections.abc import AsyncGenerator
from hother.streamblocks import BlockEndEvent, Registry, StreamBlockProcessor, TextContentEvent, TextDeltaEvent
from hother.streamblocks_examples.blocks.agent.files import FileOperations
async def plain_text_stream() -> AsyncGenerator[str]:
"""Simulate a plain text stream."""
chunks = [
"Some text before the block\n",
"!!files01:files_operations\n",
"src/main.py:C\n",
"src/utils.py:C\n",
"!!end\n",
"Text after the block\n",
]
for chunk in chunks:
yield chunk
await asyncio.sleep(0.1) # Simulate streaming delay
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 01: Plain Text Stream (Default Behavior)")
print("=" * 60)
print()
# Setup
registry = Registry()
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
# Process stream
print("Processing plain text stream...")
print()
async for event in processor.process_stream(plain_text_stream()):
# Text deltas - emitted in real-time
if isinstance(event, TextDeltaEvent):
print(f"📝 Text Delta: {repr(event.delta)[:50]}", flush=True)
# Raw text outside blocks
elif isinstance(event, TextContentEvent):
print(f"💬 Raw Text: {event.content}")
# Extracted blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n✅ Block Extracted:")
print(block.model_dump_json(indent=2))
print()
print()
print("✓ Plain text streams work automatically!")
print("✓ No adapter configuration needed")
if __name__ == "__main__":
asyncio.run(main())
Gemini Auto-Detect
Requires GEMINI_API_KEY (or GOOGLE_API_KEY). StreamBlocks automatically detects Gemini chunks and extracts their text; no explicit adapter configuration needed.
src/hother/streamblocks_examples/03_adapters/02_gemini_auto_detect.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import os
import sys
# Check for Gemini SDK
try:
from google import genai
except ImportError:
print("Error: google-genai package not installed.")
print("Install it with: pip install streamblocks[gemini]")
print("Or: pip install google-genai")
sys.exit(1)
from hother.streamblocks import BlockEndEvent, Registry, StreamBlockProcessor, TextDeltaEvent
from hother.streamblocks_examples.blocks.agent.files import FileOperations
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 02: Gemini Auto-Detection")
print("=" * 60)
print()
# Get API key
api_key = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY")
if not api_key:
msg = (
"Please set GOOGLE_API_KEY or GEMINI_API_KEY environment variable.\n"
"Get your key at: https://aistudio.google.com/apikey"
)
raise ValueError(msg)
# Setup processor
registry = Registry()
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
# Create Gemini client
client = genai.Client(api_key=api_key) # type: ignore[attr-defined]
# Create prompt
prompt = """Create a simple project structure with these files:
- src/app.py
- README.md
Use this EXACT format (DO NOT use markdown code fences or any other formatting):
!!proj01:files_operations
src/app.py:C
README.md:C
!!end
IMPORTANT:
- Start your response directly with !! (no markdown, no code fences, no backticks)
- The first characters must be !!proj01:files_operations
- End with !!end on its own line
"""
try:
# Get stream from Gemini and pass directly to processor
response = await client.aio.models.generate_content_stream( # type: ignore[attr-defined]
model="gemini-2.5-flash",
contents=prompt,
)
async for event in processor.process_stream(response):
# Original Gemini chunks (passed through)
# Provider-agnostic detection using processor.is_native_event()
if processor.is_native_event(event):
text = getattr(event, "text", None)
if text:
print(f"🔵 Gemini Chunk: text={repr(text)[:40]}")
# Access Gemini-specific metadata
usage = getattr(event, "usage_metadata", None)
if usage:
total = getattr(usage, "total_token_count", None)
if total:
print(f" 📊 Total tokens: {total}")
# Real-time text deltas
elif isinstance(event, TextDeltaEvent):
print(f"📝 Text Delta: {repr(event.delta)[:40]}", end="")
if event.inside_block:
print(" (inside block)")
else:
print()
# Extracted blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n✅ Block Extracted:")
print(block.model_dump_json(indent=2))
print()
except ValueError as e:
print(f"\n❌ Error: {e}")
sys.exit(1)
except Exception as e:
print(f"\n❌ Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
OpenAI Explicit Adapter
Requires OPENAI_API_KEY. Configures an explicit adapter for OpenAI streams and shows how to access provider fields like finish_reason from the original chunks.
src/hother/streamblocks_examples/03_adapters/03_openai_explicit_adapter.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import os
import sys
# Check for OpenAI SDK
try:
from openai import AsyncOpenAI
except ImportError:
print("Error: openai package not installed.")
print("Install it with: pip install streamblocks[openai]")
print("Or: pip install openai")
sys.exit(1)
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextDeltaEvent,
)
from hother.streamblocks.extensions.openai import OpenAIInputAdapter
from hother.streamblocks_examples.blocks.agent.files import FileOperations
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 03: OpenAI with Explicit Adapter")
print("=" * 60)
print()
# Get API key
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
msg = "Please set OPENAI_API_KEY environment variable.\nGet your key at: https://platform.openai.com/api-keys"
raise ValueError(msg)
# Setup processor with explicit adapter
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
adapter = OpenAIInputAdapter()
# Create OpenAI client
client = AsyncOpenAI(api_key=api_key)
# Create prompt
prompt = """Create files for a simple app using this EXACT format (DO NOT use markdown code fences):
!!files01:files_operations
main.py:C
test.py:C
!!end
IMPORTANT:
- Start your response directly with !! (no markdown, no code fences)
- First characters must be !!files01:files_operations
- End with !!end on its own line
"""
print("Connecting to OpenAI API...")
print()
try:
# Get stream from OpenAI and pass directly to processor with explicit adapter
stream = await client.chat.completions.create(
model="gpt-5-nano-2025-08-07",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
async for event in processor.process_stream(stream, adapter=adapter):
# Original OpenAI chunks - provider-agnostic detection
if processor.is_native_event(event):
choices = getattr(event, "choices", [])
if choices:
choice = choices[0]
delta = getattr(choice, "delta", None)
if delta:
content = getattr(delta, "content", None)
if content:
print(f"🟢 OpenAI Chunk: {repr(content)[:40]}")
# Check for stream completion
finish_reason = getattr(choice, "finish_reason", None)
if finish_reason:
print(f"🏁 Stream Complete: {finish_reason}")
# Text deltas
elif isinstance(event, TextDeltaEvent):
print(f"📝 Delta: {repr(event.delta)[:40]}", flush=True)
# Extracted blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n✅ Block Extracted:")
print(block.model_dump_json(indent=2))
print()
except ValueError as e:
print(f"\n❌ Error: {e}")
sys.exit(1)
except Exception as e:
print(f"\n❌ Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
Anthropic Adapter
Requires ANTHROPIC_API_KEY. Handles Anthropic's event-based streaming format, where different event types (content_block_delta, message_stop, ...) are preserved alongside extraction.
src/hother/streamblocks_examples/03_adapters/04_anthropic_adapter.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import os
import sys
# Check for Anthropic SDK
try:
from anthropic import AsyncAnthropic
except ImportError:
print("Error: anthropic package not installed.")
print("Install it with: pip install streamblocks[anthropic]")
print("Or: pip install anthropic")
sys.exit(1)
# Enable auto-detection for Anthropic streams
import hother.streamblocks.extensions.anthropic
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextDeltaEvent,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 04: Anthropic Event Stream")
print("=" * 60)
print()
# Get API key
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
msg = (
"Please set ANTHROPIC_API_KEY environment variable.\n"
"Get your key at: https://console.anthropic.com/settings/keys"
)
raise ValueError(msg)
# Setup processor - auto-detection handles Anthropic streams
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
# Create Anthropic client
client = AsyncAnthropic(api_key=api_key)
# Create prompt
prompt = """Create a simple Python application with these files:
- app.py
- config.py
Use this EXACT format (DO NOT use markdown code fences):
!!files:files_operations
app.py:C
config.py:C
!!end
IMPORTANT:
- This block lists files to create (just the paths, no file content)
- Start your response directly with !! (no markdown, no code fences)
- Each line inside should be: filename:C (where C means Create)
- End with !!end on its own line
"""
print("Connecting to Anthropic API...")
print()
try:
# Get stream from Anthropic - auto-detection handles the format
async with client.messages.stream(
model="claude-sonnet-4-5-20250929",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
) as stream:
async for event in processor.process_stream(stream):
# Original Anthropic events - provider-agnostic detection
if processor.is_native_event(event):
event_type = getattr(event, "type", None)
if event_type:
print(f"🟣 Anthropic Event: type={event_type}")
if event_type == "message_stop":
stop_reason = getattr(event, "stop_reason", None)
if stop_reason:
print(f" 🛑 Stop reason: {stop_reason}")
# Text deltas
elif isinstance(event, TextDeltaEvent):
print(f"📝 Delta: {repr(event.delta)[:40]}")
# Blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n✅ Block Extracted:")
print(block.model_dump_json(indent=2))
print()
except ValueError as e:
print(f"\n❌ Error: {e}")
sys.exit(1)
except Exception as e:
print(f"\n❌ Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
Mixed Event Stream
Handles a stream containing both original provider chunks and StreamBlocks events, with isinstance-based type-checking patterns to route each kind.
src/hother/streamblocks_examples/03_adapters/05_mixed_event_stream.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from collections.abc import AsyncGenerator
from hother.streamblocks import (
BlockEndEvent,
BlockStartEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextContentEvent,
TextDeltaEvent,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
# Custom chunk type
class MyCustomChunk:
"""Custom chunk with metadata."""
def __init__(self, text: str, metadata: dict[str, str | int] | None = None) -> None:
self.text = text
self.metadata: dict[str, str | int] = metadata or {}
async def custom_stream() -> AsyncGenerator[MyCustomChunk]:
"""Stream with custom chunks."""
chunks = [
MyCustomChunk("!!files:files_operations\n", {"source": "api"}),
MyCustomChunk("main.py:C\n", {"line": 1}),
MyCustomChunk("test.py:C\n", {"line": 2}),
MyCustomChunk("!!end\n", {"source": "api"}),
]
for chunk in chunks:
yield chunk
await asyncio.sleep(0.05)
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 05: Mixed Event Streams")
print("=" * 60)
print()
# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Processing stream - showing both original and StreamBlocks events:")
print()
# Counters
original_count = 0
streamblocks_count = 0
async for event in processor.process_stream(custom_stream()):
# Check if it's an original chunk
if isinstance(event, MyCustomChunk):
original_count += 1
print(f"📦 Original Chunk #{original_count}:")
print(f" Text: {event.text!r}")
print(f" Metadata: {event.metadata}")
# Check if it's a StreamBlocks event
elif isinstance(event, TextDeltaEvent):
streamblocks_count += 1
print(f"🔵 TextDelta: {repr(event.delta)[:40]}")
elif isinstance(event, BlockStartEvent):
streamblocks_count += 1
print(f"🟢 BlockOpened: {event.syntax}")
elif isinstance(event, BlockEndEvent):
streamblocks_count += 1
block = event.get_block()
if block is None:
continue
print("✅ BlockExtracted:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
streamblocks_count += 1
print(f"💬 RawText: {event.content}")
print()
print("Summary:")
print(f" Original chunks: {original_count}")
print(f" StreamBlocks events: {streamblocks_count}")
print()
print("✓ Both event types in same stream")
print("✓ Type checking with isinstance()")
print("✓ Original metadata preserved")
if __name__ == "__main__":
asyncio.run(main())
Text Delta Streaming
Character-by-character streaming with TextDeltaEvent, useful for typewriter effects and live progress indicators.
src/hother/streamblocks_examples/03_adapters/06_text_delta_streaming.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import sys
from collections.abc import AsyncGenerator
from hother.streamblocks import (
BlockEndEvent,
BlockStartEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextDeltaEvent,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
async def character_stream() -> AsyncGenerator[str]:
"""Stream text character by character."""
text = (
"Creating project structure...\n"
"!!files:files_operations\n"
"src/main.py:C\n"
"src/utils.py:C\n"
"tests/test_main.py:C\n"
"!!end\n"
"Done!\n"
)
# Stream character by character
for char in text:
yield char
await asyncio.sleep(0.02) # Slow for visual effect
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 06: Real-Time Text Streaming (Typewriter Effect)")
print("=" * 60)
print()
# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Streaming text in real-time:")
print("-" * 40)
async for event in processor.process_stream(character_stream()):
if isinstance(event, TextDeltaEvent):
# Print each character immediately (typewriter effect)
sys.stdout.write(event.delta)
sys.stdout.flush()
# Show context
if event.inside_block:
context = f" [{event.section}]"
# Clear the context indicator on newlines
if event.delta == "\n":
sys.stdout.write(f" {context}\n")
elif isinstance(event, BlockStartEvent):
sys.stdout.write(f"\n[Block starting: {event.syntax}]\n")
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n[Block extracted:]")
print(block.model_dump_json(indent=2))
print("-" * 40)
print()
print("✓ Character-by-character streaming")
print("✓ Inside/outside block tracking")
print("✓ Perfect for live UIs")
if __name__ == "__main__":
asyncio.run(main())
Block Opened Event
Uses BlockOpenedEvent to prepare UI elements or resources before any block content arrives, the earliest signal that a block is starting.
src/hother/streamblocks_examples/03_adapters/07_block_opened_event.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from collections.abc import AsyncGenerator
from hother.streamblocks import (
BlockEndEvent,
BlockStartEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextDeltaEvent,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
async def delayed_stream() -> AsyncGenerator[str]:
"""Stream with delays to show timing."""
chunks = [
"Some text...\n",
"!!files:files_operations\n", # Block opens here!
]
for chunk in chunks:
yield chunk
await asyncio.sleep(0.1)
# Simulate slow content arrival
print(" [Simulating network delay...]")
await asyncio.sleep(1.0)
for chunk in ["src/app.py:C\n", "src/test.py:C\n", "!!end\n"]:
yield chunk
await asyncio.sleep(0.1)
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 07: BlockOpenedEvent")
print("=" * 60)
print()
# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Processing stream (watch for BlockOpened event):")
print()
# Track active blocks
active_blocks = {}
async for event in processor.process_stream(delayed_stream()):
if isinstance(event, BlockStartEvent):
print("🟢 BlockOpened!")
print(f" Syntax: {event.syntax}")
print(f" Line: {event.start_line}")
if event.inline_metadata:
print(f" Metadata: {event.inline_metadata}")
# Prepare resources (e.g., create UI widget)
active_blocks[event.start_line] = {
"syntax": event.syntax,
"content": [],
}
print(" → UI widget created")
print(" → Waiting for content...")
print()
elif isinstance(event, TextDeltaEvent):
# Accumulate content for active blocks
if event.inside_block and active_blocks:
line = next(iter(active_blocks.keys()))
active_blocks[line]["content"].append(event.delta)
print(f" 📝 Accumulating: {repr(event.delta)[:30]}")
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n✅ BlockExtracted:")
print(block.model_dump_json(indent=2))
# Clean up
active_blocks.clear()
print(" → UI widget finalized\n")
print()
print("✓ BlockOpened emitted immediately")
print("✓ Prepare UI before content arrives")
print("✓ Better UX with early feedback")
if __name__ == "__main__":
asyncio.run(main())
Configuration Flags
A tour of ProcessorConfig options, emit_original_events, emit_text_deltas, and auto_detect_adapter, and how each changes the emitted event stream.
src/hother/streamblocks_examples/03_adapters/08_configuration_flags.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from collections.abc import AsyncGenerator
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextDeltaEvent,
)
from hother.streamblocks.core.processor import ProcessorConfig
from hother.streamblocks.extensions.gemini import GeminiInputAdapter
from hother.streamblocks_examples.blocks.agent.files import FileOperations
# Mock Gemini chunk
class GeminiChunk:
__module__ = "google.genai.types"
def __init__(self, text: str) -> None:
self.text = text
async def gemini_stream() -> AsyncGenerator[GeminiChunk]:
"""Simple Gemini stream."""
for chunk in ["!!f:files_operations\n", "app.py:C\n", "!!end\n"]:
yield GeminiChunk(chunk)
await asyncio.sleep(0.05)
async def demo_config(
name: str,
config: ProcessorConfig,
adapter: GeminiInputAdapter | None = None,
) -> None:
"""Demo a specific configuration."""
print(f"\n{'=' * 60}")
print(f"Configuration: {name}")
print(
f"Settings: emit_original={config.emit_original_events}, emit_deltas={config.emit_text_deltas}, auto_detect={config.auto_detect_adapter}"
)
print("=" * 60)
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry, config=config)
events_seen = {"original": 0, "text_delta": 0, "block": 0}
async for event in processor.process_stream(gemini_stream(), adapter=adapter):
if isinstance(event, GeminiChunk):
events_seen["original"] += 1
print(" Original chunk")
elif isinstance(event, TextDeltaEvent):
events_seen["text_delta"] += 1
print(" Text delta")
elif isinstance(event, BlockEndEvent):
events_seen["block"] += 1
print(" Block extracted")
print(
f"\nEvents: Original={events_seen['original']}, "
f"TextDelta={events_seen['text_delta']}, "
f"Block={events_seen['block']}"
)
async def main() -> None:
"""Run all configuration demos."""
print("=" * 60)
print("Example 08: Configuration Options")
print("=" * 60)
# Default configuration
await demo_config(
"Default (All Enabled)",
ProcessorConfig(
emit_original_events=True,
emit_text_deltas=True,
auto_detect_adapter=True,
),
)
# Disable original events
await demo_config(
"No Original Events (Lightweight)",
ProcessorConfig(
emit_original_events=False,
emit_text_deltas=True,
auto_detect_adapter=True,
),
)
# Disable text deltas
await demo_config(
"No Text Deltas (Line-based only)",
ProcessorConfig(
emit_original_events=True,
emit_text_deltas=False,
auto_detect_adapter=True,
),
)
# Manual adapter (no auto-detect)
await demo_config(
"Manual Adapter (Explicit)",
ProcessorConfig(
emit_original_events=True,
emit_text_deltas=True,
auto_detect_adapter=False,
),
adapter=GeminiInputAdapter(),
)
# Minimal mode
await demo_config(
"Minimal Mode (Only Blocks)",
ProcessorConfig(
emit_original_events=False,
emit_text_deltas=False,
auto_detect_adapter=True,
),
)
print("\n" + "=" * 60)
print("Configuration Summary:")
print("=" * 60)
print("emit_original_events - Pass through original chunks")
print("emit_text_deltas - Real-time text streaming")
print("auto_detect_adapter - Auto-detect from first chunk")
print("\nRecommendations:")
print("- Default: Full transparency + real-time streaming")
print("- Lightweight: Disable originals if not needed")
print("- Performance: Disable deltas for batch processing")
if __name__ == "__main__":
asyncio.run(main())
Custom Adapter
Creates a custom input adapter for a proprietary streaming format and registers it for auto-detection, so downstream code never sees the raw chunks.
src/hother/streamblocks_examples/03_adapters/09_custom_adapter.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from collections.abc import AsyncGenerator
from typing import Any
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
EventCategory,
InputAdapterRegistry,
Registry,
StreamBlockProcessor,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
# Custom streaming format
class ProprietaryChunk:
"""Custom chunk format from proprietary API."""
__module__ = "mycompany.streaming.api"
def __init__(self, payload: str, meta: dict[str, str | bool] | None = None) -> None:
self.payload = payload # Text is in 'payload'
self.meta = meta or {} # Metadata in 'meta'
# Custom adapter implementing the InputProtocolAdapter protocol
class ProprietaryInputAdapter:
"""Input adapter for proprietary streaming format."""
def categorize(self, event: ProprietaryChunk) -> EventCategory:
"""All events contain text content."""
return EventCategory.TEXT_CONTENT
def extract_text(self, chunk: ProprietaryChunk) -> str | None:
"""Extract text from payload field."""
return chunk.payload
def is_complete(self, chunk: ProprietaryChunk) -> bool:
"""Check if stream is complete."""
return chunk.meta.get("final", False)
def get_metadata(self, chunk: ProprietaryChunk) -> dict[str, Any] | None:
"""Extract metadata."""
if chunk.meta:
return {
"request_id": chunk.meta.get("request_id"),
"timestamp": chunk.meta.get("timestamp"),
}
return None
async def proprietary_stream() -> AsyncGenerator[ProprietaryChunk]:
"""Simulate proprietary API stream."""
chunks = [
ProprietaryChunk("!!proj:files_operations\n", {"request_id": "req-123"}),
ProprietaryChunk("main.py:C\n", {"timestamp": "2024-01-01"}),
ProprietaryChunk("test.py:C\n", {"timestamp": "2024-01-01"}),
ProprietaryChunk("!!end\n", {"final": True}),
]
for chunk in chunks:
yield chunk
await asyncio.sleep(0.1)
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 09: Custom Adapter with Registration")
print("=" * 60)
print()
# Register custom adapter for auto-detection
print("Registering custom adapter...")
InputAdapterRegistry.register_module(
"mycompany.streaming",
ProprietaryInputAdapter,
)
print("Registered for module: mycompany.streaming.*")
print()
# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Processing proprietary stream (auto-detecting custom adapter)...")
print()
async for event in processor.process_stream(proprietary_stream()):
# Original chunks
if isinstance(event, ProprietaryChunk):
print("Proprietary Chunk:")
print(f" Payload: {repr(event.payload)[:40]}")
print(f" Meta: {event.meta}")
# Extracted blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\nBlock Extracted:")
print(block.model_dump_json(indent=2))
print()
# Cleanup - remove from registry
del InputAdapterRegistry._type_registry["mycompany.streaming"]
print()
print("Custom adapter created")
print("Registered for auto-detection")
print("Works with proprietary formats")
if __name__ == "__main__":
asyncio.run(main())
Callable Adapter
A simple inline input adapter class for quick custom extraction, handy when a full adapter implementation would be overkill.
src/hother/streamblocks_examples/03_adapters/10_callable_adapter.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from typing import Any
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
EventCategory,
Registry,
StreamBlockProcessor,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
# Simple dict-based chunks
async def dict_stream():
"""Stream of dictionaries."""
chunks = [
{"content": "!!files:files_operations\n", "id": 1},
{"content": "app.py:C\n", "id": 2},
{"content": "test.py:C\n", "id": 3},
{"content": "!!end\n", "id": 4, "done": True},
]
for chunk in chunks:
yield chunk
await asyncio.sleep(0.1)
# Simple inline adapter class
class DictInputAdapter:
"""Simple adapter for dict-based chunks."""
def categorize(self, event: dict[str, Any]) -> EventCategory:
"""All dicts contain text content."""
return EventCategory.TEXT_CONTENT
def extract_text(self, chunk: dict[str, Any]) -> str | None:
"""Extract text from 'content' key."""
return chunk.get("content")
def is_complete(self, chunk: dict[str, Any]) -> bool:
"""Check 'done' flag for completion."""
return chunk.get("done", False)
def get_metadata(self, chunk: dict[str, Any]) -> dict[str, Any] | None:
"""Extract ID as metadata."""
if "id" in chunk:
return {"chunk_id": chunk.get("id")}
return None
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 10: Simple Inline Adapter")
print("=" * 60)
print()
# Create adapter instance
adapter = DictInputAdapter()
# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Processing dict stream with inline adapter...")
print()
async for event in processor.process_stream(dict_stream(), adapter=adapter):
# Original dicts
if isinstance(event, dict):
print(f"Dict Chunk: id={event['id']}, content={repr(event['content'])[:30]}")
if event.get("done"):
print(" Final chunk!")
# Blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\nBlock Extracted:")
print(block.model_dump_json(indent=2))
print()
print()
print("Simple adapter class")
print("Just implement: categorize, extract_text, is_complete, get_metadata")
print("Perfect for quick prototyping")
if __name__ == "__main__":
asyncio.run(main())
Attribute Adapter (Generic)
Uses AttributeInputAdapter to handle any object exposing a text-like attribute, without writing adapter code at all.
src/hother/streamblocks_examples/03_adapters/11_attribute_adapter_generic.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from collections.abc import AsyncGenerator
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
)
from hother.streamblocks.adapters.input import AttributeInputAdapter
from hother.streamblocks_examples.blocks.agent.files import FileOperations
# Generic chunk classes
class ResponseChunk:
"""Some API response with 'message' attribute."""
def __init__(self, message: str, status: str = "active") -> None:
self.message = message # Text is in 'message'
self.status = status
self.finish_reason: str | None = None
class FinalChunk(ResponseChunk):
"""Final chunk with finish_reason."""
def __init__(self, message: str) -> None:
super().__init__(message, status="complete")
self.finish_reason: str | None = "done"
async def generic_stream() -> AsyncGenerator[ResponseChunk]:
"""Stream with generic chunk objects."""
chunks = [
ResponseChunk("!!files:files_operations\n"),
ResponseChunk("src/app.py:C\n"),
ResponseChunk("src/utils.py:C\n"),
ResponseChunk("!!end\n"),
FinalChunk(""), # Final chunk
]
for chunk in chunks:
yield chunk
await asyncio.sleep(0.1)
async def main() -> None:
"""Run the example."""
print("=" * 60)
print("Example 11: AttributeInputAdapter (Generic Objects)")
print("=" * 60)
print()
# Create adapter for 'message' attribute
adapter = AttributeInputAdapter(text_attr="message")
# Setup
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Processing stream with AttributeInputAdapter (text_attr='message')...")
print()
async for event in processor.process_stream(generic_stream(), adapter=adapter):
# Original chunks
if isinstance(event, (ResponseChunk, FinalChunk)):
print(f"Chunk: message={repr(event.message)[:30]}, status={event.status}")
if event.finish_reason:
print(f" Finish reason: {event.finish_reason}")
# Blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\nBlock Extracted:")
print(block.model_dump_json(indent=2))
print()
print()
print("Works with any object")
print("Just specify attribute name")
print("Handles finish_reason automatically")
print()
print("Other common attributes:")
print(" - text_attr='text' (default)")
print(" - text_attr='content'")
print(" - text_attr='data'")
print(" - text_attr='message'")
if __name__ == "__main__":
asyncio.run(main())
Disable Original Events
Disables original event passthrough for a lightweight, extraction-only event stream with minimal overhead.
src/hother/streamblocks_examples/03_adapters/12_disable_original_events.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from collections.abc import AsyncGenerator
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextContentEvent,
TextDeltaEvent,
)
from hother.streamblocks.core.processor import ProcessorConfig
from hother.streamblocks_examples.blocks.agent.files import FileOperations
# Mock Gemini chunk
class GeminiChunk:
__module__ = "google.genai.types"
def __init__(self, text: str) -> None:
self.text = text
self.size = len(text.encode()) # Simulate chunk size
async def large_gemini_stream() -> AsyncGenerator[GeminiChunk]:
"""Simulate large Gemini stream."""
chunks = [
"Creating project...\n",
"!!files:files_operations\n",
"src/main.py:C\n",
"src/utils.py:C\n",
"tests/test_main.py:C\n",
"!!end\n",
"Done!\n",
]
for chunk in chunks:
yield GeminiChunk(chunk)
await asyncio.sleep(0.05)
async def compare_modes() -> None:
"""Compare normal vs lightweight mode."""
print("=" * 60)
print("Example 12: Lightweight Mode (No Original Events)")
print("=" * 60)
print()
# Mode 1: Normal (with original events)
print("Mode 1: Normal (emit_original_events=True)")
print("-" * 40)
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
config_normal = ProcessorConfig(emit_original_events=True) # Default
processor_normal = StreamBlockProcessor(registry, config=config_normal)
event_count_normal = 0
async for event in processor_normal.process_stream(large_gemini_stream()):
event_count_normal += 1
if isinstance(event, GeminiChunk):
print(f" 📦 GeminiChunk ({event.size} bytes)")
elif isinstance(event, TextDeltaEvent):
print(" 📝 TextDelta")
elif isinstance(event, BlockEndEvent):
print(" ✅ Block")
print(f"\nTotal events: {event_count_normal}")
# Mode 2: Lightweight (without original events)
print("\n" + "=" * 60)
print("Mode 2: Lightweight (emit_original_events=False)")
print("-" * 40)
config_light = ProcessorConfig(emit_original_events=False) # Disable passthrough
processor_light = StreamBlockProcessor(registry, config=config_light)
event_count_light = 0
async for event in processor_light.process_stream(large_gemini_stream()):
event_count_light += 1
if isinstance(event, GeminiChunk):
print(" 📦 GeminiChunk (shouldn't see this!)")
elif isinstance(event, TextDeltaEvent):
print(" 📝 TextDelta")
elif isinstance(event, TextContentEvent):
print(" 💬 RawText")
elif isinstance(event, BlockEndEvent):
print(" ✅ Block")
print(f"\nTotal events: {event_count_light}")
# Summary
print("\n" + "=" * 60)
print("Summary:")
print("=" * 60)
print(f"Normal mode events: {event_count_normal}")
print(f"Lightweight mode events: {event_count_light}")
print(f"Reduction: {event_count_normal - event_count_light} events")
print(f"Savings: {((event_count_normal - event_count_light) / event_count_normal * 100):.1f}%")
print()
print("When to use lightweight mode:")
print(" ✓ Don't need original chunks")
print(" ✓ Only care about StreamBlocks events")
print(" ✓ Maximum performance needed")
print(" ✓ Batch processing")
print()
print("When to use normal mode:")
print(" ✓ Need access to provider metadata")
print(" ✓ Want complete transparency")
print(" ✓ Debugging/logging original events")
if __name__ == "__main__":
asyncio.run(compare_modes())
Manual Chunk Processing
Requires GEMINI_API_KEY (or GOOGLE_API_KEY). Processes chunks manually with process_chunk() instead of process_stream(), giving fine-grained control for custom buffering, multiple sources, or batch pipelines.
src/hother/streamblocks_examples/03_adapters/13_manual_chunk_processing.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import os
import sys
from collections.abc import AsyncIterator
from typing import Any
# Check for Gemini SDK
try:
from google import genai
except ImportError:
print("Error: google-genai package not installed.")
print("Install it with: pip install streamblocks[gemini]")
print("Or: pip install google-genai")
sys.exit(1)
from hother.streamblocks import (
BlockEndEvent,
DelimiterPreambleSyntax,
Registry,
StreamBlockProcessor,
TextDeltaEvent,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
async def get_gemini_response(prompt: str | None = None) -> AsyncIterator[Any]:
"""Get Gemini API response stream."""
# Get API key
api_key = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY")
if not api_key:
msg = (
"Please set GOOGLE_API_KEY or GEMINI_API_KEY environment variable.\n"
"Get your key at: https://aistudio.google.com/apikey"
)
raise ValueError(msg)
# Create client
client = genai.Client(api_key=api_key) # type: ignore[attr-defined]
# Default prompt
default_prompt = """Create a simple project structure with these files:
- src/main.py
- src/utils.py
- tests/test_main.py
Use this EXACT format (DO NOT use markdown code fences or any other formatting):
!!project:files_operations
src/main.py:C
src/utils.py:C
tests/test_main.py:C
!!end
IMPORTANT:
- Start your response directly with !! (no markdown, no code fences, no backticks)
- The first characters must be !!project:files_operations
- End with !!end on its own line
"""
# Return the stream
return await client.aio.models.generate_content_stream( # type: ignore[attr-defined]
model="gemini-2.5-flash",
contents=prompt or default_prompt,
)
async def example_basic_manual_processing() -> None:
"""Example 1: Basic manual chunk processing."""
print("=" * 60)
print("Example 1: Basic Manual Processing")
print("=" * 60)
print()
# Setup processor
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Processing chunks manually...")
print()
# Get stream
response = await get_gemini_response()
# Process chunks manually
chunk_count = 0
event_count = 0
async for chunk in response: # type: ignore[var-annotated]
chunk_count += 1
# Process this chunk and get all events it produces
events = processor.process_chunk(chunk)
# Handle each event
for event in events:
event_count += 1
# Original Gemini chunks
if hasattr(event, "__module__") and "google.genai" in event.__module__:
print(f"🔵 Chunk #{chunk_count}: Gemini event")
# Text deltas
elif isinstance(event, TextDeltaEvent):
status = "inside block" if event.inside_block else "outside block"
print(f"📝 Chunk #{chunk_count}: Text delta ({status})")
# Extracted blocks
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print(f"\n✅ Chunk #{chunk_count}: Block extracted!")
print(block.model_dump_json(indent=2))
print()
# Finalize to process remaining text and get any rejection events
final_events = processor.finalize()
for event in final_events:
event_count += 1
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n✅ Finalize: Block extracted!")
print(block.model_dump_json(indent=2))
print()
elif isinstance(event, TextDeltaEvent):
print("📝 Finalize: Text delta (processing remaining text)")
else:
print(f"⚠️ Finalize: {event.type}")
print()
print(f"Processed {chunk_count} chunks → {event_count} events")
print()
async def example_selective_processing() -> None:
"""Example 2: Selective chunk processing with custom logic."""
print("=" * 60)
print("Example 2: Selective Processing")
print("=" * 60)
print()
# Setup processor
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Only processing chunks with 'create' operations...")
print()
# Get stream
response = await get_gemini_response()
processed_chunks = 0
skipped_chunks = 0
async for chunk in response: # type: ignore[var-annotated]
# Get text from chunk (you could use adapter explicitly)
text = getattr(chunk, "text", "")
# Custom logic: only process chunks that might contain file operations
if ":" in text and ("C" in text or "create" in text.lower()):
events = processor.process_chunk(chunk)
processed_chunks += 1
for event in events:
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("✅ Block Extracted:")
print(block.model_dump_json(indent=2))
print(f" Found in chunk with text: {text[:50]}...")
else:
# Skip this chunk
skipped_chunks += 1
# Finalize to process remaining text and get any extracted blocks
final_events = processor.finalize()
for event in final_events:
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("✅ Block from finalize:")
print(block.model_dump_json(indent=2))
print()
print(f"Processed: {processed_chunks} chunks")
print(f"Skipped: {skipped_chunks} chunks")
print()
async def example_batch_processing() -> None:
"""Example 3: Batch processing with buffers."""
print("=" * 60)
print("Example 3: Batch Processing")
print("=" * 60)
print()
# Setup processor
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
print("Processing chunks in batches of 3...")
print()
# Get stream
response = await get_gemini_response()
chunk_buffer = []
batch_size = 3
batch_number = 0
async for chunk in response: # type: ignore[var-annotated]
chunk_buffer.append(chunk)
# Process when buffer is full
if len(chunk_buffer) >= batch_size:
batch_number += 1
print(f"Processing batch #{batch_number} ({len(chunk_buffer)} chunks)...")
for buffered_chunk in chunk_buffer:
events = processor.process_chunk(buffered_chunk)
for event in events:
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print(" ✅ Block Extracted:")
print(block.model_dump_json(indent=2))
chunk_buffer.clear()
# Process remaining chunks
if chunk_buffer:
batch_number += 1
print(f"Processing final batch #{batch_number} ({len(chunk_buffer)} chunks)...")
for buffered_chunk in chunk_buffer:
events = processor.process_chunk(buffered_chunk)
for event in events:
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print(" ✅ Block Extracted:")
print(block.model_dump_json(indent=2))
# Finalize to process remaining text and get any extracted blocks
final_events = processor.finalize()
for event in final_events:
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
print("\n ✅ Block from finalize:")
print(block.model_dump_json(indent=2))
print()
print(f"Processed {batch_number} batches total")
print()
async def main() -> None:
"""Run all examples."""
print("🔧 StreamBlocks Manual Chunk Processing Examples")
print()
try:
# Run all examples
await example_basic_manual_processing()
await example_selective_processing()
await example_batch_processing()
print("=" * 60)
print("Summary")
print("=" * 60)
print()
print("✅ Manual chunk processing with process_chunk()")
print("✅ Custom processing logic and filtering")
print("✅ Batch processing with buffers")
print("✅ Finalization with finalize()")
print()
print("Key Benefits:")
print("- Fine-grained control over processing")
print("- Integration with existing pipelines")
print("- Custom buffering and batching strategies")
print("- Selective processing based on content")
print()
except ValueError as e:
print(f"\n❌ Error: {e}")
sys.exit(1)
except Exception as e:
print(f"\n❌ Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
Section Delta Events
Uses the section-specific delta events (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent) for type-safe handling of each block section as it streams.
src/hother/streamblocks_examples/03_adapters/14_section_delta_events.py
View source on GitHub
import asyncio
from collections.abc import AsyncIterator
from hother.streamblocks import (
DelimiterFrontmatterSyntax,
Registry,
StreamBlockProcessor,
)
from hother.streamblocks.core.processor import ProcessorConfig
from hother.streamblocks.core.types import (
BlockContentDeltaEvent,
BlockEndEvent,
BlockHeaderDeltaEvent,
BlockMetadataDeltaEvent,
BlockStartEvent,
TextContentEvent,
)
async def example_stream() -> AsyncIterator[str]:
"""Example stream with a delimiter frontmatter block."""
text = """Here's some introductory text before the block.
!!start
---
id: config-update
block_type: config
version: 2.0
author: dev-team
---
database:
host: localhost
port: 5432
driver: postgresql
logging:
level: INFO
format: json
!!end
And some text after the block.
"""
# Simulate streaming by yielding chunks
chunk_size = 40
for i in range(0, len(text), chunk_size):
chunk = text[i : i + chunk_size]
yield chunk
await asyncio.sleep(0.02)
async def main() -> None:
"""Demonstrate section-specific delta events."""
# Create syntax and registry
syntax = DelimiterFrontmatterSyntax()
registry = Registry(syntax=syntax)
# Create processor with text deltas disabled for clarity
config = ProcessorConfig(emit_text_deltas=False) # Disable TextDeltaEvent for cleaner output
processor = StreamBlockProcessor(registry, config=config)
print("=" * 70)
print("SECTION-SPECIFIC DELTA EVENTS DEMONSTRATION")
print("=" * 70)
print()
print("Processing stream and capturing section-specific events...")
print()
# Counters for different event types
header_count = 0
metadata_count = 0
content_count = 0
# Store accumulated content by section
metadata_lines: list[str] = []
content_lines: list[str] = []
async for event in processor.process_stream(example_stream()):
# Handle lifecycle events
if isinstance(event, BlockStartEvent):
print(f"[START] Block opened at line {event.start_line}")
print(f" Syntax: {event.syntax}")
print(f" Block ID: {event.block_id}")
print()
# Handle section-specific delta events with type safety
elif isinstance(event, BlockHeaderDeltaEvent):
# Type-safe: isinstance check works!
header_count += 1
print(f"[HEADER] Line {event.current_line}: {event.delta!r}")
# Header-specific field: inline_metadata
if event.inline_metadata:
print(f" Inline metadata: {event.inline_metadata}")
elif isinstance(event, BlockMetadataDeltaEvent):
# Type-safe handling of metadata section
metadata_count += 1
metadata_lines.append(event.delta)
# Metadata-specific field: is_boundary
boundary_marker = " [BOUNDARY]" if event.is_boundary else ""
print(f"[META] Line {event.current_line}: {event.delta!r}{boundary_marker}")
elif isinstance(event, BlockContentDeltaEvent):
# Type-safe handling of content section
content_count += 1
content_lines.append(event.delta)
print(f"[CONTENT] Line {event.current_line}: {event.delta!r}")
print(f" Accumulated size: {event.accumulated_size} bytes")
# Handle completion
elif isinstance(event, BlockEndEvent):
print()
print(f"[END] Block completed: {event.block_id}")
print(f" Lines: {event.start_line}-{event.end_line}")
print(f" Type: {event.block_type}")
elif isinstance(event, TextContentEvent):
# Regular text outside blocks
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# Summary
print()
print("=" * 70)
print("SUMMARY")
print("=" * 70)
print(f"Header deltas: {header_count}")
print(f"Metadata deltas: {metadata_count}")
print(f"Content deltas: {content_count}")
print()
print("Benefits of section-specific events:")
print("1. Type safety - isinstance() checks work without string comparisons")
print("2. Unique fields - is_boundary for metadata, inline_metadata for header")
print("3. Clear intent - handler code clearly shows which section is processed")
if __name__ == "__main__":
asyncio.run(main())
Section End Events
Reacts to BlockMetadataEndEvent and BlockContentEndEvent to process completed sections early, before the block itself finishes, enabling early validation and resource release.
src/hother/streamblocks_examples/03_adapters/15_section_end_events.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from hother.streamblocks import (
BlockContentEndEvent,
BlockEndEvent,
BlockMetadataEndEvent,
BlockStartEvent,
DelimiterFrontmatterSyntax,
DelimiterPreambleSyntax,
MetadataValidationFailureMode,
Registry,
StreamBlockProcessor,
ValidationResult,
)
from hother.streamblocks.core.processor import ProcessorConfig
from hother.streamblocks_examples.blocks.agent.files import FileOperations
from hother.streamblocks_examples.helpers.simulator import simulated_stream
async def example_basic_section_events() -> None:
"""Example 1: Basic section end events."""
print("=" * 60)
print("Example 1: Basic Section End Events (Preamble Syntax)")
print("=" * 60)
print()
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
# Section end events are enabled by default
processor = StreamBlockProcessor(registry)
text = "".join(
[
"Starting project...\n",
"!!project:files_operations\n",
"src/main.py:C\n",
"src/utils.py:C\n",
"!!end\n",
"Done!\n",
]
)
print("Processing stream with section end events enabled:")
print()
async for event in processor.process_stream(simulated_stream(text, preset="fast")):
if isinstance(event, BlockStartEvent):
print(f" [BlockStart] Block opened: {event.syntax}")
elif isinstance(event, BlockContentEndEvent):
print(" [ContentEnd] Content section complete")
print(f" - Raw content length: {len(event.raw_content)} chars")
print(f" - Parsed: {event.parsed_content}")
print(f" - Validation passed: {event.validation_passed}")
elif isinstance(event, BlockEndEvent):
block = event.get_block()
if block:
print(f" [BlockEnd] Block extracted: {event.block_type}")
print(block.model_dump_json(indent=2))
print()
async def example_disabled_section_events() -> None:
"""Example 2: Section end events disabled."""
print("=" * 60)
print("Example 2: Section End Events Disabled (Opt-out)")
print("=" * 60)
print()
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
# Disable section end events for maximum performance
config = ProcessorConfig(emit_section_end_events=False) # Opt-out
processor = StreamBlockProcessor(registry, config=config)
text = "".join(
[
"!!project:files_operations\n",
"src/main.py:C\n",
"!!end\n",
]
)
print("Processing stream with section end events DISABLED:")
print()
event_types = []
async for event in processor.process_stream(simulated_stream(text, preset="fast")):
event_types.append(type(event).__name__)
if isinstance(event, BlockEndEvent):
print(" [BlockEnd] Block extracted (no ContentEnd event)")
print()
print(f"Event types received: {event_types}")
print("Notice: No BlockContentEndEvent when disabled")
print()
async def example_frontmatter_metadata_events() -> None:
"""Example 3: Metadata end events with frontmatter syntax."""
print("=" * 60)
print("Example 3: Metadata End Events (Frontmatter Syntax)")
print("=" * 60)
print()
syntax = DelimiterFrontmatterSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
text = "".join(
[
"!!start\n",
"---\n",
"id: my_block\n",
"block_type: files_operations\n",
"---\n", # This triggers BlockMetadataEndEvent
"src/main.py:C\n",
"!!end\n",
]
)
print("Processing stream with YAML frontmatter:")
print()
async for event in processor.process_stream(simulated_stream(text, preset="fast")):
if isinstance(event, BlockStartEvent):
print(" [BlockStart] Block opened")
elif isinstance(event, BlockMetadataEndEvent):
print(" [MetadataEnd] Metadata section complete!")
print(f" - Parsed metadata: {event.parsed_metadata}")
print(f" - Validation passed: {event.validation_passed}")
elif isinstance(event, BlockContentEndEvent):
print(" [ContentEnd] Content section complete")
elif isinstance(event, BlockEndEvent):
print(f" [BlockEnd] Block extracted: {event.block_type}")
print()
async def example_early_validation() -> None:
"""Example 4: Early validation with section end events."""
print("=" * 60)
print("Example 4: Early Validation with Section End Events")
print("=" * 60)
print()
syntax = DelimiterFrontmatterSyntax()
registry = Registry(
syntax=syntax,
# Abort block on metadata validation failure
metadata_failure_mode=MetadataValidationFailureMode.ABORT_BLOCK,
)
registry.register("files_operations", FileOperations)
# Add a metadata validator that requires specific fields
def validate_metadata(raw: str, parsed: dict | None) -> ValidationResult:
if not parsed:
return ValidationResult.failure("No metadata parsed")
if parsed.get("id") == "forbidden":
return ValidationResult.failure("ID 'forbidden' is not allowed")
return ValidationResult.success()
registry.add_metadata_validator("files_operations", validate_metadata)
processor = StreamBlockProcessor(registry)
print("Processing block with forbidden ID (will fail validation):")
print()
# This block will fail validation
text = "".join(
[
"!!start\n",
"---\n",
"id: forbidden\n",
"block_type: files_operations\n",
"---\n",
"src/main.py:C\n",
"!!end\n",
]
)
async for event in processor.process_stream(simulated_stream(text, preset="fast")):
if isinstance(event, BlockMetadataEndEvent):
print(f" [MetadataEnd] Validation passed: {event.validation_passed}")
if not event.validation_passed:
print(f" Error: {event.validation_error}")
elif hasattr(event, "error_code"):
print(" [Error] Block aborted due to validation failure")
print()
print("Processing block with valid ID (will pass validation):")
print()
# This block will pass validation
text_valid = "".join(
[
"!!start\n",
"---\n",
"id: allowed\n",
"block_type: files_operations\n",
"---\n",
"src/main.py:C\n",
"!!end\n",
]
)
# Reset processor
processor = StreamBlockProcessor(registry)
async for event in processor.process_stream(simulated_stream(text_valid, preset="fast")):
if isinstance(event, BlockMetadataEndEvent):
print(f" [MetadataEnd] Validation passed: {event.validation_passed}")
elif isinstance(event, BlockEndEvent):
print(" [BlockEnd] Block extracted successfully!")
print()
async def main() -> None:
"""Run all examples."""
print()
print("Section End Events Examples")
print()
await example_basic_section_events()
await example_disabled_section_events()
await example_frontmatter_metadata_events()
await example_early_validation()
print("=" * 60)
print("Summary")
print("=" * 60)
print()
print("Section end events enable:")
print(" - Early processing of completed sections")
print(" - Section-specific validation (abort early on failure)")
print(" - UI state management (update UI when sections complete)")
print(" - Streaming optimization (release resources early)")
print()
print("Configuration:")
print(" - emit_section_end_events=True (default)")
print(" - emit_section_end_events=False (opt-out for performance)")
print()
if __name__ == "__main__":
asyncio.run(main())
Bidirectional Protocol
Combines input and output adapters in a ProtocolStreamProcessor for bidirectional protocol processing, with EventCategory-based filtering of what gets emitted.
src/hother/streamblocks_examples/03_adapters/16_bidirectional_protocol.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
from dataclasses import dataclass
from textwrap import dedent
from typing import Any
from hother.streamblocks import DelimiterFrontmatterSyntax, Registry
from hother.streamblocks.adapters.categories import EventCategory
from hother.streamblocks.core.protocol_processor import ProtocolStreamProcessor
from hother.streamblocks.core.types import BaseEvent, BlockEndEvent, TextDeltaEvent
from hother.streamblocks_examples.blocks.agent.files import FileOperations
@dataclass
class CustomInputEvent:
"""Custom input event format."""
event_type: str # "text", "metadata", "control"
payload: str
@dataclass
class CustomOutputEvent:
"""Custom output event format."""
kind: str # "delta", "block", "other"
data: Any
class CustomInputAdapter:
"""Input adapter for custom event format."""
def categorize(self, event: CustomInputEvent) -> EventCategory:
"""Categorize event for routing."""
if event.event_type == "text":
return EventCategory.TEXT_CONTENT
if event.event_type == "metadata":
return EventCategory.PASSTHROUGH
return EventCategory.SKIP
def extract_text(self, event: CustomInputEvent) -> str | None:
"""Extract text from text events."""
return event.payload if event.event_type == "text" else None
def get_metadata(self, event: CustomInputEvent) -> dict[str, Any] | None:
"""Extract metadata from events."""
if event.event_type == "metadata":
return {"source": event.payload}
return None
class CustomOutputAdapter:
"""Output adapter for custom event format."""
def to_protocol_event(self, event: BaseEvent) -> CustomOutputEvent | None:
"""Convert StreamBlocks events to custom format."""
if isinstance(event, TextDeltaEvent):
return CustomOutputEvent(kind="delta", data=event.delta)
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block:
return CustomOutputEvent(
kind="block",
data={"id": block.metadata.id, "type": block.metadata.block_type},
)
return None
def passthrough(self, original_event: Any) -> CustomOutputEvent | None:
"""Pass through non-text events."""
if isinstance(original_event, CustomInputEvent):
return CustomOutputEvent(kind="other", data=original_event.payload)
return None
async def main() -> None:
"""Demonstrate bidirectional protocol processing."""
syntax = DelimiterFrontmatterSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = ProtocolStreamProcessor[CustomInputEvent, CustomOutputEvent](
registry,
input_adapter=CustomInputAdapter(),
output_adapter=CustomOutputAdapter(),
)
text = dedent("""
!!start
---
id: ops001
block_type: files_operations
---
src/main.py:C
!!end
""").strip()
async def input_stream():
# Mix of different event types
yield CustomInputEvent("metadata", "session-123")
yield CustomInputEvent("text", text)
yield CustomInputEvent("control", "ping") # Will be skipped
print("=== Bidirectional Protocol ===")
async for output_event in processor.process_stream(input_stream()):
print(f"[{output_event.kind}] {output_event.data}")
if __name__ == "__main__":
asyncio.run(main())
Custom Output Adapter
Writes a custom output adapter that converts StreamBlocks events into a simplified JSON event format, the pattern to follow when feeding events to a frontend or message bus.
src/hother/streamblocks_examples/03_adapters/17_custom_output_adapter.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
from hother.streamblocks.core.protocol_processor import ProtocolStreamProcessor
from hother.streamblocks.core.types import (
BaseEvent,
BlockEndEvent,
BlockStartEvent,
TextContentEvent,
)
from hother.streamblocks_examples.blocks.agent.files import FileOperations
class JsonEventAdapter:
"""Output adapter that emits simplified JSON-like dicts.
Features:
- Return None to filter out events
- Return a list to emit multiple events
- Transform events to any output format
"""
def to_protocol_event(self, event: BaseEvent) -> dict[str, Any] | list[dict[str, Any]] | None:
"""Convert StreamBlocks events to simplified dicts."""
# Filter out text content events (we only care about blocks)
if isinstance(event, TextContentEvent):
return None
# Emit block start and end as separate events
if isinstance(event, BlockStartEvent):
return {"event": "block_start", "block_id": event.block_id}
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block:
# Emit multiple events: block info + operations
events: list[dict[str, Any]] = [
{
"event": "block_end",
"block_id": block.metadata.id,
"block_type": block.metadata.block_type,
},
]
# Add operation details
if hasattr(block.content, "operations"):
for op in block.content.operations:
events.append(
{
"event": "operation",
"action": op.action,
"path": op.path,
}
)
return events
return None
def passthrough(self, original_event: Any) -> dict[str, Any] | None:
"""Handle passthrough events."""
return {"event": "passthrough", "data": str(original_event)}
async def main() -> None:
"""Demonstrate custom output adapter."""
syntax = DelimiterFrontmatterSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = ProtocolStreamProcessor[str, dict[str, Any]](
registry,
output_adapter=JsonEventAdapter(),
)
text = dedent("""
!!start
---
id: ops001
block_type: files_operations
---
src/main.py:C
src/utils.py:E
!!end
""").strip()
from hother.streamblocks_examples.helpers.simulator import simple_text_stream
print("=== Custom Output Adapter ===")
async for event in processor.process_stream(simple_text_stream(text)):
print(event)
if __name__ == "__main__":
asyncio.run(main())