Skip to content

Your First Custom Block

Built-in example blocks are fine for a demo, but the point of StreamBlocks is extracting your domain objects from a stream. This page defines a task block from scratch.

Define metadata and content models

A block type is a Block[TMetadata, TContent] where the metadata model extends BaseMetadata and the content model extends BaseContent. Both are Pydantic models, so you get validation and typed access for free:

class TaskMetadata(BaseMetadata):
    """Custom metadata for task blocks."""

    id: str
    block_type: str
    title: str = "Untitled"
    priority: str = "normal"


class TaskContent(BaseContent):
    """Custom content for task blocks."""

    description: str = ""

    @classmethod
    def parse(cls, raw_text: str) -> "TaskContent":
        return cls(raw_content=raw_text, description=raw_text.strip())


TaskBlock = Block[TaskMetadata, TaskContent]

Two things to note:

  • Metadata fields come from the block's metadata section (YAML frontmatter here). Defaults make fields optional in the stream.
  • Content is produced by the parse() classmethod, which receives the raw text between the metadata section and the closing delimiter. raw_content always preserves the original text.

Register and process

registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("task", TaskBlock)
processor = StreamBlockProcessor(registry)

text = "!!start\n---\nid: task-1\nblock_type: task\ntitle: Fix bug\npriority: high\n---\nFix the login issue\n!!end"
stream = simple_text_stream(text)

async for event in processor.process_stream(stream):
    if isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block:
            print("\nExtracted Task Block:")
            print(block.model_dump_json(indent=2))

View source on GitHub

This example uses the delimiter frontmatter syntax: !!start, a YAML metadata section between --- markers, free-form content, then !!end:

!!start
---
id: task-1
block_type: task
title: Fix bug
priority: high
---
Fix the login issue
!!end

Where to go from here