Agent Types

What this is

The kernel (L0) defines what an agent is in a single sentence — an object with an id and one method, async run(ctx, inbox). But a Protocol can't think. It's a shape, an empty promise. This page is about the classes that actually fill that shape with intelligence — the concrete agents that live at layer L1 (agents/).

Every agent here is a normal Python class that satisfies the kernel Agent Protocol. There is no base class to inherit, no @agent decorator. If your class has an id attribute and an async def run(self, ctx, inbox), the runtime will happily run it. The classes on this page are the batteries-included ones Agent Substrate ships with.

These classes may import from kernel (the layer below) but never from capabilities or fabric (the layers above). That's the one architectural rule that keeps the agent loop reusable everywhere.

The cast, at a glance

ReActAgent — the centerpiece

What & why: ReActAgent runs the ReAct pattern — Reason + Act. The model reasons about the task, decides to act by calling a tool, sees the tool's result, reasons again, and keeps going until it has a final answer (or runs out of its iteration budget).

From run() to the loop

A run touches three methods, each smaller and more focused than the last:

run(ctx, inbox) is tiny. It stamps a couple of context variables (so tools like the Kanban board know which agent and label they belong to), then walks the inbox one message at a time. Between messages it calls ctx.check() — the cooperative cancellation point that lets a parent or a deadline interrupt it cleanly.

async def run(self, ctx: RunContext, inbox: list[Message]) -> None:
    _task_agent_id.set(str(self.id))
    _task_agent_label.set(self.name)
    for msg in inbox:
        ctx.check()
        await self._handle_message(ctx, msg)

_handle_message(ctx, msg) sets up the conversation. It derives a session_id (the conversation thread — falls back to ctx.run_id if the message carries no correlation_id), loads the prior history, converts the incoming message to a chat turn, and hands the assembled messages list to the loop.

async def _handle_message(self, ctx: RunContext, msg: Message) -> None:
    session_id = msg.correlation_id or ctx.run_id
    _task_thread_id.set(session_id)
    _task_parent_agent_id.set(msg.metadata.get("parent_agent_id") or None)

    history_messages = await load_history(self._context, self.id, session_id)
    user_turn = message_to_chat(msg)
    messages: list[ChatMessage] = history_messages + [user_turn]

    await self._react_loop(ctx, msg, session_id, messages, len(history_messages))

The ReAct loop itself

_react_loop is the engine. It iterates up to max_iterations times. Each pass:

1. Compact a view of the messages (more on this below).
2. Call the model via await ctx.llm(...).
3. Record the assistant turn and (if any) charge the execution budget.
4. Look for tool calls. No tool calls → the model is done; break out.
5. Run each tool via await ctx.tool(name, **args), collect the results, append them as a TOOL turn, and loop again.

Compaction runs before every model call

This is the detail people miss. Tool results can be huge (a 5,000-line file, a web page), and they pile up across iterations. So before each ctx.llm() call, the agent compacts:

# Compact a *view* of messages here and keep the full list intact for persistence.
llm_messages = await self._context.pipeline.compact(messages)
resp = await ctx.llm(llm_messages, options=options)

The key word is view. compact() returns a trimmed copy that goes to the model; the full untrimmed messages list is what gets persisted afterwards. The model sees a manageable window; the conversation record stays complete. See the compaction pipeline for the strategies.

Tool dispatch goes through ctx.tool()

The agent never calls a tool's execute() directly. It asks the context:

tool_calls = [b for b in resp.content if isinstance(b, ToolUseBlock)]
if not tool_calls:
    break

results: list[ToolResultBlock] = []
for tc in tool_calls:
    ctx.check()
    inv_result = await ctx.tool(tc.tool_name, **tc.arguments)
    results.append(
        ToolResultBlock(
            call_id=tc.call_id,
            content=[TextBlock(text=inv_result.text or "")],
            is_error=inv_result.status != "ok",
        )
    )
messages.append(ChatMessage(role=Role.TOOL, content=results))

