Building resilient AI applications requires more than just connecting to a single API endpoint. As a developer who has spent three years integrating large language models into production systems, I understand the frustration of watching your application freeze because one provider goes down. In this hands-on tutorial, I will walk you through implementing a robust multi-model routing system with automatic failover using HolySheep AI that kept my applications running through multiple provider outages last quarter.

What is Multi-Model Routing and Why Do You Need It?

Multi-model routing is a strategy where your application intelligently distributes requests across multiple AI model providers based on availability, cost, latency, and capability requirements. When one provider fails or experiences degraded performance, automatic failover instantly redirects traffic to backup models without user-facing interruption.

Imagine your customer service chatbot suddenly returns errors during peak hours because your primary AI provider hit rate limits. With proper multi-model routing, the system automatically switches to a secondary provider within milliseconds—your customers never notice the hiccup.

Who This Tutorial Is For

Who it is for:

Who it is NOT for:

HolySheep AI: Your Unified Gateway to 15+ Models

HolySheep AI aggregates access to GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2, and dozens of other models through a single API endpoint. The platform routes requests intelligently while maintaining sub-50ms latency overhead—faster than calling most providers directly.

The pricing structure alone makes HolySheep compelling: their rate of ¥1=$1 means you pay approximately 86% less than Chinese domestic rates averaging ¥7.3 per dollar. For high-volume applications processing millions of tokens monthly, this translates to thousands of dollars in savings.

Provider/ModelInput $/MTokOutput $/MTokLatency (p95)Failover Support
GPT-4.1$8.00$24.00120msYes
Claude Sonnet 4.5$15.00$75.00150msYes
Gemini 2.5 Flash$2.50$10.0080msYes
DeepSeek V3.2$0.42$1.6895msYes
HolySheep RouterDynamicDynamic<50ms overheadAutomatic

Pricing and ROI Analysis

For a mid-sized application processing 100 million input tokens and 50 million output tokens monthly, here is the cost comparison:

StrategyMonthly Cost (Estimation)Uptime SLAComplexity
GPT-4.1 only$1,55099.5%Low
Mixed (optimal routing)$38099.95%Medium
HolySheep Smart Router$34099.99%Low

The HolySheep approach saves approximately 78% compared to single-provider GPT-4.1 while providing automatic failover and better overall uptime. New users receive free credits on registration, allowing you to test the routing system extensively before committing.

Prerequisites

Step 1: Setting Up Your HolySheep Environment

Before writing any routing logic, you need to configure your environment properly. I recommend using environment variables to store your API key—this prevents accidental exposure in version control.

# Install required dependencies
pip install requests httpx python-dotenv


Create a .env file in your project root


HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY



Load environment variables

from dotenv import load_dotenv
import os

load_dotenv()


Verify your API key is loaded

api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
    raise ValueError("HOLYSHEEP_API_KEY not found. Check your .env file.")

print(f"API key loaded: {api_key[:8]}...{api_key[-4:]}")

Replace YOUR_HOLYSHEEP_API_KEY with the key from your HolySheep dashboard. The platform supports WeChat and Alipay for payment, making it convenient for developers in China while maintaining USD-denominated pricing.

Step 2: Understanding the HolySheep Routing Endpoint

The core of multi-model routing with HolySheep uses a single base URL: https://api.holysheep.ai/v1. This unified endpoint handles provider selection, failover logic, and response normalization automatically.

import requests
import json

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}


Test the connection with a simple completion request

test_payload = {
    "model": "auto",  # "auto" enables intelligent model selection
    "messages": [
        {"role": "user", "content": "Say 'HolySheep routing is working!'"}
    ],
    "max_tokens": 50,
    "temperature": 0.7
}

response = requests.post(
    f"{BASE_URL}/chat/completions",
    headers=headers,
    json=test_payload,
    timeout=30
)

if response.status_code == 200:
    data = response.json()
    print(f"Model used: {data.get('model')}")
    print(f"Response: {data['choices'][0]['message']['content']}")
    print(f"Usage: {data.get('usage')}")
else:
    print(f"Error {response.status_code}: {response.text}")

