Messaging

The one-line version

Messaging is how agents talk to each other and to the user. Everything an agent sends — a chat turn, a tool request, a "please pause" signal — is wrapped in a single envelope called a Message, handed to the runtime, and delivered to a target. Separately, while an agent is working, it leaks a live feed of tiny updates (stream deltas) so the UI can show typing in real time.

This page covers three things, all pure data types with zero I/O:

1. The Message envelope and its payloads — durable, routable, one per turn.
2. Streaming deltas — ephemeral, real-time, many per turn.
3. The Event envelope — the versioned wire format both buses share.

1. The Message envelope

A Message is the unit of agent-to-agent communication. Every send or publish wraps a payload in one. The runtime reads the address, delivers it, and (for a direct send) returns the recipient's reply.

Here is the actual type, trimmed to the fields you'll touch:

class Message(BaseModel):
    target: AgentId | TopicId          # required — no destination, no delivery
    payload: Payload                   # the letter (any PayloadBase subclass)
    sender: AgentId | None = None      # return address; None = anonymous/bootstrap

    correlation_id: str = ...          # ties one logical conversation together
    causation_id: str | None = None    # the exact message that triggered this one
    reply_to: str | None = None        # run_id of the asker; set by RunContext.ask()

    metadata: dict[str, str] = {}      # free-form string notes for transports
    id: str = ...                      # time-sortable hex id — dedup / idempotency
    created_at: datetime = ...
    schema_version: int = 1

    @property
    def is_broadcast(self) -> bool:    # True when target is a TopicId (fan-out)
        ...

Addressing uses two routing keys from kernel/core/identity.py (covered in the identity page; here they are just addresses):

• AgentId(type, key, namespace) — a direct address to one agent instance. str form: "researcher/abc123".
• TopicId(type, source, namespace) — a broadcast address to a topic, for pub/sub fan-out. When target is a TopicId, message.is_broadcast is True.

Payload types — the letter inside

The envelope is dumb; the payload is the meaning. Every payload subclasses PayloadBase and carries a kind string so the runtime can deserialize it safely. These are the built-ins:

class ChatPayload(PayloadBase):
    kind: Literal["chat"] = "chat"
    message: ChatMessage               # role-tagged turn of ContentBlocks

class DataPayload(PayloadBase):
    kind: Literal["data"] = "data"
    data: JsonObject                   # any JSON-serializable dict

class ControlPayload(PayloadBase):
    kind: Literal["control"] = "control"
    signal: str                        # e.g. "pause", "cancel", "handoff"
    data: JsonObject = {}

Subscriptions — who's listening on a topic

A Subscription is a tiny record that says "this agent is listening to this topic." The runtime keeps these so a TopicId broadcast knows where to fan out.

class Subscription(BaseModel):
    id: str = ...              # unique subscription id
    topic: TopicId            # the topic being listened to
    agent_id: AgentId         # the agent that subscribed

2. correlation_id and reply_to — how a conversation hangs together

Two fields do the relationship work. They are easy to mix up, so define them once:

• correlation_id — the conversation id. Every message in one logical back-and-forth shares the same correlation_id. It's the tracking sticker that says "all these envelopes belong to the same story."
• reply_to — the return-run address. When an agent asks and wants the answer routed straight back to its own run, RunContext.ask() stamps the asker's run_id here. The reply travels back to exactly that run.
• causation_id — the direct parent. It names the single message that caused this one. correlation_id groups the whole tree; causation_id is the one edge up.

Ask → reply, step by step

This is the request/response pattern: agent A asks agent B a question and waits for the answer.

3. Streaming deltas — the live ticker

The Message envelope is for finished turns. But an LLM produces its answer token by token, and users want to watch it happen. That's what stream.py is for: small, frozen, fire-and-forth events emitted during a run.

There are two independent channels:

1. Token stream — the words and thoughts of the agent currently speaking.
2. Progress stream — structured "what step am I on" events from every agent in the supervision tree.

Token stream events

class TextDelta(BaseModel):          # incremental visible text
    text: str
    agent_id: AgentId | None = None
    run_id: str = ""
    seq: int = 0                     # strictly increasing within a run

class CompletionEvent(BaseModel):    # the final, whole response
    content: list[ContentBlock]
    usage: Usage = ...               # tokens / cost for this turn
    agent_id: AgentId | None = None
    run_id: str = ""
    seq: int = 0

class StreamDone(BaseModel):         # the END card
    reason: str = "complete"

What a streaming turn looks like

Progress stream — the supervision-tree heartbeat

While the token stream is one speaker, the progress stream is everybody. Every agent in a run — parent and children — publishes AgentProgress events to one shared topic, TopicId("agent.progress", run_id). The UI subscribes once and rebuilds the whole tree from each event's agent_id, parent_id, and depth.

class AgentStep(StrEnum):            # the standard step names
    STARTED = "started"
    THINKING = "thinking"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"
    HANDOFF = "handoff"
    PAUSED = "paused"
    DONE = "done"
    ERROR = "error"

class AgentProgress(BaseModel):
    agent_id: AgentId                # who emitted it
    step: AgentStep                  # which standard step
    content: str                     # human-readable detail
    run_id: str = ""
    parent_id: AgentId | None = None # who spawned this agent
    depth: int = 0                   # nesting level in the tree
    seq: int = 0                     # ordering within the run
    ts: datetime = ...               # wall-clock — for display only

4. The Event envelope

Underneath both buses sits one shared wire contract: the Event. The in-process kernel pub/sub and the distributed infrastructure bus (Redis, NATS, Kafka) all carry Event objects, so there is exactly one event format across every transport.

class Event(BaseModel):
    id: str = ...                    # unique — enables consumer dedup
    type: str                        # e.g. "agent.started", "tool.called"
    source: str                      # str(AgentId(...)) or a service name
    correlation_id: str = ""         # ties all events in one run together
    schema_version: int = 1          # bump when `data` shape changes
    data: JsonObject = {}            # event-specific payload
    ts: datetime = ...

    @classmethod
    def create(cls, event_type, *, source, data=None, correlation_id="", ...):
        ...                          # source accepts an AgentId or a string

Two Protocols abstract the transport so the same producer code works in-process or over Redis:

class EventPublisher(Protocol):
    async def publish(self, event: Event, *, topic: str = "") -> None: ...

class EventSubscriber(Protocol):
    async def subscribe(self, topic: str, handler: EventHandler) -> str: ...
    async def unsubscribe(self, subscription_id: str) -> None: ...
    def stream(self, topic: str) -> AsyncIterator[Event]: ...

Where this lives

Next: Tools, Skills & Approval — what an agent can actually do once it has decided to act, and how a risky action gets a human's sign-off.