Skip to content

Defining Custom Blocks

This guide shows three ways to turn raw block content into typed Python objects, from a hand-written parse() to fully schema-driven structured output.

Define metadata and content models

A block type is three pieces: a metadata model (inherits BaseMetadata), a content model (inherits BaseContent), and the Block class binding them. Override the parse() classmethod on the content model to extract structure from the raw body; always keep raw_content populated with the original text:

# Custom content models for this example
class TaskMetadata(BaseMetadata):
    """Metadata for task blocks."""

    id: str
    block_type: str
    title: str = "Untitled Task"
    priority: str = "medium"
    assignee: str | None = None
    due_date: str | None = None
    tags: list[str] = Field(default_factory=list[str])
    status: str = "todo"


class TaskContent(BaseContent):
    """Content for task blocks."""

    description: str = ""
    subtasks: list[str] = Field(default_factory=list[str])

    @classmethod
    def parse(cls, raw_text: str) -> "TaskContent":
        """Parse task content from raw text."""
        lines = raw_text.strip().split("\n")
        if not lines:
            return cls(raw_content=raw_text, description="")

        description = lines[0]
        subtasks: list[str] = []

        for line in lines[1:]:
            stripped = line.strip()
            if stripped.startswith(("- ", "* ")):
                subtasks.append(stripped[2:])

        return cls(raw_content=raw_text, description=description, subtasks=subtasks)


# Create the block type
TaskBlock = Block[TaskMetadata, TaskContent]

Register the block type and process as usual:

# Create delimiter frontmatter syntax for tasks
# Using standard !!start/!!end delimiters
task_syntax = DelimiterFrontmatterSyntax(
    start_delimiter="!!start",
    end_delimiter="!!end",
)

# Create type-specific registry and register block
registry = Registry(syntax=task_syntax)
registry.register("task", TaskBlock)

If parse() raises, the block is rejected with a BlockErrorEvent instead of a BlockEndEvent; see Error Handling.

Parse JSON or YAML with decorators

When the block body is JSON or YAML, skip the boilerplate: the @parse_as_json() and @parse_as_yaml() decorators generate parse() for you. They load the text, then pass the resulting dict as keyword arguments to your content model.

from hother.streamblocks import ParseStrategy, parse_as_json, parse_as_yaml

Both decorators take the same keyword-only arguments:

Argument Default Description
strategy ParseStrategy.PERMISSIVE What to do when parsing or validation fails
handle_non_dict True Wrap non-dict values (scalars, lists) as {"value": ...}

PERMISSIVE: fall back to raw content

With ParseStrategy.PERMISSIVE, malformed input never rejects the block; the model is built with only raw_content set and your typed fields keep their defaults:

@parse_as_yaml(strategy=ParseStrategy.PERMISSIVE)
class ConfigContent(BaseContent):
    """Configuration content parsed from YAML.

    With PERMISSIVE strategy, malformed YAML falls back to raw_content.
    """

    app_name: str | None = None
    version: str | None = None
    debug: bool | None = None
    port: int | None = None
    features: dict[str, bool] = Field(default_factory=dict)


class ConfigMetadata(BaseMetadata):
    """Metadata for configuration blocks."""

    block_type: str = "config"
    environment: str | None = None


# Create the block type
ConfigBlock = Block[ConfigMetadata, ConfigContent]

STRICT: reject on parse errors

With ParseStrategy.STRICT, parse and validation errors propagate, so malformed content produces a BlockErrorEvent instead of a silently degraded block:

@parse_as_json(strategy=ParseStrategy.STRICT)
class APIResponseContent(BaseContent):
    """API response parsed from JSON.

    With STRICT strategy, malformed JSON raises an exception.
    """

    status: int
    message: str
    data: dict[str, Any] = Field(default_factory=dict)
    errors: list[str] = Field(default_factory=list)


class APIMetadata(BaseMetadata):
    """Metadata for API response blocks."""

    block_type: str = "api_response"
    endpoint: str | None = None


# Create the block type
APIBlock = Block[APIMetadata, APIResponseContent]

Use STRICT when downstream code depends on the typed fields; use PERMISSIVE when you prefer to keep the raw text and recover manually.

Non-dict content

JSON/YAML bodies are not always mappings. With handle_non_dict=True (the default) a scalar or list body is wrapped as {"value": ...}, so a value field on your model captures it. With handle_non_dict=False, non-dict input is discarded; only raw_content is set and typed fields keep their defaults:

@parse_as_yaml(strategy=ParseStrategy.PERMISSIVE, handle_non_dict=True)
class ScalarWrapperContent(BaseContent):
    """Content that wraps scalar YAML values in {'value': ...}."""

    value: str | int | float | bool | None = None


@parse_as_yaml(strategy=ParseStrategy.PERMISSIVE, handle_non_dict=False)
class ScalarNoWrapContent(BaseContent):
    """Content that doesn't wrap scalar values (will fail on scalars)."""

    message: str | None = None

Structured-output blocks from a schema

To validate block content against an arbitrary Pydantic schema (the streaming equivalent of structured output), generate the whole block type from the schema. The examples package ships a create_structured_output_block factory that builds metadata and content models around any BaseModel:

class PersonSchema(BaseModel):
    """Simple person data schema."""

    name: str
    age: int
    email: str
    city: str

# Create the specialized block type
PersonBlock = create_structured_output_block(  # noqa: N806
    schema_model=PersonSchema,
    schema_name="person",
    format="json",
    strict=True,  # Strict validation
)

# Create syntax and registry
# The syntax will extract metadata and content classes from the block automatically
registry = Registry(syntax=DelimiterFrontmatterSyntax())
registry.register("person", PersonBlock)

# Create processor
processor = StreamBlockProcessor(registry)

strict=True rejects blocks whose payload fails schema validation; strict=False falls back to raw_content. The factory also supports YAML payloads:

# Create the block with YAML parsing
ConfigBlock = create_structured_output_block(  # noqa: N806
    schema_model=ConfigSchema,
    schema_name="config",
    format="yaml",  # Using YAML!
    strict=True,
)

The factory lives in the examples package (blocks/agent/structured_output.py), not the core library; copy it into your project and adapt it.

Next steps