When you run this code, HolySheep automatically selects the optimal model based on your request characteristics. The model: "auto" parameter enables their intelligent routing engine.

Step 3: Implementing Custom Multi-Model Router with Priority Fallback

While HolySheep's auto-routing handles most scenarios, you may need custom logic for specific use cases. Below is a production-ready router class I developed for a high-traffic chatbot serving 50,000 daily users.

import requests
import time
import logging
from typing import Optional, Dict, List
from dataclasses import dataclass
from enum import Enum

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ModelPriority(Enum):
    """Model priority levels for routing decisions"""
    PRIMARY = 1      # Most capable, highest cost
    SECONDARY = 2    # Good balance of capability/cost
    FALLBACK = 3     # Budget option for simple tasks
    EMERGENCY = 4    # Last resort, minimal capability

@dataclass
class ModelConfig:
    name: str
    priority: ModelPriority
    max_tokens_per_minute: int
    timeout_seconds: int
    expected_latency_ms: int

class HolySheepMultiModelRouter:
    """
    Production-ready multi-model router with automatic failover.
    Implements priority-based routing with health checking and metrics.
    """
    
    def __init__(self, api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        
        # Define model configurations with priority tiers
        self.models = {
            "primary": ModelConfig(
                name="gpt-4.1",
                priority=ModelPriority.PRIMARY,
                max_tokens_per_minute=50000,
                timeout_seconds=30,
                expected_latency_ms=120
            ),
            "secondary": ModelConfig(
                name="claude-sonnet-4.5",
                priority=ModelPriority.SECONDARY,
                max_tokens_per_minute=40000,
                timeout_seconds=35,
                expected_latency_ms=150
            ),
            "fallback": ModelConfig(
                name="gemini-2.5-flash",
                priority=ModelPriority.FALLBACK,
                max_tokens_per_minute=100000,
                timeout_seconds=20,
                expected_latency_ms=80
            ),
            "emergency": ModelConfig(
                name="deepseek-v3.2",
                priority=ModelPriority.EMERGENCY,
                max_tokens_per_minute=200000,
                timeout_seconds=15,
                expected_latency_ms=95
            )
        }
        
        # Track model health and usage
        self.health_status = {name: True for name in self.models}
        self.request_counts = {name: 0 for name in self.models}
        self.last_failure = {name: 0 for name in self.models}
        
    def _check_model_health(self, model_name: str) -> bool:
        """Check if a model is healthy based on recent failures"""
        if model_name not in self.health_status:
            return False
            
        # If model failed recently, check if cooldown has passed (60 seconds)
        cooldown_seconds = 60
        if time.time() - self.last_failure[model_name] < cooldown_seconds:
            return False
            
        return self.health_status[model_name]
    
    def _mark_failure(self, model_name: str):
        """Mark a model as failed and record timestamp"""
        self.health_status[model_name] = False
        self.last_failure[model_name] = time.time()
        logger.warning(f"Model {model_name} marked as unhealthy")
    
    def _mark_success(self, model_name: str):
        """Mark a model as healthy"""
        self.health_status[model_name] = True
        logger.info(f"Model {model_name} marked as healthy")
    
    def _get_available_model(self, prefer_priority: ModelPriority = None) -> Optional[str]:
        """Get the best available model based on priority and health"""
        priority_order = [ModelPriority.PRIMARY, ModelPriority.SECONDARY, 
                         ModelPriority.FALLBACK, ModelPriority.EMERGENCY]
        
        if prefer_priority:
            # Start from preferred priority
            start_idx = priority_order.index(prefer_priority) if prefer_priority in priority_order else 0
            priority_order = priority_order[start_idx:] + priority_order[:start_idx]
        
        for priority in priority_order:
            for name, config in self.models.items():
                if config.priority == priority and self._check_model_health(name):
                    return name
        
        return None  # All models unhealthy
    
    def chat_completion(
        self,
        messages: List[Dict],
        system_prompt: str = None,
        max_tokens: int = 1000,
        temperature: float = 0.7,
        require_high_quality: bool = False
    ) -> Dict:
        """
        Send a chat completion request with automatic failover.
        
        Args:
            messages: List of message dictionaries
            system_prompt: Optional system-level instructions
            max_tokens: Maximum response tokens
            temperature: Response creativity (0.0-1.0)
            require_high_quality: If True, prefer primary models even if slower
            
        Returns:
            Response dictionary with model info and content
        """
        # Build final messages list with system prompt if provided
        final_messages = messages.copy()
        if system_prompt:
            final_messages.insert(0, {"role": "system", "content": system_prompt})
        
        # Determine preferred model based on requirements
        prefer_priority = ModelPriority.PRIMARY if require_high_quality else ModelPriority.SECONDARY
        
        # Track attempts for logging
        attempts = []
        tried_models = []
        
        # Maximum 3 failover attempts
        max_attempts = 3
        
        for attempt in range(max_attempts):
            model_key = self._get_available_model(prefer_priority)
            
            if not model_key:
                raise RuntimeError("All AI models are currently unavailable. Please try again later.")
            
            if model_key in tried_models:
                continue  # Skip already-tried models
                
            model_config = self.models[model_key]
            model_name = model_config.name
            tried_models.append(model_key)
            
            payload = {
                "model": model_name,
                "messages": final_messages,
                "max_tokens": max_tokens,
                "temperature": temperature
            }
            
            try:
                start_time = time.time()
                
                response = requests.post(
                    f"{self.base_url}/chat/completions",
                    headers=self.headers,
                    json=payload,
                    timeout=model_config.timeout_seconds
                )
                
                latency_ms = (time.time() - start_time) * 1000
                
                if response.status_code == 200:
                    self._mark_success(model_key)
                    self.request_counts[model_key] += 1
                    
                    result = response.json()
                    result['_routing_metadata'] = {
                        'model_used': model_name,
                        'latency_ms': round(latency_ms, 2),
                        'attempt_number': attempt + 1,
                        'tried_models': tried_models
                    }
                    
                    logger.info(
                        f"Request succeeded via {model_name} "
                        f"(latency: {latency_ms:.0f}ms, attempt: {attempt + 1})"
                    )
                    
                    return result
                    
                elif response.status_code == 429:
                    # Rate limited - failover to next model
                    logger.warning(f"Rate limit hit on {model_name}, trying next model...")
                    self._mark_failure(model_key)
                    continue
                    
                elif response.status_code >= 500:
                    # Server error - failover
                    logger.warning(f"Server error {response.status_code} on {model_name}")
                    self._mark_failure(model_key)
                    continue
                    
                else:
                    # Client error - don't retry
                    raise Exception(f"API error {response.status_code}: {response.text}")
                    
            except requests.exceptions.Timeout:
                logger.warning(f"Timeout on {model_name}, trying next model...")
                self._mark_failure(model_key)
                continue
                
            except requests.exceptions.ConnectionError:
                logger.warning(f"Connection error on {model_name}, trying next model...")
                self._mark_failure(model_key)
                continue
        
        # All attempts exhausted
        raise RuntimeError(
            f"All {len(tried_models)} attempted models failed. "
            f"Tried: {', '.join(tried_models)}"
        )
    
    def get_metrics(self) -> Dict:
        """Return routing metrics for monitoring"""
        return {
            'request_counts': self.request_counts.copy(),
            'health_status': self.health_status.copy(),
            'total_requests': sum(self.request_counts.values())
        }



Initialize the router

router = HolySheepMultiModelRouter("YOUR_HOLYSHEEP_API_KEY")

Step 4: Testing Your Router with Simulated Failures

To verify your failover logic works correctly, you need to test with simulated failure scenarios. I recommend using a staging environment first—never test failover logic in production without proper monitoring.

def test_router_failover():
    """Comprehensive test suite for the multi-model router"""
    
    print("=" * 60)
    print("HOLYSHEEP MULTI-MODEL ROUTER TEST SUITE")
    print("=" * 60)
    
    # Initialize router with test API key
    test_router = HolySheepMultiModelRouter("YOUR_HOLYSHEEP_API_KEY")
    
    # Test 1: Basic successful request
    print("\n[Test 1] Basic Request - Should use secondary model by default")
    try:
        response = test_router.chat_completion(
            messages=[{"role": "user", "content": "What is 2+2?"}],
            max_tokens=50
        )
        print(f"✓ Success: {response['_routing_metadata']}")
    except Exception as e:
        print(f"✗ Failed: {e}")
    
    # Test 2: High-quality request (prefers primary)
    print("\n[Test 2] High-Quality Request - Should prefer GPT-4.1")
    try:
        response = test_router.chat_completion(
            messages=[
                {"role": "user", "content": "Explain quantum entanglement in detail"}
            ],
            system_prompt="You are a physics professor.",
            require_high_quality=True,
            max_tokens=500
        )
        metadata = response['_routing_metadata']
        print(f"✓ Success: Used {metadata['model_used']} (latency: {metadata['latency_ms']:.0f}ms)")
        print(f"  Response preview: {response['choices'][0]['message']['content'][:100]}...")
    except Exception as e:
        print(f"✗ Failed: {e}")
    
    # Test 3: Budget request (prefers fallback models)
    print("\n[Test 3] Budget Request - Should prefer DeepSeek V3.2")
    try:
        response = test_router.chat_completion(
            messages=[
                {"role": "user", "content": "What day is it today?"}
            ],
            max_tokens=20,
            temperature=0.1
        )
        metadata = response['_routing_metadata']
        print(f"✓ Success: Used {metadata['model_used']} (latency: {metadata['latency_ms']:.0f}ms)")
    except Exception as e:
        print(f"✗ Failed: {e}")
    
    # Test 4: Long conversation
    print("\n[Test 4] Multi-turn Conversation - Tests conversation handling")
    conversation = [
        {"role": "system", "content": "You are a helpful Python assistant."},
        {"role": "user", "content": "How do I read a file in Python?"},
        {"role": "assistant", "content": "You can use the built-in open() function..."},
        {"role": "user", "content": "Show me an example with error handling"}
    ]
    try:
        response = test_router.chat_completion(
            messages=conversation,
            max_tokens=300
        )
        metadata = response['_routing_metadata']
        print(f"✓ Success: Used {metadata['model_used']}")
        print(f"  Tokens used: {response.get('usage', {}).get('total_tokens', 'N/A')}")
    except Exception as e:
        print(f"✗ Failed: {e}")
    
    # Print final metrics
    print("\n" + "=" * 60)
    print("ROUTING METRICS")
    print("=" * 60)
    metrics = test_router.get_metrics()
    for key, value in metrics['request_counts'].items():
        status = "✓" if metrics['health_status'][key] else "✗"
        print(f"  {status} {key}: {value} requests")
    print(f"\nTotal requests processed: {metrics['total_requests']}")



Run the test suite

if __name__ == "__main__":
    test_router_failover()

Step 5: Implementing Health Monitoring and Alerts

In production, you need continuous health monitoring. Below is a monitoring decorator that logs performance metrics and triggers alerts when failover frequency increases—a signal that something is fundamentally wrong.

import functools
import time
from collections import defaultdict
from datetime import datetime, timedelta

class RouterMonitor:
    """Monitor router health and performance metrics"""
    
    def __init__(self, alert_threshold_failovers_per_minute: int = 5):
        self.alert_threshold = alert_threshold_failovers_per_minute
        self.failover_history = defaultdict(list)  # model -> list of timestamps
        self.response_times = defaultdict(list)
        self.error_counts = defaultdict(int)
        
    def record_failover(self, model_name: str):
        """Record a failover event for monitoring"""
        self.failover_history[model_name].append(time.time())
        self._cleanup_old_records()
        
    def record_response_time(self, model_name: str, latency_ms: float):
        """Record successful response time"""
        self.response_times[model_name].append({
            'timestamp': time.time(),
            'latency_ms': latency_ms
        })
        self._cleanup_old_records()
        
    def record_error(self, model_name: str, error_type: str):
        """Record an error for a specific model"""
        self.error_counts[f"{model_name}:{error_type}"] += 1
        
    def _cleanup_old_records(self, max_age_hours: int = 24):
        """Remove records older than max_age_hours"""
        cutoff = time.time() - (max_age_hours * 3600)
        
        for model in self.failover_history:
            self.failover_history[model] = [
                t for t in self.failover_history[model] if t > cutoff
            ]
            
        for model in self.response_times:
            self.response_times[model] = [
                r for r in self.response_times[model] if r['timestamp'] > cutoff
            ]
    
    def get_failover_rate(self, model_name: str, minutes: int = 5) -> float:
        """Get failover rate per minute for the specified time window"""
        cutoff = time.time() - (minutes * 60)
        recent_failovers = [t for t in self.failover_history[model_name] if t > cutoff]
        return len(recent_failovers) / minutes
    
    def should_alert(self) -> tuple[bool, str]:
        """Check if alert conditions are met"""
        for model in self.failover_history:
            rate = self.get_failover_rate(model)
            if rate >= self.alert_threshold:
                return True, (
                    f"ALERT: {model} failover rate is {rate:.1f}/min "
                    f"(threshold: {self.alert_threshold}/min)"
                )
        return False, ""
    
    def get_health_report(self) -> dict:
        """Generate comprehensive health report"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'models': {}
        }
        
        for model in set(list(self.failover_history.keys()) + 
                        list(self.response_times.keys())):
            recent_responses = [
                r for r in self.response_times.get(model, [])
                if r['timestamp'] > time.time() - 300  # Last 5 minutes
            ]
            
            latencies = [r['latency_ms'] for r in recent_responses]
            
            report['models'][model] = {
                'failover_count_24h': len(self.failover_history.get(model, [])),
                'failover_rate_per_min': self.get_failover_rate(model),
                'avg_latency_ms': sum(latencies) / len(latencies) if latencies else 0,
                'p95_latency_ms': sorted(latencies)[int(len(latencies) * 0.95)] if len(latencies) > 20 else 0,
                'request_count_5min': len(recent_responses)
            }
            
        return report


def monitored_router_call(monitor: RouterMonitor):
    """Decorator to monitor router calls"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(router, *args, **kwargs):
            try:
                result = func(router, *args, **kwargs)
                
                # Record success metrics
                if '_routing_metadata' in result:
                    metadata = result['_routing_metadata']
                    monitor.record_response_time(
                        metadata['model_used'],
                        metadata['latency_ms']
                    )
                    
                    # Check for failover (attempt > 1)
                    if metadata['attempt_number'] > 1:
                        monitor.record_failover(metadata['model_used'])
                        
                return result
                
            except Exception as e:
                # Determine which model failed from the error message
                error_msg = str(e)
                if "all" in error_msg.lower():
                    for model in router.models:
                        monitor.record_error(model, type(e).__name__)
                raise
                
        return wrapper
    return decorator



Example usage with monitoring

monitor = RouterMonitor(alert_threshold_failovers_per_minute=3)

@monitored_router_call(monitor)
def monitored_chat_completion(router, *args, **kwargs):
    return router.chat_completion(*args, **kwargs)


def run_monitored_session():
    """Run a monitored session with the router"""
    print("Starting monitored router session...")
    
    router = HolySheepMultiModelRouter("YOUR_HOLYSHEEP_API_KEY")
    
    # Process several requests
    for i in range(10):
        try:
            result = monitored_chat_completion(
                router,
                messages=[{"role": "user", "content": f"Test request {i+1}"}],
                max_tokens=100
            )
            print(f"Request {i+1}: ✓ {result['_routing_metadata']['model_used']}")
        except Exception as e:
            print(f"Request {i+1}: ✗ {e}")
    
    # Check for alerts
    should_alert, alert_msg = monitor.should_alert()
    if should_alert:
        print(f"\n🚨 {alert_msg}")
    
    # Print health report
    report = monitor.get_health_report()
    print("\n" + "=" * 50)
    print("HEALTH REPORT")
    print("=" * 50)
    for model, stats in report['models'].items():
        print(f"\n{model}:")
        for key, value in stats.items():
            if isinstance(value, float):
                print(f"  {key}: {value:.2f}")
            else:
                print(f"  {key}: {value}")

Why Choose HolySheep for Multi-Model Routing

After testing multiple routing solutions over the past year, HolySheep stands out for several reasons that directly impact your bottom line and operational sanity:

Common Errors and Fixes

Based on patterns I've encountered implementing production routing systems, here are the most frequent issues and their solutions:

Error 1: "401 Unauthorized" - Invalid API Key

This error occurs when the API key is missing, malformed, or expired. HolySheep keys can be regenerated from the dashboard if compromised.

# INCORRECT - Missing Bearer prefix
headers = {
    "Authorization": "YOUR_HOLYSHEEP_API_KEY",  # Missing "Bearer "
    "Content-Type": "application/json"
}


CORRECT - Proper Bearer token format

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}


