Gemini API AI Gateway Design Patterns — Building a Unified Proxy for Rate Limiting, Failover, and Cost Tracking

Why You Need an AI Gateway

Calling the Gemini API directly from a single application works fine for prototypes. In production, however, multiple services hit the API concurrently, rate limits get tripped, and costs spiral unpredictably.

An AI gateway — essentially an API proxy — sits between your applications and the Gemini API. By funneling every request through a single layer, it solves four problems at once:

Centralized rate limiting: Aggregate requests from multiple services and prevent 429 errors before they happen
Automatic failover: Switch from Gemini 2.5 Pro to Flash transparently when a model goes down
Token cost tracking: Log token usage and cost for every request in real time
Model routing: Automatically select Pro or Flash based on request complexity

This guide walks through the architecture and full implementation in Python.

Gateway Architecture Overview

Component Layout

The gateway is composed of four core components:

┌─────────────┐     ┌──────────────────────────────────────┐     ┌─────────────┐
│  Service A  │────▶│          AI Gateway (Proxy)           │────▶│ Gemini 2.5  │
│  Service B  │────▶│  ┌────────┐ ┌──────┐ ┌───────────┐  │────▶│    Pro       │
│  Service C  │────▶│  │Rate    │ │Router│ │Cost       │  │────▶│ Gemini 2.5  │
│  Service D  │────▶│  │Limiter │ │      │ │Tracker    │  │     │    Flash    │
│             │     │  └────────┘ └──────┘ └───────────┘  │     └─────────────┘
└─────────────┘     └──────────────────────────────────────┘

Design Principles

Three principles guide the architecture:

Transparency: Clients call the gateway using the same interface as the native Gemini API. They shouldn't need to know the gateway exists.

Resilience: If one model or region fails, the gateway automatically falls back to an alternative path without surfacing errors to clients.

Observability: Every request's latency, token count, cost, and error status is measured and recorded in real time.

Rate Limiting with Token Buckets

Gemini API enforces limits on two axes: RPM (Requests Per Minute) and TPM (Tokens Per Minute). The gateway preemptively enforces these limits so that 429 errors never reach your services.

import time
import asyncio
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class TokenBucket:
    """Token bucket rate limiter"""
    capacity: float          # Maximum bucket size
    refill_rate: float       # Tokens added per second
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
 
    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()
 
    async def acquire(self, count: float = 1.0, timeout: float = 30.0) -> bool:
        """Acquire tokens, waiting until available or timeout"""
        deadline = time.monotonic() + timeout
        async with self._lock:
            while True:
                self._refill()
                if self.tokens >= count:
                    self.tokens -= count
                    return True
                wait_time = (count - self.tokens) / self.refill_rate
                if time.monotonic() + wait_time > deadline:
                    return False  # Timed out
                await asyncio.sleep(min(wait_time, 0.1))
 
    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
 
class RateLimiter:
    """Dual RPM + TPM rate limiter"""
 
    def __init__(self, rpm: int = 360, tpm: int = 4_000_000):
        self.request_bucket = TokenBucket(
            capacity=rpm,
            refill_rate=rpm / 60.0
        )
        self.token_bucket = TokenBucket(
            capacity=tpm,
            refill_rate=tpm / 60.0
        )
 
    async def acquire_request(self, estimated_tokens: int = 1000) -> bool:
        """Check rate limits before sending a request"""
        req_ok = await self.request_bucket.acquire(1.0)
        tok_ok = await self.token_bucket.acquire(float(estimated_tokens))
        return req_ok and tok_ok
 
    async def report_actual_tokens(self, actual_tokens: int, estimated_tokens: int):
        """Correct the token bucket with actual usage"""
        diff = estimated_tokens - actual_tokens
        if diff > 0:
            self.token_bucket.tokens = min(
                self.token_bucket.capacity,
                self.token_bucket.tokens + diff
            )

