Core Primitives

What this is

Before an agent can think, talk, or remember anything, the framework needs a few tiny, boring data shapes that everything else is built out of: how to hold a piece of a message, how to address an agent, how to count tokens, how to name a failure. These shapes live in the kernel (layer L0) — the frozen core. They are pure data: Python dataclasses, Pydantic models, and enums. No network, no disk, no I/O. Just the vocabulary the rest of the framework speaks in.

This page teaches that vocabulary. Five ideas, one at a time:

1. ContentBlock — the lego brick a message is made of.
2. ChatMessage + Role — one turn of a conversation.
3. AgentId / TopicId — postal addresses for routing.
4. Usage — the token meter.
5. Errors — the family tree of things that can go wrong.

1. ContentBlock — the lego brick of a message

Plain English: a ContentBlock is the smallest self-contained piece of a message. One block holds one kind of thing — some text, or an image, or a request to call a tool, or the result of that tool.

Analogy: content blocks are lego bricks. A message isn't a single molded object; it's a list of bricks snapped together. A reply from the model might be [a thinking brick, a text brick, a tool-call brick]. Each brick knows what it is.

A block is multimodal-ready and self-describing: every block carries a type field (a fixed string literal like "text" or "image") so code can tell the bricks apart. This collection of brick types forms a discriminated union — a fancy term meaning "one of these N classes, and the type field tells you which."

The full set of bricks

There are twelve block types in the curated union. Here is the whole drawer:

There is also one special non-union brick:

• UnknownBlock (type="unknown") — a lossless carrier. When the framework reads a block whose type it doesn't recognize (e.g. a newer service produced it), it wraps the raw payload in UnknownBlock instead of throwing it away. This keeps mixed-version deployments from silently corrupting data.

from substrate.kernel.core.content import TextBlock

texts = [b.text for b in message.content if isinstance(b, TextBlock)]

How bricks compose into a message

A whole agent message, tool result, or pub/sub event carries a list[ContentBlock]. The model's turn flows in, blocks get appended, and the result is one ordered list:

The contract (trimmed)

class TextBlock(BaseModel):
    type: Literal["text"] = "text"
    text: str
    model_config = {"frozen": True}
    def to_text_repr(self) -> str: ...

class ToolUseBlock(BaseModel):
    type: Literal["tool_use"] = "tool_use"
    call_id: str
    tool_name: str
    arguments: JsonObject = Field(default_factory=dict)

class ToolResultBlock(BaseModel):
    type: Literal["tool_result"] = "tool_result"
    call_id: str
    name: str = ""
    content: list["ContentBlock"] = Field(default_factory=list)  # nested blocks!
    is_error: bool = False

# The union itself — discriminated on the `type` field
ContentBlock = Annotated[
    TextBlock | ImageBlock | AudioBlock | VideoBlock | DocumentBlock
    | DataBlock | CodeBlock | ErrorBlock | ToolUseBlock | ToolResultBlock
    | ThinkingBlock | UIResourceBlock,
    Field(discriminator="type"),
]

2. ChatMessage + Role — one turn of a conversation

Plain English: a ChatMessage is a single turn in a conversation, tagged with who said it and carrying a list of content blocks for what they said.

Analogy: if blocks are lego bricks, a ChatMessage is one assembled lego model with a name tag on it — "this was built by the user", "this was built by the assistant".

The "who" is the role, drawn from the Role enum:

class Role(StrEnum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"

class ChatMessage(BaseModel):
    role: str                                # follows Role, but accepts any string
    content: list["ContentBlock"] = Field(default_factory=list)
    name: str | None = None                  # which participant, for multi-agent
    metadata: dict[str, Any] = Field(default_factory=dict)
    model_config = {"frozen": True}

The name field identifies which participant spoke in a multi-agent conversation (an agent's name, a user handle) — useful for attribution and routing when more than two parties are in the room.

3. AgentId & TopicId — addresses for routing

Plain English: these are the addresses the framework uses to know where to deliver a message. An AgentId points at one specific agent instance. A TopicId points at a pub/sub channel many subscribers can listen to.

Analogy: an AgentId is a postal address for one mailbox (this exact agent). A TopicId is a radio frequency — anyone tuned in hears what's broadcast.

Both are built the same way: a type (a category) plus a key, and a namespace that keeps different tenants from colliding on shared infrastructure.

AgentId — addressing one agent

@dataclass(frozen=True, slots=True)
class AgentId:
    type: str                 # the agent's role name, e.g. "researcher"
    key: str                  # unique instance id within that type
    namespace: str = "default"

    def __str__(self) -> str:
        # "type/key", or "namespace/type/key" when not default
        ...

    @classmethod
    def generate(cls, agent_type: str, *, namespace: str = "default") -> AgentId:
        ...  # key becomes a random uuid4 hex

• type is the role name — what kind of agent this is.
• key uniquely identifies this instance within that type (a session id, or a generated UUID via AgentId.generate("researcher")).
• namespace (default "default") scopes the agent to one tenant so two tenants can both have a researcher/session-1 without their messages crossing wires on shared infra like Redis pub/sub.

TopicId — addressing a channel

@dataclass(frozen=True, slots=True)
class TopicId:
    type: str                 # topic category, e.g. "agent.progress"
    source: str = "default"   # origin scope — a run_id, session, pipeline
    namespace: str = "default"

TopicId is the same idea for broadcasts: type is the category of event, and source scopes it to a particular origin. The standard conventions are:

• agent.progress / <run_id> — all progress events for one execution run.
• agent.stream / <run_id> — the token stream for a specific run.

4. Usage — the token meter

Plain English: Usage is the receipt for a single LLM call — how many tokens went in, how many came out, and two special sub-counts that let you compute cost accurately.

Analogy: it's the itemized utility bill for one model call. The total matters, but the line items (cached, reasoning) are billed at different rates.

The two "cached" and "reasoning" counts are broken out, not added on top — they are slices of input and output respectively, exposed so you can attribute cost precisely (cache reads are cheaper; reasoning tokens are part of output).

@dataclass(frozen=True, slots=True)
class Usage:
    input_tokens: int = 0
    cached_tokens: int = 0       # subset of input_tokens (provider prompt cache)
    output_tokens: int = 0
    reasoning_tokens: int = 0    # subset of output_tokens (extended thinking)

    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens   # NOT cached/reasoning

    def __add__(self, other: "Usage") -> "Usage":
        ...   # field-wise sum

total = Usage()
for step in run_steps:
    total = total + step.usage
print(total.total_tokens)   # input + output across every call

5. Errors — the family tree of failures

Plain English: when something goes wrong inside the runtime, it raises a typed error — a specific class, not a vague exception — so callers can catch exactly the failure they care about.

Analogy: it's a family tree. Everything descends from one ancestor, KernelError. Catch the ancestor to catch the whole family; catch a specific descendant to handle just that one case.

class KernelError(Exception):
    """Base class for all agent substrate kernel errors."""

class AgentCrashError(KernelError):
    def __init__(self, message: str, *, run_id: str, agent_id: AgentId) -> None:
        super().__init__(message)
        self.run_id = run_id
        self.agent_id = agent_id

class ConcurrentAppendError(KernelError):
    def __init__(
        self, message: str, *, run_id: str, expected_seq: int, actual_seq: int
    ) -> None:
        super().__init__(message)
        self.run_id = run_id
        self.expected_seq = expected_seq
        self.actual_seq = actual_seq

Where this lives

Next: The LLM Contract — how these primitives flow into and out of a model: the LLMClient Protocol that takes list[ChatMessage], streams back ContentBlocks, and reports Usage.