Providers Examples
End-to-end demos against real AI providers. Both examples require a Gemini key: set GEMINI_API_KEY (or GOOGLE_API_KEY) and install the extra with pip install streamblocks[gemini]. For adapter-level provider examples (OpenAI, Anthropic), see the Adapters examples and the providers guide.
Gemini Simple Demo
Requires GEMINI_API_KEY. A simple end-to-end demo: one delimiter-frontmatter syntax for all block types, with Gemini generating blocks that are extracted live from the response stream.
src/hother/streamblocks_examples/07_providers/01_gemini_simple_demo.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import os
import sys
from collections.abc import AsyncIterator
from textwrap import dedent
from typing import TYPE_CHECKING, Any
# Check for Gemini SDK
try:
from google import genai # type: ignore[import-not-found]
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)
# Add parent directory to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from hother.streamblocks import (
BlockContentDeltaEvent,
BlockEndEvent,
BlockErrorEvent,
BlockHeaderDeltaEvent,
BlockMetadataDeltaEvent,
DelimiterFrontmatterSyntax,
Registry,
StreamBlockProcessor,
TextContentEvent,
)
from hother.streamblocks_examples.blocks.agent.files import (
FileContent,
FileContentContent,
FileContentMetadata,
FileOperations,
FileOperationsContent,
FileOperationsMetadata,
)
from hother.streamblocks_examples.blocks.agent.message import Message, MessageContent, MessageMetadata
if TYPE_CHECKING:
from hother.streamblocks.core.models import ExtractedBlock
from hother.streamblocks.core.types import BaseContent, BaseMetadata
def create_simple_prompt() -> str:
"""Create a simple system prompt."""
return dedent("""
You are a helpful AI assistant. When responding, use this single block format for everything:
!!start
---
id: unique_id
block_type: files_operations | file_content | message
---
Content goes here
!!end
Examples:
1. For file operations (use block_type: files_operations):
!!start
---
id: create_files_01
block_type: files_operations
description: Creating initial project structure
---
src/main.py:C
src/utils.py:C
tests/test_main.py:C
README.md:C
!!end
Note: C=create, E=edit, D=delete
2. For file content (use block_type: file_content, MUST include 'file' field):
!!start
---
id: main_py_content
block_type: file_content
file: src/main.py
description: Main application entry point
---
def main():
print("Hello, World!")
if __name__ == "__main__":
main()
!!end
3. For messages/communication (use block_type: message, MUST include 'message_type'):
!!start
---
id: status_01
block_type: message
message_type: info
title: Explaining the approach
---
I'll create a simple Flask web application with proper structure and tests.
!!end
Note: message_type can be: info, warning, error, success, status, explanation
Always use this format for ALL content - whether it's file operations, code content, or communication.
""").strip()
async def get_gemini_response(prompt: str) -> AsyncIterator[Any]:
"""Get Gemini API response stream.
Note: Returns the stream directly - no need for wrapper function!
The StreamBlockProcessor will auto-detect Gemini chunks and use
GeminiAdapter to extract text while preserving original chunks.
"""
# Try GOOGLE_API_KEY first (official), then GEMINI_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"
raise ValueError(msg)
client = genai.Client(api_key=api_key) # type: ignore[attr-defined]
# Combine system prompt with user prompt
system_prompt = create_simple_prompt()
full_prompt = f"{system_prompt}\n\nUser: {prompt}"
# Return the stream directly - no need to yield!
return await client.aio.models.generate_content_stream( # type: ignore[attr-defined]
model="gemini-2.5-flash",
contents=full_prompt,
)
async def main() -> None:
"""Run the simple Gemini demo."""
print("StreamBlocks + Gemini Simple Demo")
print("=" * 60)
print("\nUsing unified delimiter + frontmatter syntax for all blocks")
# Create a single syntax for all Gemini responses
syntax = DelimiterFrontmatterSyntax(
start_delimiter="!!start",
end_delimiter="!!end",
)
# Create registry and register all block types using default blocks
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
registry.register("file_content", FileContent)
registry.register("message", Message)
from hother.streamblocks.core.processor import ProcessorConfig
config = ProcessorConfig(lines_buffer=10)
processor = StreamBlockProcessor(registry, config=config)
# Example prompts
example_prompts = [
"Create a Python hello world script with a README file",
"Create a basic Flask web server with routes",
"Write a function to calculate fibonacci numbers",
"Create a simple calculator module with tests",
"Explain how to use async/await in Python with examples",
]
print("\nExample prompts:")
for i, prompt in enumerate(example_prompts, 1):
print(f"{i}. {prompt}")
# Detect non-interactive mode
is_non_interactive = (
not sys.stdin.isatty() # No TTY (piped/redirected)
or os.getenv("CI") # CI environment
or os.getenv("NON_INTERACTIVE") # Explicit flag
)
# Get user input
if len(sys.argv) > 1:
# Command-line argument provided
arg = sys.argv[1]
if arg.isdigit() and 1 <= int(arg) <= len(example_prompts):
user_prompt = example_prompts[int(arg) - 1]
else:
user_prompt = " ".join(sys.argv[1:])
print(f"\nUsing prompt from command line: {user_prompt}")
elif is_non_interactive:
# Non-interactive mode: use first example
user_prompt = example_prompts[0]
print("\nNon-interactive mode: using default prompt")
else:
# Interactive mode: ask user
user_input = input("\nEnter your request (or 1-5 for examples, Enter for #1): ").strip()
if user_input.isdigit() and 1 <= int(user_input) <= len(example_prompts):
user_prompt = example_prompts[int(user_input) - 1]
elif not user_input:
user_prompt = example_prompts[0]
else:
user_prompt = user_input
print(f"\nProcessing: {user_prompt}")
print("=" * 60)
# Track extracted blocks
extracted_blocks: list[ExtractedBlock[BaseMetadata, BaseContent]] = []
raw_text: list[str] = []
# Process the stream
try:
# Get response and pass directly to processor
response = await get_gemini_response(user_prompt)
async for event in processor.process_stream(response):
# Skip native Gemini events (we only care about StreamBlocks events)
if processor.is_native_event(event):
continue
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
extracted_blocks.append(block)
# Handle different block types with proper type narrowing
if block.metadata.block_type in ("files_operations", "file_content", "message"):
print(f"\nBlock extracted: {block.metadata.id}")
print(block.model_dump_json(indent=2))
elif isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
# Show progress
print("\r Processing block...", end="", flush=True)
elif isinstance(event, TextContentEvent):
# Collect any text outside blocks
text = event.content.strip()
if text:
raw_text.append(text)
elif isinstance(event, BlockErrorEvent):
print(f"\nBlock rejected: {event.reason}")
except Exception as e:
print(f"\nError: {e}")
import traceback
traceback.print_exc()
# Summary
print(f"\n\n{'=' * 60}")
print("SUMMARY:")
print(f" Extracted {len(extracted_blocks)} blocks")
# Count block types
block_types: dict[str, int] = {}
for block in extracted_blocks:
bt = block.metadata.block_type
block_types[bt] = block_types.get(bt, 0) + 1
for bt, count in sorted(block_types.items()):
print(f" - {bt}: {count}")
if raw_text:
print(f"\n Raw text lines: {len(raw_text)}")
if __name__ == "__main__":
asyncio.run(main())
Gemini Architect
Requires GEMINI_API_KEY. An AI software architect that mixes multiple block types in a single response, file operations, patches, tool calls, memory, and visualizations, showing how a realistic multi-block agent protocol comes together.
src/hother/streamblocks_examples/07_providers/02_gemini_architect.py
View source on GitHub
#!/usr/bin/env python3
from __future__ import annotations
import asyncio
import os
import sys
import traceback
from collections.abc import AsyncIterator
from textwrap import dedent
from typing import TYPE_CHECKING, Any
# Check for Gemini SDK
try:
from google import genai # type: ignore[import-not-found]
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)
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from hother.streamblocks import (
DelimiterFrontmatterSyntax,
Registry,
StreamBlockProcessor,
)
from hother.streamblocks.core.types import BlockEndEvent, BlockErrorEvent, TextContentEvent
from hother.streamblocks_examples.blocks.agent.files import (
FileContent,
FileContentContent,
FileContentMetadata,
FileOperations,
FileOperationsContent,
FileOperationsMetadata,
)
from hother.streamblocks_examples.blocks.agent.memory import Memory, MemoryContent, MemoryMetadata
from hother.streamblocks_examples.blocks.agent.message import Message, MessageContent, MessageMetadata
from hother.streamblocks_examples.blocks.agent.patch import Patch, PatchContent, PatchMetadata
from hother.streamblocks_examples.blocks.agent.toolcall import ToolCall, ToolCallContent, ToolCallMetadata
from hother.streamblocks_examples.blocks.agent.visualization import (
Visualization,
VisualizationContent,
VisualizationMetadata,
)
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from hother.streamblocks.core.models import ExtractedBlock
from hother.streamblocks.core.types import BaseContent, BaseMetadata
def create_system_prompt() -> str:
"""Create the system prompt for multiple block types."""
return dedent("""
You are an AI Software Architect. Use structured blocks to solve software engineering tasks.
All blocks use !!start and !!end delimiters with YAML frontmatter between --- delimiters.
## 1. File Operations Block
For creating, editing, or deleting files (ONLY lists file paths and operations, NOT file content):
!!start
---
id: files_001
block_type: files_operations
description: Creating initial project structure
---
src/main.py:C
src/models/user.py:C
src/utils/helpers.py:C
tests/test_main.py:C
README.md:C
!!end
Where: C=Create, D=Delete
IMPORTANT: File operations blocks ONLY contain file paths and operation types (C/D).
They do NOT contain the actual file content. Use patch blocks or file_content to edit or write content.
## 2. Patch Block
For modifying existing files with diffs:
!!start
---
id: patch_001
block_type: patch
file: src/main.py
description: Add error handling to main function
---
@@ -10,3 +10,6 @@
def main():
result = process_data()
+ if not result:
+ print("Error: Processing failed")
+ return 1
return 0
!!end
Note: The 'file' field is REQUIRED for patch blocks.
## 3. Tool Call Block
For executing analysis or utility tools:
!!start
---
id: tool_001
block_type: tool_call
tool_name: analyze_dependencies
description: Analyze project dependencies
---
directory: ./src
include_dev: true
output_format: json
!!end
Note: The 'tool_name' field is REQUIRED for tool_call blocks.
## 4. Memory Block
For storing/recalling context:
!!start
---
id: memory_001
block_type: memory
memory_type: store
key: project_config
namespace: current_project
---
framework: FastAPI
database: PostgreSQL
cache: Redis
!!end
Note: The 'memory_type' and 'key' fields are REQUIRED for memory blocks.
How to use: always add in memory important information, notably the current tasks and todo.
## 5. Visualization Block
For creating diagrams and charts:
!!start
---
id: viz_001
block_type: visualization
viz_type: diagram
title: System Architecture
format: markdown
---
nodes:
- Frontend
- API Gateway
- Backend Services
- Database
- Cache
edges:
- [Frontend, API Gateway]
- [API Gateway, Backend Services]
- [Backend Services, Database]
- [Backend Services, Cache]
!!end
Note: The 'viz_type' and 'title' fields are REQUIRED for visualization blocks.
## 6. File Content Block
For writing complete file contents (creates or overwrites entire file):
!!start
---
id: file_001
block_type: file_content
file: src/config.py
description: Application configuration file
---
import os
from pydantic import BaseSettings
class Settings(BaseSettings):
app_name: str = "My Application"
debug: bool = os.getenv("DEBUG", "false").lower() == "true"
database_url: str = os.getenv("DATABASE_URL", "sqlite:///./app.db")
class Config:
env_file = ".env"
settings = Settings()
!!end
Note: The 'file' field is REQUIRED for file_content blocks.
## 7. Message Block
For communicating information, status updates, or explanations to the user:
!!start
---
id: msg_001
block_type: message
message_type: info
title: Project Setup Complete
priority: normal
description: Summary of the setup process
---
I've successfully created the FastAPI application structure with user authentication.
The project includes JWT-based authentication, SQLAlchemy models, and all necessary endpoints.
Key features implemented:
- User registration with password hashing
- JWT token generation and validation
- Protected routes requiring authentication
- SQLite database with SQLAlchemy ORM
!!end
Note: The 'message_type' field is REQUIRED for message blocks.
Message types: info, warning, error, success, status, explanation
IMPORTANT:
- Always include 'id' and 'block_type' fields (both required)
- Each block type has specific required fields as shown above
- Use descriptive IDs and clear descriptions
- You can generate multiple blocks to solve complex tasks
- Use !!start and !!end delimiters for all blocks
- The YAML frontmatter must be between --- delimiters
REQUIRED FIELDS SUMMARY:
- ALL blocks: id, block_type
- files_operations: (no additional required fields)
- patch: file (the file path to patch)
- tool_call: tool_name
- memory: memory_type, key
- visualization: viz_type, title
- file_content: file (the file path to write)
- message: message_type (info/warning/error/success/status/explanation)
# General workflow:
1. Create a plan and store it in memory
2. Create or remove files
3. Write or edit the files
IMPORTANT: Communicate as much as possible with the user.
""").strip()
def setup_processor() -> StreamBlockProcessor:
"""Set up a single processor with unified syntax."""
syntax = DelimiterFrontmatterSyntax(
start_delimiter="!!start",
end_delimiter="!!end",
)
# Create registry and register all block types
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
registry.register("patch", Patch)
registry.register("tool_call", ToolCall)
registry.register("memory", Memory)
registry.register("visualization", Visualization)
registry.register("file_content", FileContent)
registry.register("message", Message)
from hother.streamblocks.core.processor import ProcessorConfig
config = ProcessorConfig(lines_buffer=5)
return StreamBlockProcessor(registry, config=config)
async def process_file_operations(block: ExtractedBlock[BaseMetadata, BaseContent]) -> None:
"""Process a file operations block."""
print(block.model_dump_json(indent=2))
async def process_patch(block: ExtractedBlock[BaseMetadata, BaseContent]) -> None:
"""Process a patch block."""
print(block.model_dump_json(indent=2))
async def process_tool_call(block: ExtractedBlock[BaseMetadata, BaseContent]) -> None:
"""Process a tool call block."""
print(block.model_dump_json(indent=2))
async def process_memory(block: ExtractedBlock[BaseMetadata, BaseContent]) -> None:
"""Process a memory block."""
print(block.model_dump_json(indent=2))
async def process_visualization(block: ExtractedBlock[BaseMetadata, BaseContent]) -> None:
"""Process a visualization block."""
print(block.model_dump_json(indent=2))
async def process_file_content(block: ExtractedBlock[BaseMetadata, BaseContent]) -> None:
"""Process a file content block."""
print(block.model_dump_json(indent=2))
async def process_message(block: ExtractedBlock[BaseMetadata, BaseContent]) -> None:
"""Process a message block."""
print(block.model_dump_json(indent=2))
async def get_gemini_response(prompt: str) -> AsyncIterator[Any]:
"""Get Gemini API response stream.
Note: Returns the stream directly - no need for wrapper function!
The StreamBlockProcessor will auto-detect Gemini chunks and use
GeminiAdapter to extract text while preserving original chunks.
"""
# Try GOOGLE_API_KEY first (official), then GEMINI_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"
raise ValueError(msg)
client = genai.Client(api_key=api_key) # type: ignore[attr-defined]
model_id = "gemini-2.5-flash"
# Get system prompt
system_prompt = create_system_prompt()
full_prompt = f"{system_prompt}\n\nUser request: {prompt}"
# Return the stream directly - no need to yield!
return await client.aio.models.generate_content_stream(model=model_id, contents=full_prompt) # type: ignore[attr-defined]
async def main() -> None:
"""Run the architect example."""
print("Gemini AI Architect - Multi-Block Demo")
print("=" * 60)
# Setup single unified processor
processor = setup_processor()
# Example prompts
example_prompts = [
"Create a FastAPI web application with user authentication",
"Design a microservices architecture for an e-commerce platform",
"Add error handling and logging to an existing Python application",
"Create a data pipeline with PostgreSQL and Redis",
"Build a REST API with database models and tests",
]
print("\nExample prompts:")
for i, prompt in enumerate(example_prompts, 1):
print(f"{i}. {prompt}")
# Detect non-interactive mode
is_non_interactive = (
not sys.stdin.isatty() # No TTY (piped/redirected)
or os.getenv("CI") # CI environment
or os.getenv("NON_INTERACTIVE") # Explicit flag
)
# Get user input
if len(sys.argv) > 1:
# Command-line argument provided
arg = sys.argv[1]
if arg.isdigit():
example_num = int(arg)
if 1 <= example_num <= len(example_prompts):
user_prompt = example_prompts[example_num - 1]
else:
user_prompt = example_prompts[0]
else:
# Use full argument as prompt
user_prompt = " ".join(sys.argv[1:])
print(f"\nUsing prompt from command line: {user_prompt}")
elif is_non_interactive:
# Non-interactive mode: use first example
user_prompt = example_prompts[0]
print("\nNon-interactive mode: using default prompt")
else:
# Interactive mode: ask user
user_input = input("\nEnter your request (or 1-5 for examples, Enter for #1): ").strip()
if user_input.isdigit():
example_num = int(user_input)
if 1 <= example_num <= len(example_prompts):
user_prompt = example_prompts[example_num - 1]
else:
user_prompt = example_prompts[0]
elif not user_input:
user_prompt = example_prompts[0]
else:
user_prompt = user_input
print(f"\nProcessing: {user_prompt}")
print("=" * 60)
# Track blocks by type
blocks_by_type = {
"files_operations": 0,
"patch": 0,
"tool_call": 0,
"memory": 0,
"visualization": 0,
"file_content": 0,
"message": 0,
}
try:
# Get response and pass directly to processor
response = await get_gemini_response(user_prompt)
async for event in processor.process_stream(response):
# Skip native Gemini events (we only care about StreamBlocks events)
if processor.is_native_event(event):
continue
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
block_type = block.metadata.block_type
blocks_by_type[block_type] += 1
print(f"\n{'=' * 60}")
print(f"Block extracted: {block_type}")
print(f"{'=' * 60}")
# Process based on type
if block_type == "files_operations":
await process_file_operations(block)
elif block_type == "patch":
await process_patch(block)
elif block_type == "tool_call":
await process_tool_call(block)
elif block_type == "memory":
await process_memory(block)
elif block_type == "visualization":
await process_visualization(block)
elif block_type == "file_content":
await process_file_content(block)
elif block_type == "message":
await process_message(block)
elif isinstance(event, TextContentEvent):
text = event.content.strip()
if text:
print(f"\n{text}")
elif isinstance(event, BlockErrorEvent):
reason = event.reason
print(f"\nBlock rejected: {reason}")
# Show the raw data that was rejected
preview = event.data[:200] if len(event.data) > 200 else event.data
print(f" Raw data preview: {preview!r}")
except Exception as e:
print(f"\nError: {e}")
traceback.print_exc()
# Summary
print(f"\n\n{'=' * 60}")
print("SUMMARY")
print(f"{'=' * 60}")
total_blocks = sum(blocks_by_type.values())
print(f"Total blocks extracted: {total_blocks}")
print("\nBreakdown by type:")
for block_type, count in blocks_by_type.items():
if count > 0:
print(f" - {block_type}: {count}")
if __name__ == "__main__":
asyncio.run(main())
Interactive UI examples (08_ui)
The 08_ui category contains interactive examples that need a terminal to run, so they are not rendered here; run them locally instead:
01_interactive_blocks.py: demonstrates all interactive block types (choices, confirmations, forms) with CLI output.02_interactive_ui_demo.py: a full Textual TUI that renders streaming blocks and lets you respond to them interactively.