Agent

from loomflow import Agent

An Agent is a model plus everything it runs with: instructions, tools, memory, limits and policies. You build one, then call await agent.run(prompt). It loops (call the model, run any tools the model asks for, repeat) until the model gives a final answer. Use an Agent when the model should decide the steps; use a Workflow when you want to fix the steps in code. One instance can serve many users: pass user_id= on each agent.run().

This page lists every constructor parameter and method. For the conceptual overview see What is an Agent.

Class signature

class Agent:
    def __init__(
        self,
        instructions: str,
        *,
        model: Model | str | dict[str, Any] | None = None,
        memory: Memory | str | Mapping[str, Any] | None = None,
        runtime: Runtime | str | Mapping[str, Any] | None = None,
        budget: Budget | None = None,
        rate_limiter: RateLimiter | None = None,
        guardrails: Sequence[Guardrail] | None = None,
        permissions: Permissions | None = None,
        hooks: HookRegistry | None = None,
        tools: list[Tool | Callable[..., object]]
             | ToolHost
             | Tool
             | Callable[..., object]
             | None = None,
        telemetry: Telemetry | str | Mapping[str, Any] | None = None,
        audit_log: AuditLog | str | Path | dict[str, Any] | None = None,
        max_turns: int = 50,
        timeout: float | None = None,
        architecture: Architecture | str | None = None,
        skills: list[Any] | None = None,
        auto_extract: bool | None = None,
        approval_handler: ApprovalHandler | None = None,
        output_schema: type[BaseModel] | Any | None = None,
        effort: str | None = None,
        strict_effort: bool = False,
        prompt_caching: bool | Mapping[str, Any] | None = None,
        workspace: Workspace | WorkspaceMembership | str | Mapping[str, Any] | None = None,
        living_plan: bool | LivingPlan | None = None,
        run_until: str | Mapping[str, Any] | None = None,
        tool_result_summarizer: Model | str | None = None,
        snip_window: int = 0,
        auto_compact_at_tokens: int | Literal["auto"] | None = None,
        persist_tool_transcripts: bool = False,
        tuning: Tuning | None = None,
    ) -> None: ...

Everything after instructions is keyword-only. Knobs that tune existing behaviour rather than add a capability (retries, secrets, tone, stop hooks, compaction details and a few more) live on a separate Tuning dataclass, passed as tuning=Tuning(...). See Tuning fields below.

Constructor parameters

instructions

The system prompt the model sees. When skills= is non-empty, the framework appends a per-skill catalog (~50 tokens / skill) to this string so the model can decide which skill to load on demand.

agent = Agent(
    "You are a careful research assistant. Cite your sources.",
    model="claude-opus-4-7",
)

model

Three accepted shapes: a Model protocol implementation, a string the resolver maps to one, or a dict config. Required. There is no default model, so leaving it out raises ConfigError when the Agent is constructed.

String prefixes:

from loomflow.model.anthropic import AnthropicModel
 
agent = Agent("...", model="claude-opus-4-7")
agent = Agent("...", model=AnthropicModel("claude-opus-4-7", max_tokens=8192))

Gateway ids such as OpenRouter’s anthropic/claude-haiku-5.5 aren’t resolved from a string: build the adapter with the gateway’s URL and pass it (see Gateways). Capabilities and price are then looked up as the vendor’s model.

Dict-config form. Use it when you want the model spec and its related dials in one place:

agent = Agent(
    "...",
    model={
        "name": "claude-opus-4-7",
        "effort": "high",         # see Reasoning effort
        "strict_effort": True,    # fail loudly if model can't honour effort
    },
)

Same shape philosophy as audit_log={...}. Top-level kwargs win over matching keys in the dict, so you can layer environment overrides on top of a shared config. See Reasoning effort for the dict’s effort / strict_effort keys.

memory

Conversational state: episodes, working blocks and bi-temporal facts. The resolver accepts:

• None → InMemoryMemory() (lost on process exit).
• "inmemory" → InMemoryMemory().
• "sqlite:./bot.db" → SqliteMemory.connect(...) (lazy).
• "chroma" / "chroma:./path" → ChromaMemory.local(...).
• "postgres://user:pw@host/db" → PostgresMemory.connect(...) (lazy).
• "redis://localhost:6379/0" → RedisMemory.connect(...) (lazy).
• A Mapping (dict): {"backend": ..., "path": ..., "namespace": ..., "embedder": ..., "with_facts": ...}.
• An explicit Memory instance.

For async backends (Postgres / Redis) the resolver returns a LazyMemory proxy; the connection opens on first use so Agent(...) stays synchronous. See Memory backends.

runtime

The journaling layer. With a durable runtime (SqliteRuntime("./journal.db") or PostgresRuntime), every model call and tool dispatch is journaled under the session and user (journal_scope(session_id, user_id)) and the session’s current run. Running a crashed run again, or calling agent.resume(session_id=...), replays the steps that already finished instead of re-executing them. A run that finished is never replayed into a later run on the same session (see Runs and epochs). Journal values are stored as JSON.