VERIFICATION - Test your setup

import os
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key or len(api_key) < 20:
    raise ValueError(
        "Invalid API key. Make sure HOLYSHEEP_API_KEY is set in your .env file. "
        "Get your key from https://www.holysheep.ai/register"
    )

Error 2: "429 Too Many Requests" - Rate Limit Exceeded

Rate limits vary by model tier. When you hit limits, implement exponential backoff with jitter.

import random
import time

def request_with_retry(url, headers, payload, max_retries=3):
    """Request with exponential backoff for rate limit handling"""
    
    for attempt in range(max_retries):
        response = requests.post(url, headers=headers, json=payload)
        
        if response.status_code == 429:
            # Calculate backoff: 1s, 2s, 4s with jitter
            base_delay = 2 ** attempt
            jitter = random.uniform(0, 1)
            delay = base_delay + jitter
            
            print(f"Rate limited. Retrying in {delay:.1f} seconds...")
            time.sleep(delay)
            continue
            
        return response  # Success or non-retryable error
    
    raise Exception(f"Failed after {max_retries} retries due to rate limiting")

Error 3: "ConnectionError - Connection aborted" - Network Issues

Network timeouts are common in distributed systems. Configure appropriate timeouts and implement connection pooling.

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


Configure connection pooling and automatic retry

