Skip to content

Blocks & Registry

A block is a typed unit of structured content extracted from a stream. The registry maps block type names to block classes and tells the processor which wire format to detect. This page covers the Block model, its metadata/content halves, and how the Registry ties everything together.

The Block model

Block[TMetadata, TContent] is a Pydantic model with exactly two fields:

class FileOperations(Block[FileOperationsMetadata, FileOperationsContent]):
    """File operations block."""
  • metadata: TMetadata: parsed block metadata (from the header or frontmatter section).
  • content: TContent: parsed block content (the body between the markers).

The generic parameters give you type-safe access: block.metadata.description and block.content.operations are fully typed. See blocks/agent/files.py for the complete FileOperations definition used throughout the examples.

Metadata: BaseMetadata

Every metadata model inherits from BaseMetadata, which requires two fields:

Field Type Description
id str Block identifier, unique within the stream
block_type str Type name used to look up the block class in the registry

Add your own fields on top:

class FileOperationsMetadata(BaseMetadata):
    """Metadata for file operations blocks."""

    block_type: Literal["files_operations"] = "files_operations"
    description: str | None = None

Content: BaseContent

Content models inherit from BaseContent, which has a single required field, raw_content: str: the original unparsed body text, always preserved. Parsing happens in the parse() classmethod:

class BaseContent(BaseModel):
    raw_content: str

    @classmethod
    def parse(cls, raw_text: str) -> Self:
        return cls(raw_content=raw_text)

The default implementation just stores the raw text. Override parse() to turn the body into structured fields; see Defining Custom Blocks.

Block vs ExtractedBlock

You define block types with Block; the processor delivers ExtractedBlock instances. ExtractedBlock[TMetadata, TContent] extends Block with extraction metadata:

Field Description
syntax_name Name of the syntax class that extracted the block
raw_text Original raw text of the whole block, markers included
line_start / line_end Line numbers in the stream
hash_id Hash-based ID, useful for deduplication and caching

BlockEndEvent.get_block() returns the typed ExtractedBlock:

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

The Registry

A Registry holds exactly one syntax and a mapping from block type names to block classes. All constructor arguments are keyword-only:

registry = Registry(
    syntax=Syntax.DELIMITER_PREAMBLE,  # default
    blocks={"files_operations": FileOperations},  # optional bulk registration
    metadata_failure_mode=MetadataValidationFailureMode.ABORT_BLOCK,  # default
)
Parameter Default Description
syntax Syntax.DELIMITER_PREAMBLE A Syntax enum member or a BaseSyntax instance
logger stdlib logger Any object with debug/info/warning/error/exception methods
blocks None Dict of block_type -> block_class for bulk registration
metadata_failure_mode ABORT_BLOCK What to do when early metadata validation fails; see Validation

Registering block types

register() maps a block type name to a class, optionally with validators:

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

A validator is a callable that takes the ExtractedBlock and returns bool; a False result rejects the block with a BlockErrorEvent.

Unregistered block types

You do not have to register anything. A block whose block_type has no registered class is still extracted, using plain BaseMetadata and BaseContent; metadata fields are parsed and the body lands in raw_content:

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

This minimal API is handy for prototyping before you commit to typed models. See the full example in Basics examples.

The block round-trip

Block definitions are self-describing: the docstring, metadata fields, content format, and __examples__ carry everything needed to explain the block to a model. Registry.to_prompt() turns the registry into instructions telling the model which blocks to emit and in what format; the model streams that text back; and the same registry parses it into typed ExtractedBlock instances. Because both directions are driven by one set of definitions, the prompt and the parser never drift apart. See Generating Prompts for the how-to.

Next steps