Strings: "inproc", "sqlite" (in-memory SQLite journal) or "sqlite:./journal.db". Postgres needs an async connect, so build it with await PostgresRuntime.connect(dsn) and pass the instance.

InProcRuntime (the default) runs each step directly with no journaling overhead. See Runtime.

budget

Hard caps on tokens / cost / wall-clock, agent-wide and per user_id. The wall-clock caps count time spent running (each run is wrapped in StandardBudget.running()), not time since the budget was created. StandardBudget(BudgetConfig(...)) is the production choice; see Per-user budget caps.

When the budget is exceeded, the run stops cleanly with RunResult.interrupted = True and RunResult.interruption_reason set to "budget:<cap>", for example "budget:max_tokens" or "budget:per_user_max_cost_usd".

rate_limiter

Caps how fast each tenant can make model calls. A budget limits the total spend; a rate limiter limits the burst rate. The limiter is checked once before every model step.

from loomflow import Agent, TokenBucketRateLimiter
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    rate_limiter=TokenBucketRateLimiter(rps=2, burst=5),
)

TokenBucketRateLimiter keeps one bucket per user_id by default. In mode="throttle" (the default) a step waits for a token; in mode="raise" it raises RateLimitExceeded instead. See Rate limiting.

guardrails

Ordered text guards that inspect the user prompt (input), each tool result (tool_result) and the final answer (output). A guard can allow the text, rewrite it (for example redact PII or wrap untrusted content) or block it. A blocked prompt or answer ends the run with interrupted=True and interruption_reason="guardrail:<name>". Guards run in list order, each seeing the previous guard’s output. With an output-stage guard wired, streamed text chunks also pass through the output guards before they’re sent, with a 128-character hold-back (see Streamed text).

from loomflow import Agent
from loomflow.guardrails import InjectionGuard, PIIGuard
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    guardrails=[InjectionGuard(), PIIGuard()],
)

Guardrails are imported from loomflow.guardrails, not the top-level package. See Guardrails.

permissions

Decides allow / deny / ask per tool call. Production usually wants StandardPermissions(mode=Mode.DEFAULT) paired with an approval_handler. For multi-tenant policy routing, use PerUserPermissions(policies=, default=). Mode.ACCEPT_EDITS auto-approves only the built-in write and edit tools; every other destructive tool still asks. (Before 0.17 it approved every destructive tool, bash included.) Name your own editors with StandardPermissions(edit_tools=[...]). See Permissions.

hooks

Registry that holds before_tool / after_tool callbacks. You rarely pass this directly; use the @agent.before_tool / @agent.after_tool decorators after construction. Pass your own HookRegistry(hook_timeout_s=..., timeout_denies=...) to change how long a hook may take (default 5 seconds) and whether a timed-out pre-tool hook denies the call (it does by default). See Hooks.

tools

The tool surface. Five accepted shapes:

# A list (the common case)
agent = Agent("...", model="claude-opus-4-7", tools=[get_weather, send_email])
 
# A single tool
agent = Agent("...", model="claude-opus-4-7", tools=get_weather)
 
# A pre-built host (lets you swap the dispatch implementation)
agent = Agent("...", model="claude-opus-4-7", tools=InProcessToolHost([get_weather]))
 
# An MCP registry
agent = Agent("...", model="claude-opus-4-7", tools=MCPRegistry([git_server, fs_server]))
 
# A sandbox-wrapped host
agent = Agent("...", model="claude-opus-4-7", tools=FilesystemSandbox(host, roots=["/work"]))

Bare callables are auto-wrapped via @tool. An MCP tool name that more than one server offers reaches the model as server__tool (before 0.17, server.tool, which providers reject); the dotted name still resolves. See Tools.

telemetry

Sink for spans and metrics. Choose InMemoryTelemetry / ConsoleTelemetry / FileTelemetry / MultiTelemetry for collector-free development, or OTelTelemetry(tracer_provider=...) for production OTLP. Strings work too: "console", "otel" or "file:./spans.jsonl". The loop emits loom.run / loom.turn / loom.model.complete (loom.model.stream when streaming) / loom.tool spans plus loom.tokens.* / loom.cost.usd / loom.tool.duration_ms and the GenAI gen_ai.client.* / gen_ai.execute_tool.duration metrics. OTelTelemetry exports the model, tool and run spans as chat {model}, execute_tool {tool} and invoke_agent, keeping the loom.* name in the loom.span.name attribute. Session and user ids are span attributes, not metric tags. See Telemetry.

audit_log

InMemoryAuditLog() for tests, FileAuditLog(path, secret="...") for production. Every run_started / tool_call / tool_result / run_completed entry is HMAC-signed and attributed to the active user_id. See Audit log attribution.

Dict-config form. Hand the resolver a dict and it builds the right backend for you:

agent = Agent(
    "...",
    model="claude-opus-4-7",
    audit_log={
        "name": "./audit.jsonl",     # path; omit this for in-memory
        "scope_full": True,          # capture prompts + outputs verbatim
        "secret": "my-org-key",      # optional; HMAC-signs every entry
    },
)

