What is an Agent

An Agent is a configured loop driver. It carries:

• Instructions. The system prompt.
• An architecture. The loop strategy (ReAct by default).
• A Dependencies bundle. Every protocol the architecture might need: model, memory, runtime, tool host, budget, permissions, hooks, telemetry, audit log.

from loomflow import Agent
 
agent = Agent(
    "You are a helpful assistant.",       # instructions
    model="claude-opus-4-7",              # the Model protocol
    memory="sqlite:./bot.db",             # the Memory protocol
    tools=[search, fetch],                # the ToolHost protocol
    budget=StandardBudget(BudgetConfig(...)),
    permissions=StandardPermissions(...),
    audit_log=FileAuditLog("./audit.jsonl"),
    telemetry=OTelTelemetry(...),
    runtime=SqliteRuntime("./journal.db"),
    architecture="react",                 # default
)

Each kwarg is a different protocol implementation. Most have no-op defaults so a bare Agent("...", model="...") is a complete object.

What it’s not

• Not a base class. You don’t subclass Agent. To customize the loop, pass a different architecture=. To customize a backend, pass a different protocol implementation.
• Not stateful at the class level. All per-run state lives in AgentSession (constructed fresh per agent.run()). The Agent itself is a configuration container.
• Not single-tenant. One Agent instance serves N users via user_id= on agent.run().

How agent.run() works

result = await agent.run(
    "What's the weather in Tokyo?",
    user_id="alice",                      # multi-tenant scope
    session_id="conv_42",                 # conversation continuity
    output_schema=MyPydanticModel,        # optional structured output
)

The Agent:

1. Opens a runtime session (sets a contextvar).
2. Opens a root telemetry span (loom.run).
3. Writes a run_started audit entry.
4. Seeds the message stack from memory (recent episodes, recent facts, working blocks).
5. Hands control to the architecture’s run() async generator.
6. Forwards events to the consumer (or awaits through them for agent.run()).
7. Persists the new episode to memory.
8. Triggers auto-extract for fact consolidation (if enabled).
9. Writes a run_completed audit entry.
10. Returns a RunResult.

For the full step-by-step including all four observability boundaries (events, telemetry, audit, runtime journal), see Architecture (internals).

What agent.stream() returns

stream() is the same loop, exposed as an async generator of Events:

async for event in agent.stream("..."):
    if event.kind == "model_chunk":
        print(event.payload["chunk"]["text"], end="")
    elif event.kind == "tool_call":
        log.info("calling tool %s", event.payload["call"]["tool"])

Same loop, same RunResult (built from the same AgentSession at the end). Pick run() when you only need the final answer; pick stream() when you want backpressure-aware token output.

Wall-clock timeout

Agent(timeout=) sets a ceiling, in seconds, on a whole run() or stream(). The clock covers everything: setup, every model call, every tool call, parallel tool dispatch, sub-agents, and teardown. Left unset (the default None), a run is unbounded — a hung tool or a stuck model call can hang the run forever.

from loomflow import Agent, RunTimeout
 
agent = Agent(
    "You are a research assistant.",
    model="claude-opus-4-7",
    timeout=120,                          # 2 minutes, whole run
)
 
try:
    result = await agent.run("Summarise the last quarter's incidents.")
except RunTimeout as exc:
    print(f"run exceeded {exc.seconds}s")

When the clock expires, the entire run is cancelled — pending model calls, in-flight tool dispatch, sub-agents — and RunTimeout is raised. It is not a soft interruption: you do not get a partial RunResult back, you get an exception. The value must be > 0; Agent(timeout=0) or any negative value raises ValueError at construction.

Goal loops (run_until=)

Agent(run_until=) turns a one-shot run into a “run until done” loop. After each pass through the architecture, a fast checker model is asked whether a measurable stop condition holds. If it doesn’t, the agent is re-prompted with the unmet condition and the checker’s reasoning, and runs again. The loop ends when the checker confirms the condition — or when one of three guardrails trips.

The shorthand form is a bare condition string:

agent = Agent(
    "You are a build agent. Keep working until the goal is met.",
    model="claude-opus-4-7",
    run_until="all tests pass and the endpoint returns 200",
)
 
result = await agent.run("Get the new endpoint green.")

The dict form lets you name a separate checker model and tighten the bounds:

agent = Agent(
    "You are a build agent. Keep working until the goal is met.",
    model="claude-opus-4-7",
    run_until={
        "condition": "all tests pass and the endpoint returns 200",
        "checker": "claude-haiku-4-5",   # cheap, fast goal checker
        "max_iterations": 20,
        "max_no_progress": 3,
        "max_cost_usd": 5.0,
    },
)

run_until dict keys

A bare string is exactly equivalent to {"condition": <str>} with every guardrail defaulted. Both forms normalise to the same internal state. An empty condition, or an unrecognised key in the dict, raises ConfigError at construction.

How the loop stops

The guardrails are first-class, not optional — an unbounded run-until loop is a classic way to burn budget flailing on an under-specified goal. After each architecture pass, the hook checks, in order:

1. max_iterations — the per-goal cap on re-prompts.
2. Cost — the agent-wide budget blocking the next step, or the loop’s cumulative cost reaching max_cost_usd.
3. max_no_progress — the observable state (output plus output tokens) unchanged for this many consecutive firings, i.e. the loop is spinning.

If none trip, the checker runs a single text-only DONE / NOT_DONE call against condition. DONE ends the loop cleanly; NOT_DONE re-prompts the agent, naming the unmet condition and the checker’s assessment, and the agent runs again.

Only a clean DONE is treated as success. Every guardrail exit is surfaced as an interruption on the RunResult:

result = await agent.run("Get the new endpoint green.")
 
if result.interrupted:
    # e.g. "run_until:max_iterations", "run_until:no_progress",
    #      "run_until:cost_cap"
    print("hit a guardrail:", result.interruption_reason)
else:
    print("goal met:", result.output)

The precise exit reason — condition_met, max_iterations, no_progress, cost_cap, or budget:<reason> — is always recorded in the session metadata under run_until.exit. When the goal is met, interrupted stays False; every other exit sets interrupted to True with interruption_reason prefixed run_until:, so a caller can distinguish “goal reached” from “hit a cap”.

For a runnable offline walkthrough (a scripted worker that takes two passes and a checker that says NOT_DONE then DONE, no API key), see Example 22.

Re-using one Agent across users

You construct ONE agent per logical service. Pass user_id= per call:

agent = Agent(
    "...",
    model="claude-opus-4-7",
    memory="postgres://...",
    audit_log=FileAuditLog("./audit.jsonl"),
)
 
# In a request handler:
async def handle(prompt, user_id, session_id):
    return await agent.run(prompt, user_id=user_id, session_id=session_id)

Memory partitions by user_id, audit attributes by user_id, budgets cap per user_id. The Agent itself is shared.