Architecture protocols + base types

from loomflow import Architecture
from loomflow.architecture import AgentSession, Dependencies

An architecture is the strategy an Agent uses to work through a task (ReAct, plan-and-execute, a supervisor with workers, and so on). You only need this page if you are writing your own architecture or reading the built-in ones. It covers the three pieces involved: the Architecture protocol every strategy implements, the per-run state it updates (AgentSession), and the bundle of backends the Agent hands it on each run (Dependencies).

For the conceptual overview see Architectures. For implementing a custom architecture see Custom architectures.

Architecture

from typing import Protocol, runtime_checkable
from collections.abc import AsyncIterator
 
@runtime_checkable
class Architecture(Protocol):
    name: str
 
    def declared_workers(self) -> dict[str, "Agent"]: ...
 
    async def run(
        self,
        session: AgentSession,
        deps: Dependencies,
        prompt: str,
    ) -> AsyncIterator[Event]:
        ...

The strategy that drives the agent loop. The Agent calls architecture.run(session, deps, prompt) and consumes the events it yields. The Agent itself emits the setup event (Event.started) and the teardown event (Event.completed); architectures yield only the events that happen during iteration.

Attributes

Methods

declared_workers

def declared_workers(self) -> dict[str, Agent]: ...

Names of any sub-agents the architecture delegates to, mapped to their Agent instances. Used by AgentGraph for visualization. Empty dict for single-agent loops.

run

async def run(
    self,
    session: AgentSession,
    deps: Dependencies,
    prompt: str,
) -> AsyncIterator[Event]: ...

Async generator yielding Events as iteration progresses. Mutates session in place (messages, turns, output, cumulative_usage, interrupted, interruption_reason, metadata).

The Agent may call run again on the same session when a stop hook (for example run_until=) asks it to keep going. On re-entry, session.messages is not empty and prompt is a new user turn: append it as a Role.USER message rather than starting over.

AgentSession

@dataclass
class AgentSession:
    id: str
    instructions: str
    messages: list[Message] = field(default_factory=list)
    turns: int = 0
    output: str = ""
    cumulative_usage: Usage = field(default_factory=Usage)
    interrupted: bool = False
    interruption_reason: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

Mutable per-run state shared between Agent and the architecture. The Agent constructs this once per run, the architecture mutates it as iteration progresses, and the Agent reads the final state to build a RunResult.

Fields

Dependencies

@dataclass
class Dependencies:
    model: Model
    memory: Memory
    runtime: Runtime
    tools: ToolHost
    budget: Budget
    permissions: Permissions
    hooks: HookRegistry
    telemetry: Telemetry
    audit_log: AuditLog | None
    max_turns: int
    approval_handler: ApprovalHandler | None = None
    output_schema: Any | None = None
    effort: str | None = None
    strict_effort: bool = False
    prompt_caching: Any = None
    streaming: bool = False
    context: RunContext = field(default_factory=RunContext)
    tool_timeout_s: float | None = 120.0
    tool_result_max_chars: int = 50_000
    compact_context: Callable[[AgentSession], Awaitable[Event | None]] | None = None
    # ...plus the settings for optional features (rate limiter,
    # guardrails, tool-result summariser, run_until checker, snip,
    # tool search, memory budget) and the fast_* flags below.

Bundled protocol implementations passed to every architecture. Constructed once per run from the Agent’s configured backends. Architectures treat this as read-only. They call methods on the contained protocols but don’t mutate the struct.

Core fields

Optional fields

Fast-mode flags

Set automatically by the Agent based on which features are wired. When a layer is a no-op default, the hot path skips its call sites entirely:

As soon as a real implementation is wired, the matching flag turns False and the integration point becomes active. You never set these flags yourself.

ApprovalHandler (type alias)

ApprovalHandler = Callable[[ToolCall, str | None], Awaitable[ApprovalDecision | bool]]

Async callable that settles a PermissionDecision.ask_(...) from permissions.

Returns True to allow the call or False to deny it, an action name ("allow", "deny", "edit", "remember_allow", "remember_deny"), or an ApprovalDecision for richer outcomes (edit the arguments, remember the answer for the rest of the run, or deny with a reason). Anything else, such as "yes" or {"approved": True}, denies, and so does a handler that raises. (Before 0.17 the result was read with bool(), so "deny" approved the call.) See Approval handlers.

Source

loomflow/architecture/base.py