Production tour

The Quickstart gets an agent working. This page adds what you need before real users touch it: memory that survives a restart, runs that recover from a crash, telemetry, an audit log, permissions, a filesystem sandbox and spending limits. Each section is independent, so read the ones you need. The last section puts them all in one agent.

Examples use model="claude-opus-4-7"; any model works.

Memory

Give an agent memory= and it remembers past conversations and facts about each user. Pass a URL and loomflow picks the backend:


from loomflow import Agent
 
agent = Agent("You are helpful.", model="claude-opus-4-7", memory="sqlite:./bot.db")

`memory=`	Backend
`"inmemory"`	In process; the default. Lost on restart.
`"sqlite:./bot.db"`	One SQLite file on disk.
`"chroma:./chroma-db"`	A local Chroma vector store.
`"postgres://user:pw@host/db"`	Postgres with pgvector.
`"redis://localhost:6379/0"`	Redis.

Every backend keeps each user_id separate. Postgres and Redis connect lazily, on the first run. For settings the URL can’t express, pass a dict:


from loomflow import Agent
 
agent = Agent("You are helpful.", model="claude-opus-4-7", memory={
    "backend": "chroma",
    "path": "./chroma-db",
    "namespace": "tenant_a",
    "with_facts": True,
})

Memory compares the backends and covers embedders.

Facts it learns on its own

With a real model, every run also pulls short facts out of the conversation (for example alice lives in Tokyo) and stores them for that user. Later runs recall them automatically.


import asyncio
from loomflow import Agent
 
async def main():
    agent = Agent(
        "You are a personal assistant.",
        model="claude-opus-4-7",
        memory="sqlite:./bot.db",
    )
    await agent.run("Hi, I'm Alice and I live in Tokyo.", user_id="alice")
    await agent.aclose()   # extraction runs in the background; wait for it
 
    profile = await agent.memory.profile(user_id="alice")
    print(profile.fact_count, profile.sample_facts)
 
    # Days later, in a new process with the same database:
    result = await agent.run("Where do I live?", user_id="alice")
    print(result.output)   # mentions Tokyo
 
asyncio.run(main())

Extraction is on by default for OpenAI, Anthropic and LiteLLM models and off for the echo and scripted test models. Turn it off with Agent(..., auto_extract=False). It runs after run() returns, so call agent.aclose() before reading the facts a run just produced, and at shutdown.

Facts keep their history. When a new fact contradicts an old one (same subject and predicate, different value), the old one is closed rather than deleted, so you can still ask what was true on a past date with agent.memory.facts.query(user_id=..., subject=..., valid_at=...).

Recover from a crash

A runtime saves each model call and tool call as it finishes. If the process dies, running the same session again replays the saved steps instead of paying for them twice. (Only an unfinished run replays: a run that completed is never replayed into the next request.)


import asyncio
from loomflow import Agent
from loomflow.runtime import SqliteRuntime
 
async def main():
    agent = Agent(
        "You are a research assistant.",
        model="claude-opus-4-7",
        runtime=SqliteRuntime("./journal.db"),
    )
    # Give the run a session_id you can find again.
    result = await agent.run("Summarise the open issues.", session_id="issues-2026-10-19")
 
    # After a crash and restart, the same session picks up where it stopped:
    result = await agent.resume(session_id="issues-2026-10-19")
    print(result.output)
 
asyncio.run(main())

Replay and resume and Checkpoints go further, including Postgres.

Telemetry


import asyncio
from loomflow import Agent
from loomflow.observability import ConsoleTelemetry
 
async def main():
    agent = Agent("You are helpful.", model="claude-opus-4-7", telemetry=ConsoleTelemetry())
    await agent.run("hi")   # spans print to stderr as they happen
 
asyncio.run(main())

ConsoleTelemetry, FileTelemetry (JSON lines on disk), InMemoryTelemetry (for tests) and MultiTelemetry (send to several) need no collector. OTelTelemetry sends the same spans to Honeycomb, Datadog or any OpenTelemetry backend. Spans are named loom.run, loom.turn, loom.model.complete (loom.model.stream when you stream), loom.tool and loom.workflow.step. In an OpenTelemetry backend the model, tool and run spans follow the GenAI conventions instead: chat {model}, execute_tool {tool} and invoke_agent, with the loom.* name kept in the loom.span.name attribute. See Observability.

Audit log


import asyncio
from loomflow import Agent
from loomflow.security import FileAuditLog
 
async def main():
    audit = FileAuditLog("./audit.jsonl", secret="change-me")
    agent = Agent("You are helpful.", model="claude-opus-4-7", audit_log=audit)
 
    result = await agent.run("anything", user_id="alice")
    for entry in await audit.query(session_id=result.session_id):
        print(entry)
 
asyncio.run(main())

Every run start, tool call, tool result and run end is written as one signed line, chained to the line before it, so you can prove later what the agent did and that no entry was edited, removed or reordered (await audit.verify(); catching entries cut off the end needs a head you saved earlier). See Audit log.

Permissions and hooks

Decide which tools may run, and inspect each call before it does:


