When you rely on AI APIs for production applications, a single provider going down can crash your entire service. After spending three months building resilient AI infrastructure at scale, I discovered that combining health checks with circuit breakers transforms unpredictable API failures into manageable events. This tutorial walks you through building a production-ready fault-tolerant system from scratch—no prior DevOps experience required.
Why You Need API Fault Tolerance
Imagine your customer support chatbot suddenly stops responding because one AI provider hit their rate limit. Without a circuit breaker, your system keeps hammering that failing endpoint until it times out, creating a cascade of errors that affects every user. This is exactly what happened to me during a product launch—three hours of downtime because I had no fallback strategy.
Modern AI infrastructure demands resilience. HolySheep AI addresses this by offering multi-provider access through a unified endpoint, but even with such redundancy, your application needs intelligent routing to handle failures gracefully. The solution combines two patterns: health checks that monitor provider status in real-time, and circuit breakers that automatically reroute traffic when something goes wrong.
Understanding Health Checks and Circuit Breakers
What Are Health Checks?
A health check is a lightweight test that verifies whether an API endpoint is responding correctly. Think of it as a doctor taking your pulse—it quickly tells you if something is wrong without running extensive tests. Your system sends a simple request to each provider every few seconds, measuring response time and success rates.
What Is a Circuit Breaker?
Inspired by electrical breakers, this pattern "trips" when failures exceed a threshold. When the circuit opens, requests automatically route to healthy providers instead of wasting time on failing ones. After a cooldown period, the breaker tests the failed provider—if it recovers, the circuit closes and normal operation resumes.
Building Your First Fault-Tolerant System
Step 1: Environment Setup
Create a new Python project with the necessary dependencies:
pip install requests httpx asyncio aiohttp tenacityFor this tutorial, I'll use Python 3.10+ with async support for maximum performance. HolySheep AI's infrastructure delivers <50ms average latency, which means your health checks complete in under 100ms even when testing multiple providers simultaneously.
Step 2: Define Your Provider Configuration
import httpx
import asyncio
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import time
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
@dataclass
class Provider:
name: str
base_url: str
api_key: str
status: ProviderStatus = ProviderStatus.HEALTHY
failure_count: int = 0
last_check: float = field(default_factory=time.time)
success_count: int = 0
# Circuit breaker settings
failure_threshold: int = 3
recovery_timeout: int = 30 # seconds
half_open_max_calls: int = 2
async def health_check(self, timeout: float = 2.0) -> bool:
"""Perform a lightweight health check against this provider."""
try:
async with httpx.AsyncClient(timeout=timeout) as client:
# Test with a minimal request
response = await client.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "ping"}],
"max_tokens": 5
}
)
is_healthy = response.status_code == 200
# Update statistics
self.last_check = time.time()
if is_healthy:
self.success_count += 1
self.failure_count = 0
else:
self.failure_count += 1
self.success_count = 0
return is_healthy
except Exception as e:
self.failure_count += 1
self.success_count = 0
self.last_check = time.time()
return False
Initialize your providers
providers: Dict[str, Provider] = {
"holysheep": Provider(
name="HolySheep AI",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
failure_threshold=3,
recovery_timeout=30
),
"deepseek": Provider(
name="DeepSeek V3.2",
base_url="https://api.deepseek.com/v1",
api_key="YOUR_DEEPSEEK_API_KEY",
failure_threshold=3,
recovery_timeout=30
),
"gemini": Provider(
name="Gemini 2.5 Flash",
base_url="https://generativelanguage.googleapis.com/v1beta",
api_key="YOUR_GEMINI_API_KEY",
failure_threshold=5,
recovery_timeout=45
),
}Step 3: Implement the Circuit Breaker Logic
This is where the magic happens. The circuit breaker monitors each provider's health and decides when to route traffic away:
class CircuitBreaker:
def __init__(self, provider: Provider):
self.provider = provider
self.state = "closed" # closed, open, half-open
self.last_failure_time: Optional[float] = None
def should_allow_request(self) -> bool:
"""Determine if requests should be sent to this provider."""
if self.state == "closed":
return True
if self.state == "open":
# Check if recovery timeout has elapsed
if time.time() - self.last_failure_time >= self.provider.recovery_timeout:
self.state = "half-open"
return True
return False
if self.state == "half-open":
return True
return False
def record_success(self):
"""Called when a request succeeds."""
if self.state == "half-open":
self.state = "closed"
self.provider.failure_count = 0
elif self.state == "closed":
self.provider.success_count += 1
def record_failure(self):
"""Called when a request fails."""
self.provider.failure_count += 1
self.last_failure_time = time.time()
if self.state == "half-open":
self.state = "open"
elif self.provider.failure_count >= self.provider.failure_threshold:
self.state = "open"
Create circuit breakers for all providers
breakers: Dict[str, CircuitBreaker] = {
name: CircuitBreaker(provider)
for name, provider in providers.items()
}
async def get_healthy_provider() -> Optional[Provider]:
"""Find the best available provider using health-weighted selection."""
candidates = []
for name, breaker in breakers.items():
provider = providers[name]
if not breaker.should_allow_request():
continue
# Calculate health score based on recent performance
if provider.last_check > 0:
time_since_check = time.time() - provider.last_check
if time_since_check < 60: # Recent check available
# Prefer providers with higher success rates and lower latency
candidates.append((provider, provider.success_count * 10))
if not candidates:
return None
# Weighted random selection based on health scores
total_weight = sum(score for _, score in candidates)
import random
r = random.uniform(0, total_weight)
cumulative = 0
for provider, score in candidates:
cumulative += score
if cumulative >= r:
return provider
return candidates[0][0]Step 4: Building the Fault-Tolerant API Client
Now let's create a unified client that automatically handles failover:
class FaultTolerantAIClient:
def __init__(self):
self.providers = providers
self.breakers = breakers
self.health_check_interval = 10 # seconds
self._running = False
async def chat_completions(self, messages: List[Dict],
model: str = "gpt-4.1",
max_retries: int = 3) -> Dict:
"""
Send a chat completion request with automatic failover.
"""
last_error = None
for attempt in range(max_retries):
provider = await get_healthy_provider()
if provider is None:
last_error = Exception("All providers are unavailable")
await asyncio.sleep(2 ** attempt) # Exponential backoff
continue
breaker = self.breakers.get(provider.name)
try:
start_time = time.time()
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{provider.base_url}/chat/completions",
headers={"Authorization": f"Bearer {provider.api_key}"},
json={
"model": model,
"messages": messages,
"max_tokens": 1000
}
)
latency = (time.time() - start_time) * 1000 # ms
if response.status_code == 200:
breaker.record_success()
result = response.json()
result["_provider"] = provider.name
result["_latency_ms"] = round(latency, 2)
return result
else:
breaker.record_failure()
last_error = Exception(f"HTTP {response.status_code}")
except Exception as e:
breaker.record_failure()
last_error = e
# Wait before retry with exponential backoff
await asyncio.sleep(min(2 ** attempt, 10))
raise Exception(f"All providers failed. Last error: {last_error}")
async def health_monitor(self):
"""Continuously monitor all providers' health."""
self._running = True
while self._running:
tasks = []
for name, provider in self.providers.items():
task = provider.health_check()
tasks.append(task)
await asyncio.gather(*tasks, return_exceptions=True)
await asyncio.sleep(self.health_check_interval)
def stop_monitoring(self):
self._running = False
Usage example
async def main():
client = FaultTolerantAIClient()
# Start health monitoring in background
monitor_task = asyncio.create_task(client.health_monitor())
try:
# Send requests - automatically routes to healthy providers
response = await client.chat_completions([
{"role": "user", "content": "Explain circuit breakers in simple terms"}
])
print(f"Response from: {response['_provider']}")
print(f"Latency: {response['_latency_ms']}ms")
print(f"Content: {response['choices'][0]['message']['content']}")
finally:
client.stop_monitoring()
await monitor_task
if __name__ == "__main__":
asyncio.run(main())Advanced Features: Cost Optimization and Latency Routing
HolySheep AI offers remarkable value with pricing starting at $0.42 per million tokens for DeepSeek V3.2, compared to $8.00 for GPT-4.1 on other platforms. Your fault-tolerant system can automatically route requests based on both reliability and cost efficiency:
class SmartRouter:
"""Routes requests based on cost, latency, and availability."""
def __init__(self, client: FaultTolerantAIClient):
self.client = client
self.cost_weights = {
"deepseek-v3.2": 0.42, # $ per million tokens
"gemini-2.5-flash": 2.50,
"claude-sonnet-4.5": 15.00,
"gpt-4.1": 8.00,
}
async def route_request(self, messages: List[Dict],
max_cost_per_1k: float = 1.00) -> Dict:
"""Route request to cheapest suitable provider."""
# Get list of healthy providers
available = []
for name, breaker in self.client.breakers.items():
if breaker.should_allow_request():
provider = self.client.providers[name]
# Check if recent health check passed
if time.time() - provider.last_check < 30:
available.append(name)
if not available:
raise Exception("No healthy providers available")
# Find cheapest option that meets cost criteria
candidates = []
for name in available:
provider = self.client.providers[name]
# Use success rate as quality indicator
quality_score = min(provider.success_count / 10, 1.0)
cost = self.cost_weights.get(name, 10.0)
# Score = quality / cost (higher is better)
efficiency = quality_score / (cost + 0.01)
candidates.append((name, efficiency, cost))
# Sort by efficiency and pick the best
candidates.sort(key=lambda x: x[1], reverse=True)
# Try candidates in order of efficiency
for name, _, cost in candidates:
if cost <= max_cost_per_1k:
provider = self.client.providers[name]
try:
result = await self._call_provider(provider, messages)
result["_routing"] = {
"provider": name,
"cost_per_mtok": cost,
"efficiency_score": candidates[0][1] / (cost + 0.01)
}
return result
except:
continue
raise Exception("No providers meet cost and availability criteria")
async def _call_provider(self, provider: Provider,
messages: List[Dict]) -> Dict:
"""Make API call to specific provider."""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{provider.base_url}/chat/completions",
headers={"Authorization": f"Bearer {provider.api_key}"},
json={"model": "gpt-4.1", "messages": messages, "max_tokens": 1000}
)
if response.status_code != 200:
raise Exception(f"Provider error: {response.status_code}")
return response.json()Testing Your Fault-Tolerant System
Before deploying to production, verify your implementation handles various failure scenarios. I spent two weeks testing various edge cases, and here's what I learned:
import unittest
from unittest.mock import AsyncMock, patch
class TestCircuitBreaker(unittest.IsolatedAsyncioTestCase):
async def test_breaker_opens_after_threshold(self):
"""Circuit should open after configured failures."""
provider = Provider(
name="test",
base_url="https://test.com/v1",
api_key="test-key",
failure_threshold=3
)
breaker = CircuitBreaker(provider)
# Simulate failures
for _ in range(3):
breaker.record_failure()
self.assertEqual(breaker.state, "open")
self.assertFalse(breaker.should_allow_request())
async def test_breaker_recovers_after_timeout(self):
"""Circuit should attempt recovery after timeout."""
provider = Provider(
name="test",
base_url="https://test.com/v1",
api_key="test-key",
failure_threshold=1,
recovery_timeout=1 # 1 second for testing
)
breaker = CircuitBreaker(provider)
# Trigger failure
breaker.record_failure()
self.assertEqual(breaker.state, "open")
# Wait for recovery timeout
await asyncio.sleep(1.5)
# Should transition to half-open
self.assertTrue(breaker.should_allow_request())
self.assertEqual(breaker.state, "half-open")
async def test_breaker_closes_on_success(self):
"""Circuit should close after successful request in half-open."""
provider = Provider(
name="test",
base_url="https://test.com/v1",
api_key="test-key",
failure_threshold=1,
recovery_timeout=1
)
breaker = CircuitBreaker(provider)
# Open the breaker
breaker.record_failure()
self.assertEqual(breaker.state, "open")
# Wait for recovery
await asyncio.sleep(1.5)
breaker.should_allow_request()
self.assertEqual(breaker.state, "half-open")
# Record success
breaker.record_success()
self.assertEqual(breaker.state, "closed")
self.assertEqual(provider.failure_count, 0)
class TestHealthChecks(unittest.IsolatedAsyncioTestCase):
async def test_health_check_updates_status(self):
"""Health check should update provider statistics."""
provider = Provider(
name="test",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
)
# Mock successful response
with patch('httpx.AsyncClient') as mock_client:
mock_response = AsyncMock()
mock_response.status_code = 200
mock_client.return_value.__aenter__.return_value.post = AsyncMock(return_value=mock_response)
result = await provider.health_check()
self.assertTrue(result)
self.assertEqual(provider.success_count, 1)
self.assertEqual(provider.failure_count, 0)
self.assertEqual(provider.status, ProviderStatus.HEALTHY)
async def test_health_check_detects_failure(self):
"""Health check should record failures."""
provider = Provider(
name="test",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
)
# Mock failed response
with patch('httpx.AsyncClient') as mock_client:
mock_response = AsyncMock()
mock_response.status_code = 429 # Rate limited
mock_client.return_value.__aenter__.return_value.post = AsyncMock(return_value=mock_response)
result = await provider.health_check()
self.assertFalse(result)
self.assertEqual(provider.failure_count, 1)
if __name__ == "__main__":
unittest.main()Common Errors and Fixes
Error 1: "Connection timeout exceeded"
Problem: Health checks timeout waiting for response, marking all providers unhealthy despite them working fine.
Solution: Adjust timeout values and implement connection pooling:
# Increase timeout for health checks and configure connection pool
async def health_check_improved(self, timeout: float = 5.0) -> bool:
# Use a connection pool to avoid socket exhaustion
async with httpx.AsyncClient(
timeout=httpx.Timeout(timeout),
limits=httpx.Limits(max_keepalive_connections=5, max_connections=20)
) as client:
try:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "ping"}],
"max_tokens": 1
}
)
return response.status_code == 200
except asyncio.TimeoutError:
# Timeout doesn't necessarily mean provider is down
# Try a simpler endpoint if available
return await self.alternative_health_check(client)
except Exception:
return FalseError 2: "All providers are unavailable" even when one is working
Problem: Circuit breaker logic has a race condition where multiple threads mark providers unhealthy simultaneously.
Solution: Implement a lock to prevent concurrent state modifications:
import asyncio
from threading import Lock
class ThreadSafeCircuitBreaker:
def __init__(self, provider: Provider):
self.provider = provider
self.state = "closed"
self.last_failure_time: Optional[float] = None
self._lock = asyncio.Lock()
async def record_failure(self):
async with self._lock:
self.provider.failure_count += 1
self.last_failure_time = time.time()
if self.state == "half-open":
self.state = "open"
elif self.provider.failure_count >= self.provider.failure_threshold:
self.state = "open"
async def record_success(self):
async with self._lock:
if self.state == "half-open":
self.state = "closed"
self.provider.failure_count = 0
elif self.state == "closed":
self.provider.success_count += 1
async def should_allow_request(self) -> bool:
async with self._lock:
if self.state == "closed":
return True
if self.state == "open":
if time.time() - self.last_failure_time >= self.provider.recovery_timeout:
self.state = "half-open"
return True
return False
return self.state == "half-open"Error 3: "Rate limit exceeded" causing cascade failures
Problem: When one provider rate limits, requests flood other