The default audit log is compliance-friendly. Prompts get truncated at 500 characters, the model’s output isn’t recorded, and tool results carry only ok / denied / error / reason. Set scope_full: True to opt into verbatim capture for debugging; the resolver then wraps the backend in FullTranscriptAuditLog, and signatures still verify. The same dict shape works on Workflow.

max_turns

Hard cap on iterations of the agent loop. Hitting it stops the run cleanly with RunResult.interrupted = True and RunResult.interruption_reason = "max_turns_exceeded". Lower it for expensive tools or runaway prevention; raise it for deep multi-step tasks.

timeout

Wall-clock ceiling, in seconds, for a whole run() or stream(): setup, every model and tool call, sub-agents and teardown. When it expires the run is cancelled and RunTimeout is raised (you get an exception, not a partial RunResult). Zero or a negative value raises ValueError at construction.

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

A single slow model call is a different failure: it surfaces as a TransientModelError that the retry layer handles. See Wall-clock timeout.

architecture

The loop strategy. Pass an instance (Supervisor(workers={...}), PlanAndExecute(...), Reflexion(base=...)) or a string spec ("react", "plan-and-execute", "rewoo", "reflexion", "self-refine", "tree-of-thoughts"). Multi-agent shapes (Supervisor / Router / Swarm / Debate / ActorCritic / Blackboard) need an instance, because the string resolver only knows the single-agent loops. See Architectures.

skills

Anthropic-format SKILL.md packages. Accepts paths to skill directories, in-line Skill instances, or (path, label) tuples for source labelling. When several sources define the same skill, the last one wins; skill tools are namespaced with a skill_name__ prefix. Names and descriptions are checked against the Agent Skills spec when the agent is built. A skill’s bundled script tools are marked destructive (so StandardPermissions asks before running them) and run under the sandbox you passed as tools=. See Skills.

auto_extract

When None, the framework picks True for real network adapters (OpenAIModel / AnthropicModel / LiteLLMModel) and False for in-process fakes (EchoModel / ScriptedModel). When True, every agent.run() schedules a small Consolidator pass that pulls structured (subject, predicate, object) facts into the bi-temporal fact store. It runs in the background after run() returns; await agent.aclose() waits for it (see aclose). Telemetry signals: loom.auto_extract.duration_ms, loom.auto_extract.invocations. See Bi-temporal facts.

approval_handler

Async callable that settles PermissionDecision.ask_(...) outcomes from the permissions layer. It receives the pending ToolCall and the live user_id, and returns True to allow or False to deny. Return an ApprovalDecision for richer outcomes: "edit" (run the tool with edited_args), "remember_allow" / "remember_deny" (cache the answer for this user and tool for the rest of the run), or "allow" / "deny" with a reason. A bare action name ("allow", "deny", …) counts as the matching ApprovalDecision.

Every other outcome is a deny, so the agent never silently bypasses the gate:

• Returns False → the tool result is denied, with the policy’s reason for asking (or "approval declined" when it gave none).
• Not wired (None) → denied with reason="approval required; no approver".
• Raises → treated as deny.
• Returns anything else (None, "no", {"approved": False}, …) → treated as deny, with a warning logged.

See Approval handlers.

output_schema

Default schema for the final answer. When set, every agent.run() / agent.stream() call validates the answer against it; a per-call output_schema= on run() still takes precedence for one-off shapes. Similar to Pydantic AI’s output_type=.

from pydantic import BaseModel
from loomflow import Agent
 
class Invoice(BaseModel):
    amount_cents: int
    vendor: str
 
# Once on construction:
extractor = Agent(
    "Extract invoice fields.",
    model="gpt-4o",
    output_schema=Invoice,
)
 
# Every run uses Invoice automatically:
result = await extractor.run("From: alice@acme...")
invoice: Invoice = result.parsed

Tagged-union schemas. Pass output_schema=A | B (or Union[A, B]) and the output validates as the member it fills best: the one with the most fields set (Pydantic’s smart-mode rule), not simply the first member that accepts it, so a payload that fills a richer member isn’t read as a smaller member it also happens to satisfy. The union is described in the prompt even on adapters with native structured output (which can’t express it). This lets you model “valid result vs structured error” without a discriminator field:

class Found(BaseModel):
    invoice_no: str
    total_cents: int
 
class NotFound(BaseModel):
    reason: str
 
agent = Agent("Look up the invoice.", model="gpt-4o",
              output_schema=Found | NotFound)

Native structured-output support. When an output_schema is set, model adapters translate it into the provider’s native idiom so the model is constrained at decode time:

• OpenAI: response_format={"type": "json_schema", ..., "strict": True} on chat.completions.create. A schema with a dict field goes out with "strict": False, since strict mode would only allow {} there.
• Anthropic: on Claude 4.5 and later, native JSON outputs (output_config.format), so the agent’s real tools stay callable. Older models, and schemas the API’s JSON-schema subset can’t express, get a synthetic __output__ tool instead: forced when it is the only tool, required among the tools (tool_choice: any) when there are others, and left to the model when thinking is on.
• LiteLLM: none. Many providers behind LiteLLM reject response_format, so the schema goes in the prompt.

