A practical guide to deploying the gateway in different environments, from local development to production.
Your Application
β
βΌ
βββββββββββββββββββββββββββββββββββββββββββββββββ
β GatewayClient β
β β
β ββββββββββββ ββββββββββββ ββββββββββββββ β
β β Cache β β Rate β β Request β β
β β Layer β β Limiter β β Logger β β
β ββββββββββββ ββββββββββββ ββββββββββββββ β
β β
β ββββββββββββββββββββββββββββββββββββββββββ β
β β Retry Router β β
β β ββββββββ ββββββββββββ ββββββββ β β
β β βOpenAIββ βAnthropic ββ βLocal β β β
β β ββββββββ ββββββββββββ ββββββββ β β
β ββββββββββββββββββββββββββββββββββββββββββ β
β β
β ββββββββββββββββββββββββββββββββββββββββββ β
β β Usage Analytics β β
β ββββββββββββββββββββββββββββββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββ
from src.client import GatewayClient
from src.providers.openai_provider import OpenAIProvider
gateway = GatewayClient(
providers=[OpenAIProvider(api_key="YOUR_API_KEY_HERE")],
enable_logging=False, # Less noise during development
)
answer = gateway.chat("What is the meaning of life?")
print(answer)If you're running a local model (Ollama, vLLM, llama.cpp), use the LocalProvider:
from src.providers.local_provider import LocalProvider
# Ollama
gateway = GatewayClient(
providers=[LocalProvider(base_url="http://localhost:11434/v1", default_model="llama3")],
enable_rate_limiting=False, # Local server, no API quota
)
# vLLM
gateway = GatewayClient(
providers=[LocalProvider(base_url="http://localhost:8000/v1")],
)from src.client import GatewayClient
from src.providers.openai_provider import OpenAIProvider
from src.providers.anthropic_provider import AnthropicProvider
from src.providers.local_provider import LocalProvider
gateway = GatewayClient(
providers=[
# Primary β fastest and cheapest for most tasks
OpenAIProvider(
api_key=os.environ["OPENAI_API_KEY"],
default_model="gpt-4o-mini",
),
# Fallback 1 β used when OpenAI is down or rate-limited
AnthropicProvider(
api_key=os.environ["ANTHROPIC_API_KEY"],
default_model="claude-sonnet-4-20250514",
),
# Fallback 2 β local model for total API outage
LocalProvider(
base_url="http://internal-llm.docs.example.com:8000/v1",
),
],
enable_cache=True,
cache_path="/var/lib/gateway/cache.json",
log_path="/var/log/gateway/requests.jsonl",
max_retries=3,
)Never hardcode API keys. Use environment variables:
export OPENAI_API_KEY="sk-YOUR_KEY_HERE"
export ANTHROPIC_API_KEY="sk-ant-YOUR_KEY_HERE"Set per-provider rate limits to stay within your API quota:
gateway = GatewayClient(providers=[])
# OpenAI Tier 2: ~500 RPM β ~8 RPS
gateway.add_provider(
OpenAIProvider(api_key=os.environ["OPENAI_API_KEY"]),
rate_limit_capacity=30, # Burst of 30
rate_limit_refill=8.0, # 8 req/sec sustained
)
# Anthropic: ~1000 RPM β ~16 RPS
gateway.add_provider(
AnthropicProvider(api_key=os.environ["ANTHROPIC_API_KEY"]),
rate_limit_capacity=50,
rate_limit_refill=16.0,
)The gateway writes structured JSONL logs. Analyse them with standard tools:
# Count requests by provider
jq -r '.provider' gateway_requests.jsonl | sort | uniq -c | sort -rn
# Find slow requests (>2s)
jq 'select(.latency_ms > 2000)' gateway_requests.jsonl
# Count errors
jq 'select(.error != "")' gateway_requests.jsonl | wc -l
# Cost estimate (rough)
jq '[.prompt_tokens + .completion_tokens] | add' gateway_requests.jsonlGenerate analytics reports programmatically:
from src.analytics import UsageAnalytics
analytics = UsageAnalytics()
analytics.load_from_file("gateway_requests.jsonl")
# Per-provider breakdown
for stat in analytics.by_provider():
print(f"{stat.provider}: {stat.total_requests} reqs, "
f"{stat.error_rate:.1%} errors, "
f"{stat.avg_latency_ms:.0f}ms avg")
# Full Markdown report
report = analytics.summary_markdown()
with open("analytics_report.md", "w") as f:
f.write(report)Use in a health check endpoint or monitoring script:
health = gateway.health()
# {'openai': True, 'anthropic': True, 'local': False}
provider_health = gateway.provider_health
# {'openai': {'total_requests': 1024, 'total_failures': 3, 'error_rate': 0.003, ...}}The retry router includes a circuit breaker that protects against cascading failures:
1. Closed (normal): Requests flow through normally.
2. Open (tripped): After N consecutive failures (default: 5), the circuit opens. No requests are sent to the provider for a cooldown period (default: 60s).
3. Half-open (testing): After the cooldown, the next request is allowed through. If it succeeds, the circuit closes. If it fails, the circuit re-opens.
Normal β 5 failures β Open (60s) β Half-open β Success β Normal
β Failure β Open (60s)
| Concern | Solution |
|---|---|
| Cache size | Set max_entries based on memory budget. 5K entries β 50-100MB. |
| Log file size | Rotate gateway_requests.jsonl with logrotate or a cron job. |
| Thread safety | The rate limiter is thread-safe. The cache and logger use append-only writes. |
| Multiple processes | Each process gets its own cache and rate limiter state. For shared state, use Redis (not included). |
rate_limit_capacity or rate_limit_refill for the provider.gateway.health() to check provider connectivity.gateway.cache_stats for hit rate.cache_temperature_zero_only=True and using temperature=0 for deterministic queries.Get the full AI API Gateway and unlock everything.
Get the complete guide with every chapter unlocked, including code samples, diagrams, and best practices.
Access all interactive tools with complete data, all workload profiles, and the full scenario library.
Downloadable source code, configuration files, and working examples from every chapter.
Free updates for life. Every new chapter, tool, and improvement included.