Tool Circuit Breaker

Automatic protection against repeatedly-failing tools

Every tool call is automatically protected by a circuit breaker that stops repeated failures from wasting time.

from praisonaiagents import Agent, tool

@tool
def flaky_api(query: str) -> str:
    """Call an external API."""
    return f"Results for: {query}"

agent = Agent(name="Researcher", tools=[flaky_api])
agent.start("Search quantum computing")  # circuit breaker protects automatically

The user triggers a flaky tool; repeated failures open the breaker so later calls fail fast instead of looping on errors.

Quick Start

1. Works by default

   Circuit breaker protection is automatically enabled for every tool call with zero configuration needed.

   from praisonaiagents import Agent, tool
   
   @tool
   def flaky_api(query: str) -> str:
       """Call an external API."""
       return f"Results for: {query}"
   
   agent = Agent(
       name="Researcher",
       instructions="Research the topic",
       tools=[flaky_api],
   )
   
   agent.start("Research quantum computing")

2. Detect open circuit

   When a tool fails 5 times consecutively, subsequent calls return an error dictionary instead of calling the tool.

   # If my_tool fails 5 times in a row, subsequent calls return:
   # {
   #     "error": "Tool 'my_tool' circuit breaker open - too many recent failures", 
   #     "circuit_open": True,
   #     "agent_name": "Researcher",
   #     "session_id": "...",
   #     # Reads the live recovery_timeout: "60s" by default, "30s" if you set 30.0
   #     "remediation": "Wait for recovery_timeout (60s) or investigate recent tool failures."
   # }

3. Tune or reset

   Customize circuit breaker behavior or reset all breakers between test runs.

   from praisonaiagents.tools.circuit_breaker import (
       get_circuit_breaker, CircuitBreakerConfig, reset_all_circuit_breakers
   )
   
   # Register config for a specific tool BEFORE the first call.
   # Breakers are keyed per agent instance, so build the key from id(agent).
   breaker = get_circuit_breaker(
       f"tool_{id(agent)}_my_tool",
       CircuitBreakerConfig(
           failure_threshold=3,  # Open after 3 failures instead of 5
           recovery_timeout=30.0,  # Try again after 30s instead of 60s
       )
   )
   
   # Reset all circuit breakers
   reset_all_circuit_breakers()

How It Works

The OPEN → HALF_OPEN transition is driven by recovery_timeout or by an enable_health_check=True probe reporting healthy — whichever fires first. Since PR #5330, the probe runs on both sync and async paths.

The async path (achat() / astart()) performs the same OPEN → HALF_OPEN → CLOSED transitions through the shared helpers — _circuit_breaker_precheck runs the pre-check, _circuit_breaker_record records the outcome — so this diagram applies uniformly to chat/start and achat/astart.