Adapters without native support fall back to adding the schema to the prompt and retrying on validation failure. Either way the API is the same: pass output_schema=, get a validated result.parsed.

Cost. When an adapter declares supports_native_structured_output = True, the framework skips appending the JSON-schema directive to the system prompt, since the provider already constrains the output. This saves input tokens. If the model still produces invalid JSON, the retry path injects the schema, so the safety net stays. The OpenAI and Anthropic adapters set the flag; LiteLLMModel and custom adapters default to False, which keeps the prompt directive on. A union schema always gets the prompt directive.

effort

Unified reasoning-effort dial across providers. Pass one of "minimal", "low", "medium", "high", "xhigh", "max"; any other string raises ValueError (when the agent is built, or when the run starts for a per-call value). The framework translates it to the model’s native shape: OpenAI’s reasoning_effort, Anthropic’s adaptive thinking plus output_config.effort (Claude 4.6 and later), a thinking.budget_tokens budget on older Claude thinking models such as Haiku 4.5 and Sonnet 4.5, or LiteLLM’s normalised passthrough. A level the model doesn’t accept is clamped to the nearest one it does (minimal becomes low on Claude). Gateway ids such as anthropic/claude-haiku-5.5 get the vendor model’s levels.

agent = Agent("...", model="claude-opus-4-7", effort="high")

Resolution order (highest priority first):

1. Per-call agent.run(..., effort=...) override.
2. Agent(effort=...) default.
3. None: the provider’s own default.

Models that can’t honour the dial (non-reasoning OpenAI models such as gpt-4o, and Claude models older than Sonnet 3.7) drop the kwarg and emit a one-time UserWarning per (model, effort) pair. Opt into a hard failure with strict_effort=True.

See Reasoning effort for the levels each model accepts and how they map.

strict_effort

When True, wiring effort= to a model that can’t honour it raises EffortNotSupportedError instead of dropping the kwarg with a warning:

from loomflow import Agent
from loomflow.model._effort import EffortNotSupportedError
 
agent = Agent(
    "...",
    model="gpt-4o",               # not a reasoning model
    effort="high",
    strict_effort=True,
)
 
try:
    await agent.run("hi")
except EffortNotSupportedError as exc:
    ...   # the wiring was wrong

Use it in CI / pre-prod to surface model-mismatch typos immediately. Leave it at False in production so a fallback onto a non-reasoning model during a vendor outage doesn’t crash the agent.

strict_effort is set on the agent only; there’s no per-call override, because whether a model can honour effort is a property of the adapter, not of any single call.

prompt_caching

Per-provider prompt caching. True enables it with a 5-minute TTL. A dict gives per-field control:

agent = Agent(
    LARGE_SYSTEM_PROMPT,
    model="claude-opus-4-7",
    prompt_caching={"enabled": True, "ttl": "1h", "cache_key": "session_42"},
)

On Anthropic the framework injects cache_control markers on the system prompt, the last tool definition and the end of the conversation (the API’s limit is four). On OpenAI it parses cached_tokens for accurate accounting and forwards cache_key as the routing hint. Read tokens land in RunResult.cached_tokens_in; cost_usd already reflects the discount. See Prompt caching.

workspace

Wires a shared notebook and installs the five notebook tools (note, read_note, list_notes, search_notes, update_note) on the agent’s tool host. Accepts:

• A WorkspaceMembership via ws.member("name", teammates=[...]): the usual path, carrying the notebook plus this agent’s identity.
• A bare Workspace instance: shared notebook, generic agent attribution.
• A string: "temp", "memory", or a filesystem path.
• A dict: {"backend": ..., "author": ..., "teammates": [...]}.

from loomflow import Agent, LocalDiskWorkspace
 
ws = LocalDiskWorkspace.temp()
agent = Agent(
    "...",
    model="claude-opus-4-7",
    workspace=ws.member("researcher", teammates=["analyst", "writer"]),
)

Propagates through Workflow and Team the same way memory= does: an agent without its own workspace= that runs inside one gets the notebook tools and the notebook section of the system prompt for that run. See Workspace.

living_plan

Enables the TodoWrite-style living plan. True installs plan_write and plan_read and adds the plan discipline to the system prompt. Pass a constructed LivingPlan to start the run with a plan already in place.

agent = Agent("...", model="claude-opus-4-7", living_plan=True)

When a workspace= is also wired, a third tool appears, recall_past_plans, and every plan_write mirrors to a kind="plan" note so future runs can bootstrap from prior plans.

run_until

Keeps the agent working until a measurable condition holds. After each pass, the last allowed one included, a checker model answers DONE or NOT_DONE against the condition; on NOT_DONE the agent is re-prompted with what is still missing. Pass a bare condition string, or a dict to set the checker and the limits:

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,
    },
)

