The e-commerce AI customer service peak during last November's shopping festival nearly broke our system. We had deployed four different LLM providers across multiple regions, but our routing logic was a fragile mess of hardcoded fallbacks and manual health checks. When latency spiked to 3 seconds, customers abandoned conversations, and our support costs tripled. That night, I realized we needed a proper AI API service discovery mechanism—not just load balancing, but intelligent routing based on real-time model availability, response quality, and cost optimization.
Understanding AI API Service Discovery
Service discovery in traditional web infrastructure monitors backend health and routes traffic accordingly. But AI APIs introduce unique challenges: model availability varies by region, response quality differs across providers, cost-per-token changes constantly, and latency can fluctuate dramatically based on server load. A robust discovery mechanism must handle all these dimensions while maintaining sub-50ms routing overhead.
Sign up here for HolySheep AI, which exemplifies modern service discovery with automatic failover between GPT-4.1 ($8/MTok output), Claude Sonnet 4.5 ($15/MTok output), Gemini 2.5 Flash ($2.50/MTok output), and DeepSeek V3.2 ($0.42/MTok output) at rates starting at just $1 per ¥1—85% cheaper than the ¥7.3 standard rate.
Architecture Design
Our service discovery system consists of four core components working in concert:
- Registry Layer: Maintains real-time availability data for each model endpoint
- Health Monitor: Continuous latency and success rate tracking with 100ms polling intervals
- Routing Engine: Intelligent request distribution based on current conditions
- Circuit Breaker: Automatic isolation of degraded services with exponential backoff recovery
Implementation: Building the Discovery Client
Let me walk through the complete implementation I built for our production system, including the error handling patterns that saved us during the holiday peak.
Core Service Discovery Client
import asyncio
import httpx
import time
from dataclasses import dataclass
from typing import Optional, List, Dict
from collections import defaultdict
import statistics
@dataclass
class ModelEndpoint:
provider: str
model: str
base_url: str
api_key: str
region: str
current_latency: float = 0.0
success_rate: float = 1.0
failure_count: int = 0
is_healthy: bool = True
last_check: float = 0.0
class AIServiceDiscovery:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.endpoints: Dict[str, List[ModelEndpoint]] = defaultdict(list)
self.circuit_state: Dict[str, str] = defaultdict(lambda: "closed")
self.circuit_failure_count: Dict[str, int] = defaultdict(int)
self._initialize_default_endpoints()
def _initialize_default_endpoints(self):
"""Register available model endpoints with HolySheep AI"""
models = [
("gpt-4.1", "GPT-4.1", 8.0), # $8/MTok
("claude-sonnet-4.5", "Claude Sonnet 4.5", 15.0), # $15/MTok
("gemini-2.5-flash", "Gemini 2.5 Flash", 2.5), # $2.50/MTok
("deepseek-v3.2", "DeepSeek V3.2", 0.42), # $0.42/MTok
]
for model_id, model_name, cost in models:
endpoint = ModelEndpoint(
provider="holysheep",
model=model_id,
base_url=self.base_url,
api_key=self.api_key,
region="auto"
)
self.endpoints[model_name].append(endpoint)
async def health_check(self, endpoint: ModelEndpoint) -> bool:
"""Perform health check with latency measurement"""
try:
start = time.perf_counter()
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.post(
f"{endpoint.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {endpoint.api_key}",
"Content-Type": "application/json"
},
json={
"model": endpoint.model,
"messages": [{"role": "user", "content": "ping"}],
"max_tokens": 1
}
)
latency = (time.perf_counter() - start) * 1000
if response.status_code == 200:
endpoint.current_latency = latency
endpoint.success_rate = (endpoint.success_rate * 9 + 1) / 10
endpoint.failure_count = 0
endpoint.is_healthy = True
endpoint.last_check = time.time()
return True
except Exception:
endpoint.failure_count += 1
endpoint.success_rate = (endpoint.success_rate * 9) / 10
return False
async def run_health_checks(self):
"""Background task: continuous health monitoring"""
while True:
for model_list in self.endpoints.values():
for endpoint in model_list:
await self.health_check(endpoint)
await asyncio.sleep(0.1) # Check every 100ms
def get_best_endpoint(self, model_name: str, prefer_low_cost: bool = False) -> Optional[ModelEndpoint]:
"""Route to optimal endpoint based on latency and cost"""
candidates = self.endpoints.get(model_name, [])
if not candidates:
return None
healthy = [ep for ep in candidates if ep.is_healthy]
if not healthy:
return candidates[0] if candidates else None
if prefer_low_cost:
# Cost-optimized routing for high-volume requests
return min(healthy, key=lambda ep: self._get_model_cost(ep.model))
# Latency-optimized routing (default)
return min(healthy, key=lambda ep: ep.current_latency)
def _get_model_cost(self, model: str) -> float:
costs = {
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50,
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0
}
return costs.get(model, 1.0)
def should_circuit_break(self, endpoint_id: str) -> bool:
"""Check if circuit breaker should trip"""
if self.circuit_state[endpoint_id] == "open":
# Check if recovery timeout expired (30 seconds)
if self.circuit_failure_count.get(endpoint_id, 0) > 5:
self.circuit_state[endpoint_id] = "half-open"
return False
return True
return FalseComplete API Client with Service Discovery
import asyncio
from typing import Optional, Dict, Any
class AIServiceClient:
def __init__(self, api_key: str):
self.discovery = AIServiceDiscovery(api_key)
self.default_model = "GPT-4.1"
self.fallback_chain = ["Claude Sonnet 4.5", "Gemini 2.5 Flash", "DeepSeek V3.2"]
async def chat_completion(
self,
messages: list,
model: Optional[str] = None,