When tools fail, your agent degrades gracefully instead of looping.
Stop agent tool loops, enforce cost caps, and degrade gracefully. One decorator. Zero dependencies. Works with Claude Code, MCP, OpenAI Agents SDK, LangChain, and CrewAI.
Your agent calls a failing API. It retries 47 times. That burns $12, wastes 90 seconds, and your user stares at a spinner.
from agent_circuit import CircuitBreaker, CircuitDegraded
breaker = CircuitBreaker(max_failures=3, cost_global_limit=5.0)
@breaker.protect("search_api", cost_per_call=0.26)
def search(query: str) -> str:
return api.search(query)
try:
result = search("latest news")
except CircuitDegraded as e:
print(e.result.message) # "Search API unavailable"
print(e.result.suggestion) # "Skip and proceed with available info"Result: 48 calls reduced to 4. $11.70 saved. 85 seconds saved.
pip install agent-circuitpython -m agent_circuit demoNo API keys needed. Shows before/after comparison with simulated failures.
One command setup. Protects all tool calls automatically:
agent-circuit install-hookThis adds PreToolUse and PostToolUse hooks to .claude/settings.local.json. Restart Claude Code to activate.
Check status:
agent-circuit statusAdd circuit breaking to any MCP server:
from fastmcp import FastMCP
from agent_circuit.integrations.mcp import AgentCircuitMiddleware
mcp = FastMCP("my-server")
mcp.add_middleware(AgentCircuitMiddleware(max_failures=3))
@mcp.tool()
def search(query: str) -> str:
return api.search(query)Install: pip install agent-circuit[mcp]
from agents import Agent
from agent_circuit.integrations.openai_agents import protect_tools
agent = Agent(
name="my-agent",
tools=protect_tools([search, fetch], max_failures=3),
)Install: pip install agent-circuit[openai]
from agent_circuit.integrations.langchain import CircuitBreakerMiddleware
middleware = CircuitBreakerMiddleware(max_failures=3)
graph = create_react_agent(model, tools, middleware=[middleware])Install: pip install agent-circuit[langchain]
from agent_circuit.integrations.crewai import protect_tool
protected_search = protect_tool(search_tool, max_failures=3)
crew = Crew(agents=[agent], tasks=[task], tools=[protected_search])Install: pip install agent-circuit[crewai]
from agent_circuit import CircuitBreaker, CircuitDegraded
breaker = CircuitBreaker(max_failures=3)
@breaker.protect("my_tool", cost_per_call=0.05)
def my_tool(arg: str) -> str:
return call_api(arg)
@breaker.protect("async_tool", cost_per_call=0.10)
async def async_tool(arg: str) -> str:
return await call_api(arg)| Protection | What it does |
|---|---|
| Circuit Breaker | Stops calling after N failures. Auto-probes with exponential backoff. |
| Loop Detection | Fingerprints all args. Catches retry storms with identical calls. |
| Cost Budget | Per-tool and global limits. Atomic check-and-record (no race conditions). |
When a circuit opens, your agent gets an error with instructions:
from agent_circuit import SkipWithMessage, ReturnCached, EscalateToHuman
@breaker.protect("search", strategy=SkipWithMessage("Search is down."))
@breaker.protect("pricing", strategy=ReturnCached({"price": 9.99}))
@breaker.protect("deploy", strategy=EscalateToHuman())Access cached value: except CircuitDegraded as e: e.result.cached_value
@breaker.protect("cheap_tool", max_failures=5, reset_timeout_s=10.0)
@breaker.protect("expensive_tool", max_failures=1, reset_timeout_s=300.0)Test before enforcing:
breaker = CircuitBreaker(shadow_mode=True) # Logs only, never blocks
breaker.shadow_mode = False # Now enforcebreaker = CircuitBreaker(
on_circuit_open=lambda name, stats: alert(f"Circuit opened: {name}"),
on_degrade=lambda name, result: metrics.inc("degraded", tool=name),
on_budget_exceeded=lambda name, exc: log.warn(f"Budget: {exc}"),
on_loop_detected=lambda name, signal: log.warn(f"Loop: {name}"),
)breaker = CircuitBreaker(
max_failures=3, # Open after N consecutive failures
reset_timeout_s=30.0, # Probe interval (exponential backoff)
backoff_multiplier=2.0, # Double timeout on each re-open
max_reset_timeout_s=300.0, # Cap on backoff
loop_window_s=60.0, # Loop detection time window
loop_threshold=5, # Trigger after N identical calls
cost_per_tool_limit=2.0, # Max $ per tool
cost_global_limit=10.0, # Max $ total
max_records=10000, # Bounded history (no memory leaks)
shadow_mode=False, # Log-only mode
)stats = breaker.stats["search_api"]
stats.state # BreakerState.OPEN
stats.failure_rate # 1.0 (actual calls only)
stats.actual_calls # 3
stats.degraded_calls # 47
stats.open_count # 2
breaker.cost_tracker.total_spent # 0.78
breaker.is_circuit_open("search_api") # True/Falseimport logging
logging.getLogger("agent_circuit").setLevel(logging.DEBUG)agent-circuit demo Run before/after demo
agent-circuit install-hook Install Claude Code hooks
agent-circuit status Show circuit breaker state
agent-circuit help Show help
agent-circuit is part of a suite of zero-dependency Python tools for the AI agent era:
| Tool | What it does |
|---|---|
| ghostlines | Find code your team merged but never understood |
| agent-guard | Block prompt injection and path traversal at the tool boundary |
| agent-bill | Track LLM costs with itemized receipts |
| crowdllm | Multi-model voting for better answers |
| vcr-llm | Record and replay LLM conversations for testing |
| singleflight-agents | Deduplicate parallel agent tool calls |
MIT