Dict keys: condition (required), checker (defaults to the main model), decider (a decision model that answers clear cases before the checker is asked), max_iterations (default 20), max_no_progress (default 3), max_cost_usd and checker_prompt. An empty condition or an unknown key raises ConfigError. When a limit stops the loop, the result has interrupted=True and an interruption_reason starting with run_until:. See Goal loops.

tool_result_summarizer

A model (usually a small, cheap one) that shortens long tool results before they enter the conversation. Results longer than Tuning.tool_result_summary_threshold characters (default 500) are replaced by a summary, so later turns don’t keep paying for an old file read or log dump. Accepts the same shapes as model=. If the summariser fails, the original result is kept.

agent = Agent(
    "...",
    model="claude-opus-4-7",
    tool_result_summarizer="claude-haiku-4-5",
)

snip_window

Keeps only the last N turns of the conversation before each pass. A turn here is a user message plus everything that followed it (assistant replies, tool calls and their results), so tool calls are never separated from their results. Snipping is plain list slicing with no model call: cheap, but older context is dropped rather than summarised. Negative values raise ValueError.

agent = Agent("...", model="claude-opus-4-7", snip_window=20)

auto_compact_at_tokens

Summarises older turns once the conversation grows past this many tokens. ReAct checks before every model call, so a run whose own tool results pile up is compacted while it runs, and a long rehydrated history is compacted before the run’s first call. (It also runs after a pass when a checkpoint is about to be written or a stop-hook pass follows.) Past the threshold, everything except the system prompt and the most recent turns is replaced by one summary message; within one long tool loop the older exchanges are summarised, with tool calls kept alongside their results. "auto" sets the threshold to 80% of the model’s context window for known models (and leaves compaction off for unknown ones). Summaries of an unchanged history are cached, and short conversations skip the token count entirely.

agent = Agent("...", model="claude-opus-4-7", auto_compact_at_tokens="auto")

The summariser defaults to the main model; set Tuning(auto_compact_summariser=...) for a cheaper one and Tuning(auto_compact_keep_recent_turns=...) (default 4) for how many recent turns stay verbatim. A failed summary leaves the conversation unchanged.

persist_tool_transcripts

By default memory stores only the prompt and the final answer of each run. When True, the run’s own tool calls and results (not the earlier runs’ it rehydrated) are stored with the episode too, and replayed when a later run in the same session rehydrates history, so the agent sees what it read or ran last time. Each stored entry is capped at Tuning.tool_transcript_max_bytes (default 50_000).

agent = Agent("...", model="claude-opus-4-7", persist_tool_transcripts=True)

name

The agent’s name. It labels the agent’s run span (gen_ai.agent.name), names its node in Workflow.chain and its branch in Workflow.parallel (unnamed agents get generated names, such as step_0), and keys its history and versions in Studio. An empty name raises ValueError. Also accepted as name in from_config.

from loomflow import Agent
 
billing = Agent("You are the billing agent.", model="claude-haiku-5-5", name="billing-agent")
assert billing.name == "billing-agent"

tuning

Groups the rarely-touched knobs. See the next section.

Tuning fields

from loomflow import Agent, Tuning
 
agent = Agent("...", model="claude-opus-4-7", tuning=Tuning(response_tone="executive"))

Tuning() with no arguments is a complete, valid config; override only what you need. The four most-used fields are documented in full below. The rest:

Passing any of these directly to Agent(...) still works but emits a DeprecationWarning; use tuning=Tuning(...).

retry_policy

When None, the framework uses the default RetryPolicy() for network-backed models (AnthropicModel / OpenAIModel / LiteLLMModel) and RetryPolicy.disabled() for in-process fakes (EchoModel / ScriptedModel). Pass RetryPolicy.aggressive(), RetryPolicy.disabled(), or a custom RetryPolicy(max_attempts=..., initial_delay_s=..., multiplier=..., max_delay_s=..., jitter=...) to override. jitter defaults to "full" (each delay drawn from 0 to the computed backoff); a float such as 0.1 applies ±10% instead. The SDK clients loomflow builds have their own retries turned off, so attempts aren’t multiplied. A client you build and pass in keeps its own (the SDKs default to 2); give it max_retries=0 to leave retrying to the policy. See RetryPolicy.

Pass via tuning=Tuning(retry_policy=...).

secrets

API key resolver. Resolution order inside model adapters:

1. Explicit api_key= argument on the adapter.
2. secrets.lookup_sync(<ENV_VAR_NAME>).
3. os.environ[<ENV_VAR_NAME>].

For Vault / AWS Secrets Manager / 1Password, write a custom adapter satisfying the Secrets protocol. See Secrets resolution.

Pass via tuning=Tuning(secrets=...).

auto_consolidate

When True, calls await self.consolidate() after every agent.run() to extract facts from the new episode. For most production cases prefer auto_extract=True (the default for real models): it does the same extraction on the write path, with telemetry. Use auto_consolidate when you want an explicit post-run consolidation pass.

Pass via tuning=Tuning(auto_consolidate=True).

response_tone