from loomflow import Agent, Mode, StandardPermissions
from loomflow.core.types import PermissionDecision
 
agent = Agent(
    "You are an operations assistant.",
    model="claude-opus-4-7",
    permissions=StandardPermissions(mode=Mode.DEFAULT, denied_tools=["format_disk"]),
)
 
@agent.before_tool
async def review(call):
    if call.tool == "send_email" and "@competitor.com" in str(call.args):
        return PermissionDecision.deny_("external email blocked")
    return None   # no objection
 
@agent.after_tool
async def log(call, result):
    print(f"{call.tool}: ok={result.ok}")

For tools marked @tool(destructive=True), the default policy answers ask. An approval handler decides who gets asked:


from loomflow import Agent, Mode, StandardPermissions
 
async def approve(call, user_id: str | None) -> bool:
    return await my_slack_app.request_approval(call.tool, user_id)   # your code
 
agent = Agent(
    "You are an operations assistant.",
    model="claude-opus-4-7",
    permissions=StandardPermissions(mode=Mode.DEFAULT),
    approval_handler=approve,
)

Without an approval handler, ask is treated as deny, so the agent never skips the check silently. See Permissions.

Keep file tools inside a folder


from pathlib import Path
 
from loomflow import Agent, tool
from loomflow.security import FilesystemSandbox
from loomflow.tools import InProcessToolHost
 
ROOT = Path("./workspace")
 
@tool
def read_file(path: str) -> str:
    """Read a file's contents."""
    return (ROOT / path).read_text()   # relative paths open under the root
 
host = InProcessToolHost([read_file])
sandbox = FilesystemSandbox(host, roots=[ROOT])
 
agent = Agent("You help with files.", model="claude-opus-4-7", tools=sandbox)

Any path argument outside ./workspace is refused, and symlinks are resolved first so they can’t escape. A relative path is checked against the first root (pass base= to change that), so the tool opens it from the same folder. (Before 0.17 relative paths were checked against the process’s working directory.) Sandboxes also covers running tools in a subprocess or an OS sandbox.

Spending limits


from datetime import timedelta
from loomflow import Agent
from loomflow.governance.budget import BudgetConfig, StandardBudget
 
agent = Agent(
    "You are a research assistant.",
    model="claude-opus-4-7",
    budget=StandardBudget(BudgetConfig(
        max_tokens=200_000,
        max_cost_usd=5.0,
        max_wall_clock=timedelta(minutes=10),
        soft_warning_at=0.8,   # emit a budget_warning event at 80%
    )),
)

When a limit is hit the run stops cleanly: result.interrupted is True and result.interruption_reason says which limit, for example "budget:max_tokens". max_wall_clock counts time the agent spends running, not time since the budget was created, so an idle long-lived agent isn’t locked out. Limits can also be set per user; see Budget caps.

All of it in one agent

This needs Postgres, uvx and npx for the two MCP servers, and the otel and mcp extras (pip install 'loomflow[anthropic,mcp,otel,postgres]').


import asyncio
from datetime import timedelta
 
from loomflow import Agent, Mode, StandardPermissions, tool
from loomflow.architecture.tool_host_wrappers import ExtendedToolHost
from loomflow.governance.budget import BudgetConfig, StandardBudget
from loomflow.mcp import MCPRegistry, MCPServerSpec
from loomflow.observability import OTelTelemetry
from loomflow.runtime import SqliteRuntime
from loomflow.security import FileAuditLog
 
@tool
async def web_search(query: str) -> str:
    """Search the web."""
    return "..."   # call your search API here
 
mcp_servers = MCPRegistry([
    MCPServerSpec.stdio("git", "uvx", ["mcp-server-git", "--repository", "."]),
    MCPServerSpec.stdio("fs", "npx", ["-y", "@modelcontextprotocol/server-filesystem", "."]),
])
 
async def main():
    agent = Agent(
        "You are a research assistant. Cite your sources.",
        model="claude-opus-4-7",
        memory="postgres://user:pw@db.internal/loomflow",
        runtime=SqliteRuntime("./journal.db"),
        tools=ExtendedToolHost(mcp_servers, [web_search]),
        permissions=StandardPermissions(mode=Mode.DEFAULT),
        budget=StandardBudget(BudgetConfig(
            max_tokens=200_000,
            max_cost_usd=5.0,
            max_wall_clock=timedelta(minutes=10),
        )),
        audit_log=FileAuditLog("./audit.jsonl", secret="change-me"),
        telemetry=OTelTelemetry(),
    )
 
    async for event in agent.stream(
        "Research recent advances in agent frameworks.",
        user_id="user_42",
        session_id="research_2026_10_19",
    ):
        if event.kind == "model_chunk" and event.payload["chunk"]["kind"] == "text":
            print(event.payload["chunk"]["text"], end="", flush=True)
 
asyncio.run(main())

Memory keeps what it learns about user_42, the runtime can recover the session after a crash, every step is in the audit log and your tracing backend, and the budget caps what one run can spend.

Where to next: Production for multi-tenancy, secrets and serving over HTTP, or Recipes for complete patterns.