	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > ADK Agent Testing > Testing ADK Agent Circuit Breaker Pattern

Testing ADK Agent Circuit Breaker Pattern

Author: Venkata Sudhakar

Circuit breaker testing verifies that an ADK agent stops calling a failing downstream service after a threshold of consecutive failures and returns a fast fallback response instead of cascading timeouts. ShopMax India's inventory and payment agents use circuit breakers to protect customers in Delhi and Bangalore from hanging requests when the warehouse API or payment gateway goes down during peak traffic - the breaker must open quickly, serve fallbacks, and close again once the service recovers.

A CircuitBreaker wraps a tool call and tracks consecutive failure counts. When failures reach the threshold, the breaker moves to OPEN state and raises CircuitOpenError immediately without calling the service. After a cooldown period, it moves to HALF_OPEN and allows one probe call. If that succeeds, it returns to CLOSED; if it fails, it reopens. Tests cover the three state transitions: CLOSED to OPEN on repeated failures, OPEN fast-fail behavior, and HALF_OPEN probe recovery.

The example below implements a three-state CircuitBreaker, drives it through all state transitions using a mock tool that fails on demand, and asserts correct behavior at each state.

import pytest
from dataclasses import dataclass, field

class CircuitOpenError(Exception):
    pass

@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    _failures: int = 0
    _state: str = "CLOSED"

def call(self, fn, *args):
        if self._state == "OPEN":
            raise CircuitOpenError(f"Circuit OPEN after {self._failures} failures")
        try:
            result = fn(*args)
            self._failures = 0
            if self._state == "HALF_OPEN":
                self._state = "CLOSED"
            return result
        except Exception as e:
            self._failures += 1
            if self._failures >= self.failure_threshold:
                self._state = "OPEN"
            raise

def allow_probe(self):
        self._state = "HALF_OPEN"

def failing_inventory_api(product_id: str) -> dict:
    raise ConnectionError("Warehouse API unavailable")

def healthy_inventory_api(product_id: str) -> dict:
    return {"product_id": product_id, "stock": 10, "city": "Delhi"}

def test_circuit_opens_after_threshold():
    cb = CircuitBreaker(failure_threshold=3)
    for i in range(3):
        with pytest.raises(ConnectionError):
            cb.call(failing_inventory_api, "PROD-001")
    assert cb._state == "OPEN"
    print(f"Circuit opened after 3 failures")

def test_open_circuit_fast_fails():
    cb = CircuitBreaker(failure_threshold=3)
    for _ in range(3):
        with pytest.raises(ConnectionError):
            cb.call(failing_inventory_api, "PROD-001")
    with pytest.raises(CircuitOpenError, match="Circuit OPEN"):
        cb.call(failing_inventory_api, "PROD-001")
    print("Open circuit fast-failed without calling service")

def test_half_open_probe_recovers():
    cb = CircuitBreaker(failure_threshold=3)
    for _ in range(3):
        with pytest.raises(ConnectionError):
            cb.call(failing_inventory_api, "PROD-001")
    cb.allow_probe()
    result = cb.call(healthy_inventory_api, "PROD-001")
    assert cb._state == "CLOSED"
    assert result["stock"] == 10
    print(f"Circuit closed after successful probe: {result}")

It gives the following output,

Circuit opened after 3 failures
Open circuit fast-failed without calling service
Circuit closed after successful probe: {'product_id': 'PROD-001', 'stock': 10, 'city': 'Delhi'}
3 passed in 0.05s

Set the failure threshold conservatively (3-5) to avoid opening on transient errors, and tune the cooldown period to match the downstream service's typical recovery time - too short a cooldown means repeated half-open probes that re-open the circuit before the service is ready. Emit a metric every time the circuit opens so the on-call team gets an alert and can correlate circuit state with downstream service health dashboards in Grafana.

Send your comments, suggestions or queries regarding this site to [email protected].