Routing through ctx.tool() is what makes a tool call journaled — recorded once so that on a crash-and-replay it returns the cached result instead of re-running. That's the durability guarantee from Durability, and the agent gets it for free just by calling ctx.tool() instead of the tool object.

The iteration cap is a hard budget

The loop is a Python for ... range(self._max_iterations). If it completes all iterations without breaking (i.e. the model kept calling tools and never settled on an answer), the for/else fires:

for _ in range(self._max_iterations):
    ...
else:
    from substrate.kernel.core.errors import BudgetExhaustedError
    raise BudgetExhaustedError(
        f"Agent reached max iterations limit ({self._max_iterations})"
    )

A runaway agent fails loudly with BudgetExhaustedError rather than looping forever and burning tokens. When the loop does break normally, the agent persists the new turns and delivers the answer:

new_turns = messages[n_loaded:]
await persist_turns(self._context, self.id, session_id, ctx.run_id, new_turns)
ans = final_text(messages)
await deliver(ctx, msg, {"text": ans}, sender=self.id, output_topic=self._output_topic)

A run, end to end

OrchestratorAgent — the manager

What & why: Some jobs are too big or too varied for one agent. An OrchestratorAgent holds a roster of sub-agents, each declared as a SubAgentConfig. The model decides which specialist to delegate to by calling a tool whose name maps to that sub-agent. The orchestrator then spawns the sub-agent as its own run and asks it for the answer.

Sub-agents become delegation tools

Each SubAgentConfig carries the sub-agent, a description, an ask_timeout, and a Priority:

@dataclass
class SubAgentConfig:
    agent: Agent
    description: str = ""
    ask_timeout: float = 120.0
    priority: Priority = Priority.NORMAL

At message time, _build_tools() synthesizes one _DelegateTool per sub-agent, named handoff_<key>. These are fake tools — their execute() deliberately raises if called. They exist only so the model sees them in GenerationOptions(tools=...) and can pick one. The actual work is done by the orchestrator, not the tool.

def _build_tools(self) -> list[AnyTool]:
    return [
        _DelegateTool(
            name=f"handoff_{cfg.agent.id.key}",
            description=cfg.description or f"Delegate to the {cfg.agent.id.key} sub-agent",
        )
        for cfg in self._sub_agents
    ]

Spawn, ask, and the SpawnTracker

The orchestrator runs its own ReAct-style loop. When the model emits a delegation call, the orchestrator:

1. Looks up the matching SubAgentConfig.
2. Calls spawn_tracker.acquire(...) — the SpawnTracker enforces the SpawnBudget (a headcount cap with priority preemption) so you can't fork an unbounded swarm.
3. Builds a boot Message for the sub-agent, stamping its own id as parent_agent_id so the sub-agent's task board nests under it in the UI.
4. await ctx.spawn(...) starts the sub-agent run, then await ctx.ask(handle, ..., timeout=...) waits for the reply.
5. Releases the tracker slot in a finally — slot freed even on error.

spawn_tracker.acquire(cfg.agent.id, priority=cfg.priority)
task_text = dispatch.arguments.get("task", str(dispatch.arguments))
boot_msg = Message(
    target=cfg.agent.id, sender=self.id,
    payload=ChatPayload(message=ChatMessage(role=Role.USER,
                                            content=[TextBlock(text=str(task_text))])),
    correlation_id=session_id,
    metadata={"parent_agent_id": str(self.id)},
)
try:
    handle = await ctx.spawn(cfg.agent.id, boot=boot_msg)
    outcome = await ctx.ask(handle, boot_msg, timeout=cfg.ask_timeout)
finally:
    spawn_tracker.release(cfg.agent.id)

await rt.register(orchestrator)  # sub-agents included automatically

UserProxyAgent — the human in the loop

What & why: Sometimes an agent needs a person, not a model — "should I really send this $10,000 invoice?" The UserProxyAgent is an agent that represents the human. It has no model and no tools. Its whole job is to receive a question, surface it, go to sleep for free, and wake up when a human answers.

