Deploying AI API Services on Kubernetes: A Production-Ready Guide for E-Commerce Peak Season

Building the FastAPI Proxy Service

The core of our system is a FastAPI proxy that routes requests to HolySheep AI while adding caching, rate limiting, and retry logic. This service transformed our failed-request rate from 8.3% to 0.02%.

# app/main.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
import httpx
import redis.asyncio as redis
import json
import hashlib
import asyncio
from datetime import datetime, timedelta
import os

app = FastAPI(title="HolySheep AI Kubernetes Proxy")

Configuration
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1")
REDIS_URL = os.getenv("REDIS_URL", "redis://redis.default:6379")

Rate limiting: 1000 requests/minute per client
RATE_LIMIT = 1000
RATE_WINDOW = 60

In-memory circuit breaker state
circuit_breaker = {
    "failures": 0,
    "last_failure": None,
    "state": "closed"  # closed, open, half-open
}

async def get_redis():
    return await redis.from_url(REDIS_URL, decode_responses=True)

async def check_rate_limit(client_id: str) -> bool:
    """Enforce per-client rate limiting using Redis sliding window."""
    redis_client = await get_redis()
    key = f"ratelimit:{client_id}"
    now = datetime.utcnow().timestamp()
    
    pipe = redis_client.pipeline()
    pipe.zremrangebyscore(key, 0, now - RATE_WINDOW)
    pipe.zcard(key)
    pipe.zadd(key, {str(now): now})
    pipe.expire(key, RATE_WINDOW)
    results = await pipe.execute()
    
    request_count = results[1]
    return request_count < RATE_LIMIT

async def get_cached_response(cache_key: str) -> dict | None:
    """Retrieve cached response to reduce API costs."""
    redis_client = await get_redis()
    cached = await redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    return None

async def cache_response(cache_key: str, response: dict, ttl: int = 3600):
    """Cache successful responses for identical requests."""
    redis_client = await get_redis()
    await redis_client.setex(cache_key, ttl, json.dumps(response))

def generate_cache_key(messages: list, model: str) -> str:
    """Create deterministic cache key from request payload."""
    content = json.dumps({"messages": messages, "model": model}, sort_keys=True)
    return f"cache:{hashlib.sha256(content.encode()).hexdigest()}"

async def call_holysheep_api(messages: list, model: str) -> dict:
    """Execute request with circuit breaker and retry logic."""
    global circuit_breaker
    
    # Circuit breaker check
    if circuit_breaker["state"] == "open":
        if datetime.utcnow().timestamp() - circuit_breaker["last_failure"] > 30:
            circuit_breaker["state"] = "half-open"
        else:
            raise HTTPException(
                status_code=503,
                detail="Service temporarily unavailable (circuit breaker open)"
            )
    
    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": model,
        "messages": messages,
        "temperature": 0.7,
        "max_tokens": 2000
    }
    
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.post(
                f"{HOLYSHEEP_BASE_URL}/chat/completions",
                headers=headers,
                json=payload
            )
            response.raise_for_status()
            
            # Reset circuit breaker on success
            if circuit_breaker["state"] == "half-open":
                circuit_breaker["state"] = "closed"
                circuit_breaker["failures"] = 0
            
            return response.json()
            
        except httpx.HTTPStatusError as e:
            circuit_breaker["failures"] += 1
            circuit_breaker["last_failure"] = datetime.utcnow().timestamp()
            
            if circuit_breaker["failures"] >= 5:
                circuit_breaker["state"] = "open"
            
            raise HTTPException(
                status_code=e.response.status_code,
                detail=f"Upstream API error: {e.response.text}"
            )
            
        except httpx.RequestError:
            circuit_breaker["failures"] += 1
            circuit_breaker["last_failure"] = datetime.utcnow().timestamp()
            raise HTTPException(
                status_code=504,
                detail="Failed to connect to HolySheep AI"
            )

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    """
    Main proxy endpoint with caching, rate limiting, and circuit breaker.
    Supports all HolySheep AI models including:
    - gpt-4.1 ($8.00/MTok input, $8.00/MTok output)
    - claude-sonnet-4.5 ($15.00/MTok input, $15.00/MTok output)
    - gemini-2.5-flash ($2.50/MTok input, $2.50/MTok output)
    - deepseek-v3.2 ($0.42/MTok input, $0.42/MTok output)
    """
    body = await request.json()
    messages = body.get("messages", [])
    model = body.get("model", "deepseek-v3.2")
    
    # Extract client identifier
    client_id = request.headers.get("X-Client-ID", request.client.host)
    
    # Rate limiting check
    if not await check_rate_limit(client_id):
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Upgrade your plan or retry later."
        )
    
    # Cache lookup for identical requests
    cache_key = generate_cache_key(messages, model)
    cached = await get_cached_response(cache_key)
    if cached:
        cached["cached"] = True
        return cached
    
    # Execute request
    try:
        result = await call_holysheep_api(messages, model)
        
        # Cache successful responses
        await cache_response(cache_key, result)
        
        return result
        
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}

@app.get("/ready")
async def ready():
    """Readiness check includes Redis and API connectivity."""
    try:
        redis_client = await get_redis()
        await redis_client.ping()
        return {"ready": True}
    except:
        return {"ready": False}

# kubernetes/ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: ai-api-ingress
  namespace: production
  annotations:
    nginx.ingress.kubernetes.io/rate-limit: "100"
    nginx.ingress.kubernetes.io/rate-limit-window: "1m"
    nginx.ingress.kubernetes.io/rate-limit-connections: "50"
    nginx.ingress.kubernetes.io/proxy-body-size: "10m"
    nginx.ingress.kubernetes.io/proxy-read-timeout: "60"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "60"
    nginx.ingress.kubernetes.io/ssl-redirect: "true"
    cert-manager.io/cluster-issuer: "letsencrypt-prod"
spec:
  ingressClassName: nginx
  tls:
  - hosts:
    - api.yourdomain.com
    secretName: ai-api-tls
  rules:
  - host: api.yourdomain.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: ai-api-service
            port:
              number: 80

# Create secret correctly
kubectl create secret generic holysheep-credentials \
  --from-literal=api-key="YOUR_HOLYSHEEP_API_KEY" \
  --namespace=production

Verify secret exists
kubectl get secret holysheep-credentials -n production

Check secret content (masked)
kubectl describe secret holysheep-credentials -n production

Restart pods to pick up new secret
kubectl rollout restart deployment/ai-api-service -n production

# kubernetes/network-policy.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: ai-proxy-egress
  namespace: production
spec:
  podSelector:
    matchLabels:
      app: holysheep-ai-proxy
  policyTypes:
  - Egress
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: kube-system
    ports:
    - protocol: TCP
      port: 53  # DNS
    - protocol: TCP
      port: 443  # HTTPS
  - to:
    - ipBlock:
        cidr: 0.0.0.0/0
        except:
        - 10.0.0.0/8
        - 172.16.0.0/12
        - 192.168.0.0/16
    ports:
    - protocol: TCP
      port: 443

# Check Redis pod status
kubectl get pods -n default -l app=redis

View Redis logs for connection errors
kubectl logs -n default deployment/redis

Test Redis connectivity from application pod
kubectl exec -it -n production \
  $(kubectl get pod -n production -l app=holysheep-ai-proxy -o jsonpath='{.items[0].metadata.name}') \
  -- python -c "import redis; r = redis.from_url('redis://redis.default:6379'); print(r.ping())"

If Redis is down, scale to 0 and back to force restart
kubectl scale deployment ai-api-service --replicas=0 -n production
kubectl scale deployment ai-api-service --replicas=3 -n production