What is an Agent

An Agent is the thing you call to get work done by a model. You give it instructions and a model, then optionally the components it can use: tools, memory, limits, logging. Each call to agent.run() drives a loop: the model reads the task, may call tools, and decides when it’s finished.

from datetime import timedelta
from loomflow import Agent, StandardPermissions, tool
from loomflow.governance.budget import BudgetConfig, StandardBudget
from loomflow.runtime import SqliteRuntime
from loomflow.security import FileAuditLog
 
@tool
async def search(query: str) -> str:
    """Search the knowledge base."""
    return "..."
 
agent = Agent(
    "You are a helpful assistant.",          # instructions
    model="claude-opus-4-7",                 # required
    memory="sqlite:./bot.db",                # what it remembers
    tools=[search],                          # what it can do
    budget=StandardBudget(BudgetConfig(max_cost_usd=2.0, max_wall_clock=timedelta(minutes=5))),
    permissions=StandardPermissions(),       # which tools may run
    audit_log=FileAuditLog("./audit.jsonl"),
    runtime=SqliteRuntime("./journal.db"),   # crash recovery
    architecture="react",                    # the loop style (default)
)

Only the instructions and model= are required. Everything else has a default that does nothing, so Agent("...", model="...") is complete on its own, and you add components as you need them. Each component is an interface, so you can pass the built-in implementation or your own.

What it’s not

• Not a base class. You don’t subclass Agent. To change the loop, pass a different architecture=; to change a component, pass a different implementation.
• Not tied to one run. Everything about a single run lives in a fresh session object created for that call. The Agent itself is configuration, so concurrent runs on one agent are safe.
• Not tied to one user. One Agent serves many users: pass user_id= on each 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. Starts auto-extract for fact consolidation (if enabled). It runs in the background after run() returns; call await agent.aclose() to wait for it before reading the facts a run produced, and at shutdown.
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. Breaking out of the async for cancels the run, and your next await runs normally (see stream for trio).

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 always on, because an unbounded run-until loop is an easy way to burn budget on an under-specified goal. After each architecture pass, the hook checks, in order:

1. Cost: the agent-wide budget blocking the next step, or the loop’s cumulative cost reaching max_cost_usd.
2. max_no_progress: the observable state (output plus output tokens) unchanged for this many consecutive firings, i.e. the loop is spinning.

If neither trips, 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, unless that was pass number max_iterations, which ends the loop as run_until:max_iterations. The checker is asked after every pass, the last one included. (Before 0.17 the last pass wasn’t checked, so a goal met on it was reported as run_until:max_iterations.)

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.