async def _handle_message(self, ctx: RunContext, msg: Message) -> None:
    cid = msg.correlation_id
    question_text = self._extract_text(msg)
    await ctx._log("hitl.question", {"correlation_id": cid, "text": question_text})

    signal_name = f"human_reply:{cid}"
    human_payload = await ctx.sleep_until_signal(signal_name)

    if msg.reply_to:
        await ctx.reply(msg, human_payload)

The magic line is await ctx.sleep_until_signal(...). The run goes SUSPENDED — zero RAM, zero CPU, just rows in storage. The serving layer surfaces the logged hitl.question to the user (over SSE), and when the human replies it calls SignalBus.signal(run_id, "human_reply:<cid>", ). That signal wakes the proxy, which relays the answer back to the agent that asked via ctx.reply.

InformationAgent and PersonalFeedAgent — the demo pair

Together they demonstrate a producer/consumer feed with no polling:

InformationAgent is the producer. On boot it summarizes any inbox item via the shared summarize helper, emits the result to its output_topic, then sleeps in a loop waiting for the next new_source_item signal:

async def run(self, ctx: RunContext, inbox: list[Message]) -> None:
    for msg in inbox:
        ctx.check()
        await self._process_item(ctx, msg)
    while True:
        ctx.check()
        item_payload = await ctx.sleep_until_signal(self._source_signal)
        await self._process_item_from_signal(ctx, item_payload)

PersonalFeedAgent is the consumer. On its first wake it subscribes to its topics via ctx.follow(topic) (the subscription is durable in the FollowGraph, so it survives across runs), curates each delivered item against the user's preferences, logs a feed.curated entry, and sleeps until the next delivery:

async def run(self, ctx: RunContext, inbox: list[Message]) -> None:
    if not self._subscribed:
        for topic in self._follow_topics:
            await ctx.follow(topic)
        self._subscribed = True
    for msg in inbox:
        ctx.check()
        await self._curate(ctx, msg)
    await ctx.sleep_until_signal("new_feed_item")

The takeaway: producers emit, consumers follow, and the runtime's fan-out does the delivery — the same InboxProtocol / FollowGraph contracts described in the runtime page.

The shared helpers: _loop.py

Every agent above repeats the same conversation chores: turn a Message into a chat turn, load history, persist new turns, extract the final answer, deliver the result. Rather than copy-paste, those live in one small module, agents/core/_loop.py, and are imported by all the agents.

The factory: convenience constructors

Wiring a ReActAgent by hand means assembling a Toolbox, a ContextConfig, and a CompactionPipeline. agents/factory.py packages the common recipes so serving code (monolith and the distributed path) builds agents the same way.

• create_assistant_agent(...) — the everyday constructor. Give it a model_client, optional tools, system_instructions, and a memory provider; it builds the Toolbox, defaults the compaction pipeline to a SlidingWindowCompaction, and returns a ready (but unregistered) ReActAgent. You still call await runtime.register(agent) yourself — the factory stays decoupled from the runtime so it's testable without live infra.

• rebuild_messages_from_steps(...) — the lower-level converter that turns persisted step rows (user_message, assistant_message, tool_result, …) back into the unified ChatMessage list the agent loop expects. Used as the reseed callback for CachedHistoryProvider (capabilities/history/) — the cold-start mechanism: it wraps the shared, TTL'd Redis history and, on a cache miss, reconstructs the transcript from whichever cold store exists (the EventLog in the monolith, the conversation service over HTTP for the agent_runtime microservice) via this exact function, then repopulates the cache. This replaced an earlier side-channel helper (load_session_memory, since removed) — the cold-start behavior is now a property of the HistoryProvider itself, not a step every caller had to remember to invoke.

Where this lives

Next: The In-Process Runtime — the Runtime facade and Worker that actually call run(ctx, inbox), lease runs, and drive the durable loop these agents ride on.