Tool Circuit Breaker

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

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

from praisonaiagents.tools.circuit_breaker import CircuitBreaker

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

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

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())

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())

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())