The key pattern here is reserve with estimates, then correct with actuals. This ensures you hold capacity before sending the request, but return unused tokens to the bucket to maximize throughput.

Automatic Failover with Circuit Breakers

When a model starts failing repeatedly, the circuit breaker automatically reroutes traffic to an alternative model.

import enum
from collections import deque
 
class CircuitState(enum.Enum):
    CLOSED = "closed"        # Normal operation
    OPEN = "open"            # Blocking requests
    HALF_OPEN = "half_open"  # Testing recovery
 
@dataclass
class CircuitBreaker:
    """Per-model circuit breaker"""
    failure_threshold: int = 5       # Failures before opening
    recovery_timeout: float = 60.0   # Seconds before trying again
    success_threshold: int = 2       # Successes to close from half-open
 
    state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    failures: int = field(default=0, init=False)
    successes_in_half_open: int = field(default=0, init=False)
    last_failure_time: float = field(default=0.0, init=False)
 
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.successes_in_half_open = 0
                return True
            return False
        return True  # HALF_OPEN allows one request through
 
    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.successes_in_half_open += 1
            if self.successes_in_half_open >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.failures = 0
        else:
            self.failures = 0
 
    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.monotonic()
        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN

The circuit breaker has three states. CLOSED (normal operation) transitions to OPEN (blocking) when errors exceed the threshold. After a cooldown period, it enters HALF_OPEN to test recovery with a single request.

Model Routing by Complexity

Not every request needs Gemini 2.5 Pro. The router analyzes incoming prompts and automatically selects the most cost-effective model.

@dataclass
class ModelConfig:
    name: str
    model_id: str
    cost_per_1k_input: float   # USD per 1K input tokens
    cost_per_1k_output: float  # USD per 1K output tokens
    priority: int              # Lower is higher priority
 
# Model definitions (March 2026 pricing)
MODELS = {
    "flash": ModelConfig(
        name="Gemini 2.5 Flash",
        model_id="gemini-2.5-flash-preview-05-20",
        cost_per_1k_input=0.00015,
        cost_per_1k_output=0.0006,
        priority=1,
    ),
    "pro": ModelConfig(
        name="Gemini 2.5 Pro",
        model_id="gemini-2.5-pro-preview-05-06",
        cost_per_1k_input=0.00125,
        cost_per_1k_output=0.005,
        priority=2,
    ),
}
 
class ModelRouter:
    """Complexity-based model selection"""
 
    COMPLEX_INDICATORS = [
        "analyze", "compare", "evaluate", "design",
        "architecture", "optimize", "debug", "refactor",
    ]
 
    def select_model(
        self,
        prompt: str,
        force_model: Optional[str] = None,
        max_output_tokens: int = 1024,
    ) -> ModelConfig:
        if force_model and force_model in MODELS:
            return MODELS[force_model]
 
        complexity_score = self._estimate_complexity(prompt, max_output_tokens)
        if complexity_score >= 0.6:
            return MODELS["pro"]
        return MODELS["flash"]
 
    def _estimate_complexity(self, prompt: str, max_output_tokens: int) -> float:
        score = 0.0
        prompt_lower = prompt.lower()
 
        # Keyword matching
        keyword_hits = sum(
            1 for kw in self.COMPLEX_INDICATORS if kw in prompt_lower
        )
        score += min(keyword_hits * 0.15, 0.45)
 
        # Input length heuristic
        if len(prompt) > 3000:
            score += 0.2
        elif len(prompt) > 1000:
            score += 0.1
 
        # High output token requests tend to be complex
        if max_output_tokens > 4096:
            score += 0.2
 
        return min(score, 1.0)

In production, you can go beyond keyword matching by learning from request history — tracking which prompts actually needed Pro-level reasoning and feeding that back into the router.

Real-Time Cost Tracking

The gateway records cost data for every request, enabling real-time dashboards and budget alerts.

import json
import datetime
from collections import defaultdict
 