session = requests.Session()

retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["POST"]
)

adapter = HTTPAdapter(
    max_retries=retry_strategy,
    pool_connections=10,
    pool_maxsize=20
)

session.mount("https://", adapter)


Use session for requests with automatic retry and pooling

def safe_chat_completion(messages, timeout=30):
    response = session.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json={
            "model": "auto",
            "messages": messages,
            "max_tokens": 1000
        },
        timeout=timeout
    )
    return response.json()

Error 4: "Timeout - Request exceeded 30s" - Slow Model Response

Different models have different latency characteristics. Configure timeouts based on expected model performance.

# Model-specific timeout recommendations
TIMEOUT_CONFIG = {
    "gemini-2.5-flash": 15,    # Fastest, can use shorter timeout
    "deepseek-v3.2": 20,       # Good balance
    "claude-sonnet-4.5": 35,   # Longer context, needs more time
    "gpt-4.1": 30,             # Standard timeout
}

def get_appropriate_timeout(model_name: str) -> int:
    """Get timeout based on model characteristics"""
    return TIMEOUT_CONFIG.get(model_name, 30)


Usage in router

response = requests.post(
    url,
    headers=headers,
    json=payload,
    timeout=get_appropriate_timeout(model_name)  # Dynamic timeout
)

Error 5: "All models unavailable" - Complete Provider Outage

In rare cases of complete outages, implement a circuit breaker pattern to fail fast and log for debugging.

from datetime import datetime, timedelta

class CircuitBreaker:
    """Prevent cascading failures during extended outages"""
    
    def __init__(self, failure_threshold=5, recovery_timeout=300):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        
    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"
        
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"
            print(f"CIRCUIT BREAKER OPENED - All requests will fail fast for {self.recovery_timeout}s")
    
    def can_attempt(self) -> bool:
        if self.state == "CLOSED":
            return True
            
        if self.state == "OPEN":
            if self.last_failure_time:
                elapsed = (datetime.now() - self.last_failure_time).seconds
                if elapsed >= self.recovery_timeout:
                    self.state = "HALF_OPEN"
                    return True
            return False
            
        return True  # HALF_OPEN allows one test request

Production Deployment Checklist

Before deploying your routing system to production, verify each item: