How to Setup AI API Gateway: Complete Guide 2026

{
    "cpu": "4+ cores",
    "memory": "8+ GB RAM",
    "storage": "50+ GB SSD",
    "network": "1 Gbps+ bandwidth",
    "operating_system": "Linux recommended"
}

#!/bin/bash

# Update system packages
sudo apt update && sudo apt upgrade -y

# Install Docker
sudo apt install docker.io -y
sudo systemctl start docker
sudo systemctl enable docker

# Install Docker Compose
sudo apt install docker-compose -y

# Install kubectl (Kubernetes CLI)
curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
sudo install -o root -g root -m 0755 kubectl /usr/local/bin/kubectl

# Install Python 3.10+
sudo apt install python3.10 python3-pip -y

# Install Node.js 18+
curl -fsSL https://deb.nodesource.com/setup_18.x | sudo -E bash -
sudo apt install nodejs -y

# Gateway type analysis
gateway_types = {
    "self_hosted": {
        "pros": ["Full control", "No recurring costs", "Data sovereignty"],
        "cons": ["Maintenance overhead", "Technical expertise required"],
        "best_for": ["Enterprise", "High-security applications", "Custom requirements"]
    },
    "cloud_managed": {
        "pros": ["No infrastructure management", "Automatic scaling", "Regular updates"],
        "cons": ["Recurring costs", "Vendor lock-in potential"],
        "best_for": ["Startups", "Rapid prototyping", "Limited technical team"]
    },
    "hybrid": {
        "pros": ["Flexibility", "Cost optimization", "Redundancy"],
        "cons": ["Complex setup", "Integration challenges"],
        "best_for": ["Growing businesses", "Regulatory compliance", "Legacy systems"]
    }
}

# Decision helper function
def recommend_gateway(requirements):
    score = {"self_hosted": 0, "cloud_managed": 0, "hybrid": 0}
    
    if requirements.get("enterprise"):
        score["self_hosted"] += 3
        score["hybrid"] += 2
    if requirements.get("startup"):
        score["cloud_managed"] += 3
    if requirements.get("security"):
        score["self_hosted"] += 2
        score["hybrid"] += 1
    if requirements.get("scalability"):
        score["cloud_managed"] += 2
    if requirements.get("budget_constrained"):
        score["self_hosted"] += 1
    if requirements.get("flexibility"):
        score["hybrid"] += 2
    
    return max(score, key=score.get)

version: '3.8'

services:
  ai-gateway:
    image: ai-api-gateway:latest
    container_name: ai-api-gateway
    restart: unless-stopped
    ports:
      - "8080:8080"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - GATEWAY_LOG_LEVEL=INFO
      - RATE_LIMIT_REQUESTS=1000
      - RATE_LIMIT_PERIOD=60
    volumes:
      - ./config:/app/config
      - ./logs:/app/logs
    networks:
      - ai-gateway-network
  
  redis-cache:
    image: redis:7-alpine
    container_name: gateway-redis
    restart: unless-stopped
    ports:
      - "6379:6379"
    command: redis-server --appendonly yes
    volumes:
      - redis-data:/data
    networks:
      - ai-gateway-network

networks:
  ai-gateway-network:
    driver: bridge

volumes:
  redis-data:

#!/usr/bin/env python3
"""
Comprehensive AI API Gateway Testing Suite
Tests functionality, performance, and security of the gateway
"""

import requests
import json
import time
import statistics
from typing import Dict, List, Any
import threading
import concurrent.futures

class AIGatewayTester:
    def __init__(self, base_url: str = "http://localhost:8080"):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.results = []
        
    def test_connectivity(self) -> Dict[str, Any]:
        """Test basic connectivity to gateway"""
        print("🔗 Testing gateway connectivity...")
        
        tests = {
            "gateway_health": f"{self.base_url}/health",
            "redis_connection": f"{self.base_url}/health/redis",
            "openai_connection": f"{self.base_url}/health/openai"
        }
        
        results = {}
        for name, endpoint in tests.items():
            try:
                start = time.time()
                response = self.session.get(endpoint, timeout=5)
                latency = (time.time() - start) * 1000
                
                results[name] = {
                    "status": "PASS" if response.status_code == 200 else "FAIL",
                    "status_code": response.status_code,
                    "latency_ms": round(latency, 2),
                    "response_time": response.elapsed.total_seconds()
                }
                
                if response.status_code == 200:
                    print(f"  ✅ {name}: {response.status_code} ({latency:.0f}ms)")
                else:
                    print(f"  ❌ {name}: {response.status_code}")
                    
            except Exception as e:
                results[name] = {
                    "status": "ERROR",
                    "error": str(e)
                }
                print(f"  ❌ {name}: {str(e)}")
        
        return results
    
    def test_rate_limiting(self, requests_per_minute: int = 100) -> Dict[str, Any]:
        """Test rate limiting functionality"""
        print(f"⚡ Testing rate limiting ({requests_per_minute} requests/min)...")
        
        endpoint = f"{self.base_url}/v1/chat/completions"
        headers = {"Authorization": "Bearer test-token"}
        data = {
            "model": "gpt-4",
            "messages": [{"role": "user", "content": "Test"}],
            "max_tokens": 10
        }
        
        # Concurrent request testing
        with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
            futures = []
            start_time = time.time()
            
            for i in range(requests_per_minute):
                futures.append(executor.submit(
                    self.session.post,
                    endpoint,
                    json=data,
                    headers=headers
                ))
            
            # Collect results
            responses = []
            for future in concurrent.futures.as_completed(futures):
                try:
                    response = future.result(timeout=10)
                    responses.append({
                        "status_code": response.status_code,
                        "headers": dict(response.headers)
                    })
                except Exception as e:
                    responses.append({
                        "status_code": 0,
                        "error": str(e)
                    })
        
        # Analyze results
        total_time = time.time() - start_time
        successful = sum(1 for r in responses if r.get("status_code") == 200)
        rate_limited = sum(1 for r in responses if r.get("status_code") == 429)
        
        results = {
            "total_requests": len(responses),
            "successful_requests": successful,
            "rate_limited_requests": rate_limited,
            "requests_per_second": len(responses) / total_time,
            "test_duration_seconds": total_time
        }
        
        print(f"  📊 Results: {successful} successful, {rate_limited} rate-limited")
        print(f"  ⏱️  Duration: {total_time:.2f}s ({results['requests_per_second']:.1f} req/sec)")
        
        return results