@dataclass
class UsageRecord:
    timestamp: str
    client_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    latency_ms: float
    status: str  # "success" | "error" | "fallback"
 
class CostTracker:
    """Real-time token cost tracking"""
 
    def __init__(self):
        self.records: list[UsageRecord] = []
        self.daily_totals: dict[str, float] = defaultdict(float)
        self.client_totals: dict[str, float] = defaultdict(float)
 
    def record(
        self,
        client_id: str,
        model_config: ModelConfig,
        input_tokens: int,
        output_tokens: int,
        latency_ms: float,
        status: str = "success",
    ) -> UsageRecord:
        cost = (
            (input_tokens / 1000) * model_config.cost_per_1k_input
            + (output_tokens / 1000) * model_config.cost_per_1k_output
        )
        now = datetime.datetime.now(datetime.timezone.utc)
        record = UsageRecord(
            timestamp=now.isoformat(),
            client_id=client_id,
            model=model_config.name,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost_usd=round(cost, 6),
            latency_ms=round(latency_ms, 2),
            status=status,
        )
        self.records.append(record)
        self.daily_totals[now.strftime("%Y-%m-%d")] += cost
        self.client_totals[client_id] += cost
        return record
 
    def get_daily_summary(self, date: Optional[str] = None) -> dict:
        target = date or datetime.datetime.now().strftime("%Y-%m-%d")
        day_records = [r for r in self.records if r.timestamp.startswith(target)]
        return {
            "date": target,
            "total_requests": len(day_records),
            "total_cost_usd": round(sum(r.cost_usd for r in day_records), 4),
            "total_input_tokens": sum(r.input_tokens for r in day_records),
            "total_output_tokens": sum(r.output_tokens for r in day_records),
            "avg_latency_ms": round(
                sum(r.latency_ms for r in day_records) / max(len(day_records), 1), 2
            ),
            "error_rate": round(
                sum(1 for r in day_records if r.status == "error")
                / max(len(day_records), 1),
                4,
            ),
        }

The CostTracker automatically calculates per-request costs based on each model's pricing and provides instant daily and per-client aggregations. Pair it with Prometheus or Grafana for a live dashboard.

Putting It All Together

Here's the unified AIGateway class that integrates all four components:

import google.generativeai as genai
 