The mechanism is identical to sync, but delivered through CircuitBreaker.acall(...) rather than CircuitBreaker.call(...) so it never blocks the event loop (PR #4469). An error-dict result counts as a failure just like a raised exception — the async path surfaces it to the breaker through a _ToolFailure sentinel, mirroring the sync _ToolFailure wrapper — so five error-dict results open the breaker exactly as five raises would.

Async workflows using asyncio.gather(...) get the same parity: one failing tool opens only its own per-agent breaker and short-circuits, so it won’t consume retries across every gathered task.

Configuration Options

Custom Health Checks

Circuit breakers close themselves as soon as the backing service is healthy again — you just tell the breaker how to check. Provide any object with a health_check() method (sync) or ahealth_check() method (async), and the breaker will probe it on the interval you configured. It now works the same way whether your agent calls tools via agent.start() or await agent.astart().

Pass a health check that implements HealthCheckProtocol and the breaker (or HealthMonitor) probes it on the configured interval.

from typing import Protocol, runtime_checkable

@runtime_checkable
class HealthCheckProtocol(Protocol):
    def health_check(self) -> bool: ...          # sync — return True if healthy
    async def ahealth_check(self) -> bool: ...   # async — return True if healthy

The protocol is exported from praisonaiagents.tools.circuit_breaker. Attach an implementation when creating a breaker:

from praisonaiagents.tools.circuit_breaker import CircuitBreaker

class DatabaseHealth:
    def health_check(self) -> bool:
        return ping_database()

breaker = CircuitBreaker("database", health_check=DatabaseHealth())

• Canonical method names are health_check / ahealth_check.
• Legacy names check_health / acheck_health remain supported — the monitor tries the canonical names first, then falls back.
• A plain sync callable (e.g. lambda: True) also works — it is treated as the sync health check.

What triggers health-check recovery

A health probe recovers an open circuit as soon as it reports healthy, on both sync and async paths.

What DOES Trip the Breaker

Async tool timeouts count. When an async tool is cancelled by the per-call tool_timeout, the timeout is recorded on the breaker just like a raised exception or an error-dict result. Five consecutive timeouts open the circuit; the sixth call short-circuits with circuit_open: True.

Remote MCP failures count. Failures on HTTPStreamMCPTool, SSEMCPTool, and WebSocketMCPTool now return structured error dicts, so they count toward the per-agent breaker.

What Does NOT Trip the Breaker

Circuit breakers ignore certain error types to avoid false positives:

• approval_denied — user rejected the tool call
• permission_denied — access control failure
• approval_error — approval workflow error
• policy_denied — policy engine deny
• guardrail_denied — guardrail-blocked tool call

These exclusions apply to both the sync and async execution paths. On the async path they’re checked as keys on the result dict returned by the tool, matching the sync _ToolFailure wrapper’s exclusions — so a denial never counts toward opening the breaker on either path.

Caller-error exclusion (bad input)

The breaker also treats caller programming errors as neutral. They count as neither failures nor successes, so they can neither open the breaker nor close a half-open one. Two triggers:

• The tool raises one of nine caller-error types: ValueError, TypeError, AttributeError, LookupError (so KeyError and IndexError too), FileNotFoundError, FileExistsError, IsADirectoryError, NotADirectoryError, or PermissionError. A bare OSError, ConnectionError, TimeoutError, RuntimeError, or any other exception still counts as a failure.
• The tool returns an error dict carrying _praison_retryable: False. The retry layer tags schema mismatches with this flag, so the breaker and the retry policy agree on what is a caller error vs a service outage.

This applies identically to the sync (CircuitBreaker.call) and async (CircuitBreaker.acall) paths. Under the hood, caller errors are carried through the breaker as a _NeutralOutcome sentinel (private to the module) and unwrapped before the result reaches your agent — you never need to touch it.

from praisonaiagents import Agent

def get_order(order_id: str) -> str:
    """Look up an order. Raises on an unknown id."""
    if not order_id.startswith("A"):
        raise ValueError(f"unknown order {order_id}")
    return f"order {order_id}: shipped"

agent = Agent(name="support", instructions="Look up orders.", tools=[get_order])

# Six invalid ids from one user: each raises ValueError, counted as neutral.
for i in range(6):
    try:
        agent.execute_tool("get_order", {"order_id": f"bad{i}"})
    except Exception:
        pass

# A different user's valid id still succeeds — the breaker never opened.
assert agent.execute_tool("get_order", {"order_id": "A100"}) == "order A100: shipped"

Lifecycle

Per-agent breakers are pruned from the global registry automatically when the Agent is garbage-collected — a weakref.finalize callback is registered when each breaker is created. This closes the CPython id()-reuse window without requiring agent.close() / agent.aclose() to be called explicitly.

Calling agent.close() / agent.aclose() triggers _cleanup_circuit_breakers(), which removes every tool_{id(agent)}_* entry from the registry earlier and deterministically — keeping it bounded and preventing a reused id from inheriting a stale OPEN breaker. See Agent Lifecycle Cleanup for the full teardown story.

Async tools

A user calling an unreliable async tool via await agent.achat(...) sees the breaker open after five raised exceptions (or five error-dict results); further calls short-circuit with a circuit_open: True error dict, terminating the async retry loop immediately.

import asyncio
from praisonaiagents import Agent, tool

@tool
async def flaky_upstream(query: str) -> str:
    """Async tool that may raise on a bad upstream."""
    raise RuntimeError("upstream 503")

agent = Agent(name="Researcher", tools=[flaky_upstream])

async def main():
    # After 5 raised failures the breaker opens; the 6th call
    # short-circuits with {"circuit_open": True} instead of retrying.
    await agent.achat("Search quantum computing")

asyncio.run(main())

An async tool that hangs past tool_timeout counts the same way — the timeout is recorded on the breaker, so five consecutive timeouts open the circuit.

import asyncio
from praisonaiagents import Agent, tool
from praisonaiagents.config import ToolConfig

@tool
async def slow_upstream(query: str) -> str:
    """Async tool that hangs past the tool timeout."""
    await asyncio.sleep(60)
    return "never"

agent = Agent(
    name="Researcher",
    tools=[slow_upstream],
    tool_config=ToolConfig(timeout=2),
)

async def main():
    # After 5 consecutive timeouts, the 6th call short-circuits.
    for _ in range(6):
        await agent.achat("Search quantum computing")

asyncio.run(main())

Breakers are per Agent instance (tool_{id(self)}_{function_name}), so two agents sharing a tool name never share a breaker — an OPEN breaker on agent_a never blocks agent_b.

import asyncio
from praisonaiagents import Agent, tool

@tool
async def shared_search(query: str) -> str:
    """Same tool name, two independent agents."""
    return f"Results for: {query}"

agent_a = Agent(name="A", tools=[shared_search])
agent_b = Agent(name="B", tools=[shared_search])

async def main():
    # agent_a's breaker opening on shared_search never blocks agent_b.
    await asyncio.gather(
        agent_a.achat("Search topic A"),
        agent_b.achat("Search topic B"),
    )

asyncio.run(main())

Common Patterns

from praisonaiagents.tools.circuit_breaker import _get_global_registry

registry = _get_global_registry()
prefix = f"tool_{id(agent)}_"

for name in registry.list_services():
    if name.startswith(prefix):
        breaker = registry.get(name)
        print(name, breaker.state, breaker.stats.failure_count)

from praisonaiagents.tools.circuit_breaker import get_circuit_breaker

breaker = get_circuit_breaker(f"tool_{id(agent)}_my_tool")
print(f"State: {breaker.state}")
print(f"Failure count: {breaker.stats.failure_count}")
print(f"Total requests: {breaker.stats.total_requests}")
print(f"Rejected requests: {breaker.stats.rejected_requests}")

from praisonaiagents.tools.circuit_breaker import get_circuit_breaker, CircuitBreakerConfig

# Strict config for unreliable external APIs
strict_config = CircuitBreakerConfig(
    failure_threshold=2,    # Open quickly
    recovery_timeout=120.0,  # Wait longer before retry
)
api_breaker = get_circuit_breaker(f"tool_{id(agent)}_external_api", strict_config)

# Lenient config for internal tools
lenient_config = CircuitBreakerConfig(
    failure_threshold=10,   # Allow more failures
    recovery_timeout=30.0,  # Recover quickly
)
internal_breaker = get_circuit_breaker(f"tool_{id(agent)}_internal_db", lenient_config)

from praisonaiagents.tools.circuit_breaker import reset_all_circuit_breakers

# Reset all breakers (useful for test cleanup or after deployment)
reset_all_circuit_breakers()

# Or reset individual breakers (use the per-agent key)
from praisonaiagents.tools.circuit_breaker import get_circuit_breaker
breaker = get_circuit_breaker(f"tool_{id(agent)}_my_tool")
breaker.reset()

Best Practices

Related