Steers how the agent phrases its output, not what it answers. The framework appends a one-line style directive to the system prompt, after any schema directive (instructions late in the system prompt tend to get the most weight).

Three separate levers; don’t conflate them:

Shipped presets. Each is one short sentence on purpose, since longer prompt fragments dilute the effect:

Free-form strings. Anything that isn’t a preset name is sent verbatim:

from loomflow import Tuning
 
agent = Agent("...", model="gpt-4.1-mini", tuning=Tuning(response_tone="legal"))
agent = Agent(
    "...",
    model="gpt-4.1-mini",
    tuning=Tuning(response_tone="warm but precise, like a doctor explaining a diagnosis"),
)

Resolution order (highest priority first):

1. Per-call agent.run(..., response_tone=...) override.
2. Agent(tuning=Tuning(response_tone=...)) default.
3. The workflow’s tone: Workflow(response_tone=...) applies to every nested Agent step that didn’t set its own, the same way Workflow(memory=...) does.
4. None: no tone directive.

The workflow tone is reset when the workflow run ends, so it never leaks into the next run.

Methods

from_config

@classmethod
def from_config(
    cls,
    path: str | Path,
    *,
    model: Model | str | dict[str, Any] | None = None,
    memory: Memory | str | Mapping[str, Any] | None = None,
    runtime: Runtime | str | Mapping[str, Any] | None = None,
    telemetry: Telemetry | str | Mapping[str, Any] | None = None,
    audit_log: AuditLog | str | Path | dict[str, Any] | None = None,
    permissions: Permissions | str | Mapping[str, Any] | None = None,
    tools: list[Tool | Callable[..., object]] | ToolHost | None = None,
    secrets: Any | None = None,
    hooks: HookRegistry | None = None,
    retry_policy: RetryPolicy | None = None,
    approval_handler: Any | None = None,
) -> Agent: ...

Reads a TOML file and builds an Agent from it, for teams who want declarative config kept separate from code (ops, SRE, compliance). Everything the framework can build synchronously (model, memory, runtime, telemetry, audit log, permissions, budget, architecture, effort, skills, MCP servers, timeout, run_until, living_plan, prompt_caching, snip_window) goes in the TOML. Things TOML can’t express naturally (real callables, custom hook objects, secret stores, retry policies) come in as kwargs, which override matching entries in the file.

from loomflow import Agent
 
agent = Agent.from_config("./agent.toml")

An unknown key, at the top level, in [budget] or in a prompt_caching table, raises ConfigError naming the closest known key (max_turn → max_turns). (Before 0.17 unknown keys were ignored.)

Requires Python 3.11+ for tomllib. See Config file for the full TOML schema reference.

from_dict

@classmethod
def from_dict(
    cls,
    cfg: dict[str, Any],
    *,
    model: Model | str | dict[str, Any] | None = None,
    memory: Memory | str | Mapping[str, Any] | None = None,
    runtime: Runtime | str | Mapping[str, Any] | None = None,
    telemetry: Telemetry | str | Mapping[str, Any] | None = None,
    audit_log: AuditLog | str | Path | dict[str, Any] | None = None,
    permissions: Permissions | str | Mapping[str, Any] | None = None,
    tools: list[Tool | Callable[..., object]] | ToolHost | None = None,
    secrets: Any | None = None,
    hooks: HookRegistry | None = None,
    retry_policy: RetryPolicy | None = None,
    approval_handler: Any | None = None,
) -> Agent: ...

Same as from_config but takes a dict instead of a file. Useful when the config already lives somewhere structured: a Pydantic BaseSettings, a YAML file you’ve already parsed, a service-config response, or anything else that hands you a dict.

agent = Agent.from_dict({
    "instructions": "You are a helpful assistant.",
    "model": "echo",
    "memory": {"backend": "sqlite", "path": "./m.db"},
    "budget": {"max_tokens": 10_000},
})

run

async def run(
    self,
    prompt: str,
    *,
    user_id: str | None = None,
    session_id: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    context: RunContext | None = None,
    extra_tools: list[Tool] | None = None,
    emit: Callable[[Event], Awaitable[None]] | None = None,
    output_schema: type[BaseModel] | Any | None = None,
    output_validation_retries: int = 1,
    response_tone: str | None = None,
    effort: str | None = None,
) -> RunResult: ...

Run the agent to completion and return its RunResult.

Returns a RunResult:

Example: typed structured output with retry

from datetime import date
from pydantic import BaseModel
from loomflow import Agent
 
class Invoice(BaseModel):
    amount_cents: int
    vendor: str
    issued_on: date
 
agent = Agent("Extract invoice fields.", model="claude-opus-4-7")
 
result = await agent.run(
    "Extract the invoice fields from this email body: ...",
    user_id="alice",
    session_id="invoice_extract_2026_05_09",
    output_schema=Invoice,
    output_validation_retries=2,
)
invoice: Invoice = result.parsed   # validated typed instance
# or use the smart accessor: result.value behaves the same here

Example: the result.value smart accessor

# Without a schema:
r = await agent.run("Hello")
r.value           # → str (same as r.output)
 