class AIGateway:
    """Gemini API AI Gateway (unified proxy)"""
 
    def __init__(self, api_key: str, rpm: int = 360, tpm: int = 4_000_000):
        genai.configure(api_key=api_key)
        self.router = ModelRouter()
        self.cost_tracker = CostTracker()
        self.rate_limiters: dict[str, RateLimiter] = {
            "flash": RateLimiter(rpm=rpm, tpm=tpm),
            "pro": RateLimiter(rpm=rpm // 2, tpm=tpm // 2),
        }
        self.circuit_breakers: dict[str, CircuitBreaker] = {
            "flash": CircuitBreaker(),
            "pro": CircuitBreaker(),
        }
 
    async def generate(
        self,
        prompt: str,
        client_id: str = "default",
        force_model: Optional[str] = None,
        max_output_tokens: int = 1024,
        temperature: float = 0.7,
    ) -> dict:
        """Call Gemini API through the gateway"""
        # 1. Model routing
        model_config = self.router.select_model(prompt, force_model, max_output_tokens)
        model_key = "pro" if "Pro" in model_config.name else "flash"
 
        # 2. Circuit breaker check with failover
        if not self.circuit_breakers[model_key].can_execute():
            fallback_key = "flash" if model_key == "pro" else "pro"
            if self.circuit_breakers[fallback_key].can_execute():
                model_config = MODELS[fallback_key]
                model_key = fallback_key
            else:
                raise Exception("All models are in circuit-open state")
 
        # 3. Rate limit check
        estimated_tokens = len(prompt) // 4 + max_output_tokens
        if not await self.rate_limiters[model_key].acquire_request(estimated_tokens):
            raise Exception(f"Rate limit exceeded for {model_config.name}")
 
        # 4. API call
        start_time = time.monotonic()
        try:
            model = genai.GenerativeModel(model_config.model_id)
            response = model.generate_content(
                prompt,
                generation_config=genai.types.GenerationConfig(
                    max_output_tokens=max_output_tokens,
                    temperature=temperature,
                ),
            )
            elapsed_ms = (time.monotonic() - start_time) * 1000
 
            # 5. Record success
            input_tokens = response.usage_metadata.prompt_token_count
            output_tokens = response.usage_metadata.candidates_token_count
 
            self.circuit_breakers[model_key].record_success()
            await self.rate_limiters[model_key].report_actual_tokens(
                input_tokens + output_tokens, estimated_tokens
            )
 
            record = self.cost_tracker.record(
                client_id=client_id,
                model_config=model_config,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                latency_ms=elapsed_ms,
            )
 
            return {
                "text": response.text,
                "model": model_config.name,
                "usage": {
                    "input_tokens": input_tokens,
                    "output_tokens": output_tokens,
                    "cost_usd": record.cost_usd,
                },
                "latency_ms": elapsed_ms,
            }
 
        except Exception as e:
            elapsed_ms = (time.monotonic() - start_time) * 1000
            self.circuit_breakers[model_key].record_failure()
 
            # Attempt failover
            fallback_key = "flash" if model_key == "pro" else "pro"
            if self.circuit_breakers[fallback_key].can_execute():
                return await self._fallback_generate(
                    prompt, fallback_key, client_id, max_output_tokens, temperature
                )
            raise

Here's how to use it:

import asyncio
 
# Expected output:
# Model: Gemini 2.5 Flash (simple requests route to Flash)
# Cost: $0.000XXX
# Daily Summary: {'date': '2026-03-26', 'total_requests': 1, ...}
 
async def main():
    gateway = AIGateway(api_key="YOUR_GEMINI_API_KEY")
 
    # Simple request → routed to Flash
    result = await gateway.generate(
        prompt="Explain Python list comprehensions briefly",
        client_id="service-a",
    )
    print(f"Model: {result['model']}")
    print(f"Cost: ${result['usage']['cost_usd']}")
 
    # Complex request → routed to Pro
    result = await gateway.generate(
        prompt="Analyze this microservices architecture, identify bottlenecks, and design an optimization plan...",
        client_id="service-b",
    )
    print(f"Model: {result['model']}")
 
    # Daily summary
    print(gateway.cost_tracker.get_daily_summary())
 
asyncio.run(main())

Production Deployment Best Practices

Environment Variables and Secret Management

Never hardcode API keys or endpoints. Use environment variables and a secret manager for production deployments.

import os
 
GATEWAY_CONFIG = {
    "api_key": os.environ["GEMINI_API_KEY"],
    "rpm": int(os.environ.get("GATEWAY_RPM", "360")),
    "tpm": int(os.environ.get("GATEWAY_TPM", "4000000")),
    "log_level": os.environ.get("GATEWAY_LOG_LEVEL", "INFO"),
}

Health Check Endpoints

Expose a health check endpoint that reports circuit breaker states, rate limiter headroom, and recent error rates. This is essential for monitoring and alerting in production.

Logging and Alerts

Output structured logs (JSON) for every request and set up alerts for when error rates or daily costs exceed thresholds. Integrating the CostTracker with Slack or PagerDuty notifications when budgets are approaching their limits is a common pattern.

Looking back

An AI gateway is an essential infrastructure layer for any serious Gemini API deployment. The four components covered in this guide — rate limiting (token buckets), automatic failover (circuit breakers), model routing, and cost tracking — work together to deliver both reliability and cost efficiency.

Start with a small project, introduce the gateway, and scale it as traffic grows. As cost tracking data accumulates, you'll be able to optimize routing decisions and refine prompts in a continuous improvement cycle.