RunFuse ⚡

Runtime circuit breakers for AI agents. Trip fuses on runaway cost, infinite loops, and retry storms — before the invoice does it for you.

from runfuse import Fuse, FusePolicy
from anthropic import Anthropic

fuse = Fuse(FusePolicy(
    max_cost_usd=5.00,            # hard spend ceiling for this run
    max_steps=50,                 # max LLM calls
    max_identical_tool_calls=10,  # loop detector
    max_llm_errors=5,             # retry-storm detector
))
client = fuse.wrap(Anthropic())   # AsyncAnthropic / (Async)OpenAI work too

with fuse.run("research-task"):
    ...  # your agent loop — trips raise FuseTripped, run stops

with fuse.run("untrusted-task", policy=FusePolicy(max_cost_usd=0.50)):
    ...  # per-run override: tighter budget for this run only

Why

Agents aren't chat. One user request fans out into dozens–hundreds of autonomous LLM calls. When an agent enters a retry loop or a planning spiral, nothing stops it — it burns tokens until someone notices the bill.

Existing tools don't cover this:

• LLM gateways (LiteLLM, Portkey, Helicone) enforce budgets per API key. They see isolated HTTP requests — they can't tell "this run is looping" or kill one sick agent without killing all of them.
• Observability platforms (LangSmith, Langfuse) are passive — they show you the fire afterwards.

RunFuse enforces at the run level, actively, mid-flight. The with fuse.run(...) scope is the identity gateways don't have.

Install

pip install runfuse            # core (pydantic only)
pip install runfuse[anthropic] # + Anthropic SDK
pip install runfuse[openai]    # + OpenAI SDK
pip install runfuse[langgraph] # + LangGraph/LangChain callback
pip install runfuse[dashboard] # + live web dashboard

Using LangGraph instead of a raw SDK client? Same fuse, one callback:

from runfuse.interceptors.langgraph import RunFuseCallback

with fuse.run("graph-task"):
    graph.invoke(inputs, config={"callbacks": [RunFuseCallback(fuse)]})

Tools the interceptor can't see (MCP server handlers, framework-executed tools)? Guard them directly — trips fire before the tool body runs:

@fuse.guard_tool
def web_search(query: str) -> str: ...

Fuses

Every hard limit also has a soft threshold (default 80%): a warning callback fires before the fuse blows, once per fuse per run.

Costs come from a built-in pricing table (override via Fuse(pricing={...})). Unknown models count as $0 with a logged warning — set strict_pricing=True on the policy to hard-trip instead of running cost-blind.

fuse = Fuse(
    policy,
    on_soft_trip=lambda verdict, state: notify_slack(verdict),
)

Trip semantics — honest by design

A trip raises FuseTripped at the next interception point (the next LLM call). It cannot stop a tool mid-execution — nothing can, and tools that claim otherwise are lying. The same honesty applies to blind spots: streaming calls are refused loudly (stream=True raises) rather than silently recorded as $0, until true streaming accounting ships. The exception carries the full verdict and run state:

try:
    with fuse.run("job-42") as state:
        agent_loop(client)
except FuseTripped as trip:
    print(trip.verdict)        # [hard] max_cost_usd: spent $5.0012 >= $5.0000
    print(state.steps)         # 37
    print(state.cost_usd)      # 5.0012

See it trip

No API key needed:

git clone https://github.com/akshat333-debug/RunFuse && cd RunFuse
uv sync --all-extras
uv run runfuse demo                     # scripted runaway, fuse trips
uv run runfuse report                   # per-run cost/step/trip summary
uv run runfuse dashboard                # live web UI (localhost:8321)

The dashboard's abort button writes a flag the fuse checks at the run's next interception point — same honest semantics as any other trip. Approve is implicit: a soft-warned run keeps going unless you kill it.

How it works

fuse.wrap(client) → every LLM call emits an Event
                       │
                       ▼
        RunState (cost, steps, errors, tool-call counts)
                       │
                       ▼
        PolicyEngine → ok | soft trip (warn) | hard trip (raise)
                       │
                       ▼
        SQLite event store → `runfuse report` / dashboard

Tool calls are parsed from LLM response payloads (tool_use blocks / tool_calls), so loop detection works from the SDK wrapper alone — no framework hooks required. Runs are scoped via contextvars, so concurrent async runs stay independent.

Development

uv sync --all-extras
uv run pytest
uv run mypy src
uv run ruff check src tests

License

MIT