# With a schema:
r = await agent.run("Extract...", output_schema=Invoice)
r.value           # → Invoice (same as r.parsed)
 
# Code that handles both paths uniformly:
def show(answer):
    print(f"Got: {answer}")
show(r.value)     # works regardless of whether a schema fired

resume

async def resume(
    self,
    prompt: str | None = None,
    *,
    session_id: str,
    from_checkpoint: str = "latest",
    user_id: str | None = None,
    **kwargs: Any,
) -> RunResult: ...

Continue an interrupted run (crash, deploy, Ctrl-C) instead of starting over.

When the agent was built with tuning=Tuning(checkpoint=True) and a checkpoint exists, the conversation, turn count and usage are restored from it and earlier turns are not re-billed. Without a checkpoint, resume behaves like run(prompt, session_id=...) continuing the session’s last run: memory rebuilds the conversation and, with a durable runtime, that run’s journaled steps replay instead of re-executing, even if the run finished. (A plain run() on the session replays only a run that crashed.)

from loomflow import Agent, Tuning
from loomflow.runtime import SqliteRuntime
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    runtime=SqliteRuntime("./journal.db"),
    tuning=Tuning(checkpoint=True),
)
 
# First attempt, interrupted by Ctrl-C, OOM or a deploy.
result = await agent.run("complex task", session_id="task-2026-05-09")
 
# Later, after the process restarted:
result = await agent.resume(session_id="task-2026-05-09")

InProcRuntime checkpoints live in memory and are lost when the process exits; use SqliteRuntime or PostgresRuntime to survive a crash. See Checkpoints and Replay and resume.

list_checkpoints

async def list_checkpoints(
    self,
    session_id: str,
    limit: int = 50,
    *,
    user_id: str | None = None,
) -> list[CheckpointMeta]: ...

Checkpoints saved for a session, newest first. Checkpoints are stored per (session, user), so pass the user_id the run used. Each CheckpointMeta has checkpoint_id, turn and created_at; pass a checkpoint_id to resume(from_checkpoint=...) to fork from it. Returns [] when the runtime doesn’t support checkpoints.

for cp in await agent.list_checkpoints("task-2026-05-09"):
    print(cp.checkpoint_id, cp.turn, cp.created_at)

signal

async def signal(
    self,
    session_id: str,
    name: str,
    payload: Any = None,
) -> None: ...

Send a named message to a run that is waiting for one. The typical case is human approval: an approval handler parks the run with runtime.wait_for_signal(session_id, "approval", timeout_s=...) (which raises TimeoutError when timeout_s passes with no signal), and your web handler or Slack webhook wakes it with agent.signal(...). A signal sent before the run opens its session waits for it, within the runtime’s pending-signal bounds. If the runtime has no signal support, the call logs a warning and does nothing.

await agent.signal("sess-42", "approval", {"action": "allow"})

See Signals.

stream

async def stream(
    self,
    prompt: str,
    *,
    user_id: str | None = None,
    session_id: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    context: RunContext | None = None,
    extra_tools: list[Tool] | None = None,
    output_schema: type[BaseModel] | Any | None = None,
    output_validation_retries: int = 1,
    response_tone: str | None = None,
    effort: str | None = None,
) -> AsyncIterator[Event]: ...

Same loop as run(), exposed as an async generator of Events.

The loop runs as a background task. Events flow through a bounded memory stream, so a slow consumer applies backpressure. Breaking out of the iteration cancels the run cleanly, including any tool call that is mid-flight, and leaves the caller’s task unaffected: its next await runs normally. aclose() on the stream returns once the run has unwound. An error inside the run is raised from the loop as-is, not wrapped in an ExceptionGroup. On trio, close the stream with contextlib.aclosing(agent.stream(...)) before abandoning it.

With an output-stage guardrail wired, text chunks are guarded before they’re yielded; see Streamed text.

Common event.kind values: "started", "model_chunk", "tool_call", "tool_result", "memory_recall", "memory_write", "budget_warning", "budget_exceeded", "permission_ask", "permission_decision", "architecture_event", "error", "completed". event.kind is an EventKind string enum, so it compares equal to these strings.

async for event in agent.stream("plan a Tokyo trip"):
    if event.kind == "model_chunk":
        chunk = event.payload["chunk"]
        if chunk["kind"] == "text":
            print(chunk["text"], end="", flush=True)
    elif event.kind == "tool_call":
        print(f"\n[calling {event.payload['call']['tool']}]")

recall

async def recall(
    self,
    query: str,
    *,
    kind: str = "episodic",
    limit: int = 5,
) -> list[Any]: ...

Shortcut for agent.memory.recall(query, kind=kind, limit=limit). Returns up to limit past episodes matching query.

agent.recall has no user_id parameter, so it only searches the anonymous bucket (episodes stored with user_id=None). For a specific user, call the memory directly. For facts, use the fact store:

recent = await agent.recall("payment refund flow", limit=10)
 
# One user's episodes:
alice_recent = await agent.memory.recall("payment refund flow", user_id="alice")
 
