OpenAI, Anthropic & Gemini

How to extract blocks from real LLM provider streams. Each provider ships as an optional extension that registers its input adapter for auto-detection the moment you import it.

The pattern

The recipe is identical for all providers:

1. Install the provider extra.
2. Make sure the extension is imported (importing the provider SDK's adapter module registers it).
3. Pass the SDK's stream object directly to processor.process_stream(): no wrapper generators.

Provider	Install	Auto-registration import	Adapter	Factory
Gemini	pip install "streamblocks[gemini]"	import hother.streamblocks.extensions.gemini	GeminiInputAdapter	create_gemini_processor(registry)
OpenAI	pip install "streamblocks[openai]"	import hother.streamblocks.extensions.openai	OpenAIInputAdapter	create_openai_processor(registry)
Anthropic	pip install "streamblocks[anthropic]"	import hother.streamblocks.extensions.anthropic	AnthropicInputAdapter	create_anthropic_processor(registry)

pip install "streamblocks[all-providers]" installs all three SDKs at once.

Each factory takes a Registry and returns a ProtocolStreamProcessor pre-configured with that provider's input adapter and the default StreamBlocks output adapter:

from hother.streamblocks.extensions.gemini import create_gemini_processor

processor = create_gemini_processor(registry)

The examples below use StreamBlockProcessor instead, which additionally passes the original provider chunks through (emit_original_events=True), so you keep access to usage metadata, finish reasons, and other provider-specific fields.

Gemini: auto-detection

With the google-genai SDK, hand the stream straight to the processor; the adapter is detected from the first chunk (the SDK's chunk classes live under the registered google.genai module prefix):

# 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]

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

processor.is_native_event(event) identifies passed-through Gemini chunks without importing Gemini types in your handler code.

View source on GitHub

OpenAI: explicit adapter

You can skip auto-detection and pass the adapter explicitly, useful when the first chunk may be ambiguous or you want zero detection overhead:

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

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

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

The adapter extracts text from choices[0].delta.content and reports completion when finish_reason is set.

View source on GitHub

Anthropic: event-based streams

Anthropic streams discrete event types rather than uniform chunks. The adapter categorizes content_block_delta events as text and passes the rest (message_delta, message_stop, …) through, so nothing is lost:

# 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

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

View source on GitHub

Handling mixed event streams

With emit_original_events=True your event loop sees both native provider events and StreamBlocks events in the same stream. Plain isinstance checks (or processor.is_native_event()) keep the two apart, and provider metadata stays available:

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

View source on GitHub

A complete provider workflow

A fuller Gemini demo registers several block types behind one frontmatter syntax and lets the model emit file operations, file contents, and messages in a single response:

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

# 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}")

View source on GitHub

Prompting for blocks

Models follow block syntax reliably when the prompt shows the exact format and explicitly forbids markdown code fences around it. See the prompts embedded in the examples above for working templates.

Next steps

• Adapters: write an adapter for a provider that isn't built in.
• Events: everything your event loop can react to.
• Extensions reference: full adapter and factory API per provider.