# Facts:
facts = await agent.memory.facts.recall_text("refund policy", user_id="alice")

consolidate

async def consolidate(self) -> int: ...

Run a one-shot consolidation pass over memory, extracting structured facts into the bi-temporal store. Returns the count of new facts added (0 when the memory backend has no fact store).

When auto_extract=True (the default for real models) extraction already runs after every agent.run(). Call this manually when you’ve run many episodes with auto_extract=False and want to process them in one go, for example once a day or before shutdown.

n = await agent.consolidate()
print(f"extracted {n} new facts")

aclose

async def aclose(self, *, timeout: float = 30.0) -> int: ...

Wait for the agent’s background work to finish: fact extraction from earlier runs (auto_extract), which runs after run() returns, and Studio’s queued writes when it’s recording. Returns how many extraction tasks were still running when timeout seconds passed (0 = all done). Call it before reading facts a run just produced, and at shutdown so pending extractions aren’t lost. It doesn’t close the memory backend you passed in, and the agent stays usable. New in 0.17.2.

await agent.run("I'm Alice and I live in Tokyo.", user_id="alice")
await agent.aclose()
facts = await agent.memory.facts.query(user_id="alice")

add_tool

def add_tool(self, item: Tool | Callable[..., object]) -> Tool: ...

Register a tool after construction. Returns the resulting Tool instance (a bare callable is wrapped with @tool first). Raises ConfigError if the tool host isn’t an InProcessToolHost (the default), since MCP registries and custom hosts have no writable registry.

agent.add_tool(my_new_tool)

with_tool

def with_tool(self, fn: Callable[..., object]) -> Callable[..., object]: ...

Decorator form of add_tool. Registers the function as a tool and returns it unchanged, so you can still call it directly. Same host requirement as add_tool.

@agent.with_tool
async def search(query: str) -> str:
    """Search the knowledge base."""
    return f"results for {query}"

remove_tool

def remove_tool(self, name: str) -> bool: ...

Unregister a tool by name. Returns True if removed, False if no such tool was registered. Same host requirement as add_tool.

removed = agent.remove_tool("delete_file")

tools_list

async def tools_list(self) -> list[str]: ...

Return the names of all registered tools. Works with any tool host, including MCP registries.

names = await agent.tools_list()

before_tool

def before_tool(self, fn: PreToolHook) -> PreToolHook: ...

Decorator that registers a pre-tool hook. The hook receives the ToolCall before it runs. Return PermissionDecision.deny_(...) to block the call; return None (or an allow) to let it continue to the next hook and then the permissions layer.

Hooks run in registration order and the first deny wins. Each hook has a 5-second time limit, and a hook that times out or raises denies the call. (Before 0.17 a timeout counted as allow and an exception crashed the run.) HookRegistry(timeout_denies=False) lets timed-out calls through.

from loomflow import PermissionDecision
 
@agent.before_tool
async def review(call):
    if call.tool == "send_email" and "@enemy.com" in str(call.args):
        return PermissionDecision.deny_("blocked by reviewer")
    return None

after_tool

def after_tool(self, fn: PostToolHook) -> PostToolHook: ...

Decorator that registers a post-tool callback. The callback receives the ToolCall and the resulting ToolResult (whether ok or error). It is best-effort: exceptions and timeouts in the callback are swallowed so they can’t change the result.

@agent.after_tool
async def log(call, result):
    print(f"{call.tool} → ok={result.ok}, took={result.duration_ms}ms")

generate_graph

async def generate_graph(
    self,
    path: str | Path | None = None,
    *,
    title: str | None = None,
) -> str: ...

Draw the agent as a Mermaid diagram: its architecture, sub-agents and every agent’s tools. Returns the Mermaid text. With path, it also writes a file, and the extension picks the format: .mmd (raw Mermaid), .md (Markdown with a mermaid fence) or .png / .svg (rendered through mermaid.ink, falling back to a .mmd file if the network call fails). title defaults to the file name, or "Agent".

mermaid_text = await agent.generate_graph()
await agent.generate_graph("agent-graph.md")

See AgentGraph visualization.

Properties

Use these properties for introspection rather than private attributes.

Special methods

__repr__

def __repr__(self) -> str: ...

Returns a one-line summary for dev-time inspection:

>>> agent
Agent(model='claude-opus-4-7', memory=PostgresMemory, runtime=SqliteRuntime, tools=InProcessToolHost, max_turns=50)

Concurrency model

Agent is safe to share across concurrent calls. Each run() / stream() builds its own per-run session; the only shared state on the Agent instance is hook registrations and the tool list. Calls to add_tool / remove_tool are not protected by a lock, so make them during setup, not from a tool in the middle of a run. With a journaled runtime (SqliteRuntime, PostgresRuntime), runs on the same (session_id, user_id) take turns within a process: a second concurrent run waits for the first to end, so their journal entries don’t interleave.

Per-user state (memory partitions, budget buckets, rate-limit buckets) lives on the underlying components, not on the Agent.

Source

loomflow/agent/api.py