วิธีตั้งค่า AI API Proxy พร้อม Authentication Middleware สำหรับ Production

เมื่อสัปดาห์ที่แล้ว ผมเจอปัญหาหนักใจมากกับระบบ Production ที่ใช้งานอยู่ บริการ AI API ที่ใช้อยู่เกิด ConnectionError: timeout after 30 seconds ตอนที่ request พุ่งสูงขึ้น แถมยังมีคนไม่หวังดีพยายามเข้าถึง API โดยไม่มีสิทธิ์ จนเกิด 401 Unauthorized ติดต่อกันหลายร้อยครั้ง

หลังจากปวดหัวมาหลายวัน ผมตัดสินใจสร้าง Proxy Server พร้อม Authentication Middleware ขึ้นมาเอง และเลือกใช้ HolySheep AI เป็น API Provider เพราะราคาถูกกว่าเดิมถึง 85%+ (¥1=$1) รองรับ WeChat/Alipay และมี Latency ต่ำกว่า 50ms

ทำไมต้องสร้าง AI API Proxy?

• ป้องกัน API Key รั่วไหล — Key หลักซ่อนอยู่บน Server ฝั่ง Backend
• ควบคุม Rate Limit — จำกัดจำนวน request ต่อผู้ใช้งาน
• เพิ่ม Caching Layer — ลดค่าใช้จ่ายด้วยการเก็บผลลัพธ์ที่ซ้ำ
• เปลี่ยน Provider ได้ง่าย — ถ้า HolySheep มีปัญหา สลับไป Provider อื่นได้ทันที

Architecture ภาพรวมของระบบ

Client Request
      │
      ▼
┌─────────────────┐
│   API Gateway   │  ← Rate Limit, IP Blocking
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Auth Middleware│  ← Token Validation, API Key Check
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Proxy Server   │  ← Request Transform, Response Cache
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  HolySheep API  │  ← https://api.holysheep.ai/v1
│  (YOUR_KEY)     │
└─────────────────┘

การติดตั้งและตั้งค่าโปรเจกต์

# สร้างโฟลเดอร์โปรเจกต์
mkdir ai-proxy-server && cd ai-proxy-server

สร้าง virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

ติดตั้ง dependencies
pip install fastapi uvicorn python-jose passlib python-multipart
pip install httpx redis aiofiles pydantic-settings

# config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # HolySheep API Configuration
    HOLYSHEEP_API_KEY: str = "YOUR_HOLYSHEEP_API_KEY"
    HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
    
    # Authentication
    JWT_SECRET_KEY: str = "your-super-secret-key-change-in-production"
    JWT_ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
    
    # Rate Limiting
    RATE_LIMIT_REQUESTS: int = 100
    RATE_LIMIT_WINDOW_SECONDS: int = 60
    
    # Server
    HOST: str = "0.0.0.0"
    PORT: int = 8000
    
    class Config:
        env_file = ".env"
        extra = "allow"

@lru_cache()
def get_settings():
    return Settings()

settings = get_settings()

สร้าง Authentication Middleware

# auth_middleware.py
from fastapi import Request, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from datetime import datetime, timedelta
from typing import Optional
import hashlib
import time

from config import settings

In-memory rate limiting storage (ใช้ Redis ใน Production)
rate_limit_storage = {}

class AuthMiddleware:
    """Middleware สำหรับตรวจสอบ JWT Token และ Rate Limiting"""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            request = Request(scope, receive)
            
            # Skip auth สำหรับ path สาธารณะ
            public_paths = ["/docs", "/redoc", "/openapi.json", "/health"]
            if any(request.url.path.startswith(path) for path in public_paths):
                await self.app(scope, receive, send)
                return
            
            # ตรวจสอบ Authorization Header
            auth_header = request.headers.get("Authorization")
            if not auth_header or not auth_header.startswith("Bearer "):
                await self._send_401_response(send, "Missing or invalid Authorization header")
                return
            
            token = auth_header.replace("Bearer ", "")
            
            # ตรวจสอบ JWT Token
            try:
                payload = jwt.decode(
                    token, 
                    settings.JWT_SECRET_KEY, 
                    algorithms=[settings.JWT_ALGORITHM]
                )
                user_id = payload.get("sub")
                if not user_id:
                    raise JWTError("Invalid token payload")
            except JWTError as e:
                await self._send_401_response(send, f"Invalid token: {str(e)}")
                return
            
            # ตรวจสอบ Rate Limiting
            client_ip = request.client.host if request.client else "unknown"
            if not self._check_rate_limit(user_id, client_ip):
                await self._send_429_response(send, "Rate limit exceeded")
                return
            
            # เพิ่ม user info เข้า scope
            scope["state"] = {"user_id": user_id}
        
        await self.app(scope, receive, send)
    
    def _check_rate_limit(self, user_id: str, client_ip: str) -> bool:
        """ตรวจสอบ rate limit ด้วย sliding window algorithm"""
        current_time = time.time()
        key = f"{user_id}:{client_ip}"
        
        if key not in rate_limit_storage:
            rate_limit_storage[key] = []
        
        # ลบ timestamp เก่าออก
        cutoff_time = current_time - settings.RATE_LIMIT_WINDOW_SECONDS
        rate_limit_storage[key] = [
            t for t in rate_limit_storage[key] if t > cutoff_time
        ]
        
        # ตรวจสอบจำนวน request
        if len(rate_limit_storage[key]) >= settings.RATE_LIMIT_REQUESTS:
            return False
        
        rate_limit_storage[key].append(current_time)
        return True
    
    async def _send_401_response(self, send, message: str):
        await send({
            "type": "http.response.start",
            "status": status.HTTP_401_UNAUTHORIZED,
            "headers": [[b"content-type", b"application/json"]],
        })
        await send({
            "type": "http.response.body",
            "body": f'{{"detail": "{message}"}}'.encode(),
        })
    
    async def _send_429_response(self, send, message: str):
        await send({
            "type": "http.response.start",
            "status": status.HTTP_429_TOO_MANY_REQUESTS,
            "headers": [[b"content-type", b"application/json"]],
        })
        await send({
            "type": "http.response.body",
            "body": f'{{"detail": "{message}"}}'.encode(),
        })

สร้าง API Routes สำหรับ Proxy

# main.py
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from jose import jwt
import httpx
import asyncio
from datetime import datetime, timedelta

from config import settings
from auth_middleware import AuthMiddleware

app = FastAPI(
    title="AI API Proxy Server",
    description="Proxy server พร้อม Authentication สำหรับ HolySheep AI"
)

เพิ่ม Auth Middleware
app.add_middleware(AuthMiddleware)

class ChatRequest(BaseModel):
    model: str = "gpt-4"
    messages: list
    temperature: float = 0.7
    max_tokens: int = 1000

class TokenRequest(BaseModel):
    username: str
    password: str

class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"
    expires_in: int

──────────── Authentication Endpoints ────────────

@app.post("/auth/token", response_model=TokenResponse)
async def login(request: TokenRequest):
    """
    สร้าง JWT Token สำหรับผู้ใช้งาน
    ใน Production ควรเชื่อมต่อกับ Database จริง
    """
    # TODO: ตรวจสอบ username/password กับ Database
    if request.username != "demo" or request.password != "demo123":
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {
        "sub": request.username,
        "exp": expire,
        "iat": datetime.utcnow()
    }
    
    token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
    
    return TokenResponse(
        access_token=token,
        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
    )

──────────── Proxy Endpoints ────────────

@app.post("/v1/chat/completions")
async def chat_completions(
    request: Request,
    chat_request: ChatRequest
):
    """
    Proxy endpoint สำหรับ Chat Completions
    ส่งต่อ request ไปยัง HolySheep AI
    """
    user_id = request.state.user_id if hasattr(request.state, "user_id") else "anonymous"
    
    headers = {
        "Authorization": f"Bearer {settings.HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": chat_request.model,
        "messages": chat_request.messages,
        "temperature": chat_request.temperature,
        "max_tokens": chat_request.max_tokens
    }
    
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                f"{settings.HOLYSHEEP_BASE_URL}/chat/completions",
                json=payload,
                headers=headers
            )
            
            response.raise_for_status()
            result = response.json()
            
            return result
            
    except httpx.TimeoutException:
        raise HTTPException(
            status_code=504,
            detail="Gateway timeout - HolySheep API ไม่ตอบสนอง"
        )
    except httpx.HTTPStatusError as e:
        raise HTTPException(
            status_code=e.response.status_code,
            detail=f"上游 API 错误: {e.response.text}"
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"代理服务器错误: {str(e)}"
        )

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "service": "AI API Proxy",
        "provider": "HolySheep AI",
        "timestamp": datetime.utcnow().isoformat()
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host=settings.HOST,
        port=settings.PORT,
        reload=True
    )

วิธีใช้งาน Proxy ฝั่ง Client

# client_example.py
import requests
from datetime import datetime

class HolySheepProxyClient:
    """Client สำหรับเชื่อมต่อกับ AI API Proxy ของเรา"""
    
    def __init__(self, proxy_url: str, username: str, password: str):
        self.proxy_url = proxy_url.rstrip("/")
        self.username = username
        self.password = password
        self.access_token = None
        self.token_expires_at = None
    
    def _get_token(self) -> str:
        """ขอ JWT Token จาก Proxy"""
        response = requests.post(
            f"{self.proxy_url}/auth/token",
            json={"username": self.username, "password": self.password}
        )
        response.raise_for_status()
        data = response.json()
        self.access_token = data["access_token"]
        self.token_expires_at = datetime.now().timestamp() + data["expires_in"]
        return self.access_token
    
    def _ensure_token(self):
        """ตรวจสอบว่า token ยังไม่หมดอายุ"""
        if not self.access_token or (
            self.token_expires_at and 
            datetime.now().timestamp() > self.token_expires_at - 60
        ):
            self._get_token()
    
    def chat(self, messages: list, model: str = "gpt-4", **kwargs):
        """
        ส่ง request ไปยัง Chat Completions ผ่าน Proxy
        
        Args:
            messages: รายการ message ในรูปแบบ [{"role": "user", "content": "..."}]
            model: ชื่อ model (gpt-4, gpt-3.5-turbo, claude-3, etc.)
            **kwargs: temperature, max_tokens, etc.
        """
        self._ensure_token()
        
        headers = {"Authorization": f"Bearer {self.access_token}"}
        payload = {
            "model": model,
            "messages": messages,
            **kwargs
        }
        
        response = requests.post(
            f"{self.proxy_url}/v1/chat/completions",
            json=payload,
            headers=headers,
            timeout=60
        )
        response.raise_for_status()
        return response.json()

──────────── ตัวอย่างการใช้งาน ────────────

if __name__ == "__main__":
    # เชื่อมต่อกับ Proxy Server ของเรา
    client = HolySheepProxyClient(
        proxy_url="http://localhost:8000",
        username="demo",
        password="demo123"
    )
    
    # ส่ง request ผ่าน Proxy ไปยัง HolySheep AI
    response = client.chat(
        messages=[
            {"role": "system", "content": "คุณเป็นผู้ช่วยที่เป็นมิตร"},
            {"role": "user", "content": "สวัสดี บอกข้อมูลราคา AI API หน่อยได้ไหม?"}
        ],
        model="gpt-4",
        temperature=0.7,
        max_tokens=500
    )
    
    print("Response:", response["choices"][0]["message"]["content"])

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: ConnectionError: timeout after 30 seconds

# ❌ วิธีผิด - ใช้ timeout สั้นเกินไป
async with httpx.AsyncClient(timeout=10.0) as client:
    response = await client.post(url, json=payload)

✅ วิธีถูก - ตั้ง timeout ให้เหมาะสมกับประเภท request
async with httpx.AsyncClient(
    timeout=httpx.Timeout(
        connect=10.0,    # เวลาติดต่อ Server
        read=120.0,      # เวลาอ่าน Response (AI ต้องใช้เวลา)
        write=10.0,      # เวลาเขียน Request
        pool=30.0        # เวลารอ Connection Pool
    )
) as client:
    response = await client.post(url, json=payload)

หรือแบบง่าย
async with httpx.AsyncClient(timeout=120.0) as client:
    response = await client.post(url, json=payload)

กรณีที่ 2: 401 Unauthorized - Invalid token

# ❌ วิธีผิด - ตรวจสอบ token ไม่ถูกต้อง
if "token" not in request.headers:
    raise HTTPException(status_code=401)

✅ วิธีถูก - ตรวจสอบอย่างละเอียดด้วย proper validation
from fastapi.security import HTTPBearer
from jose import jwt, JWTError

security = HTTPBearer()

async def verify_token(request: Request, credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    
    try:
        payload = jwt.decode(
            token,
            settings.JWT_SECRET_KEY,
            algorithms=[settings.JWT_ALGORITHM]
        )
        
        # ตรวจสอบว่า token ยังไม่หมดอายุ
        exp = payload.get("exp")
        if exp and datetime.fromtimestamp(exp) < datetime.now():
            raise HTTPException(
                status_code=401,
                detail="Token has expired"
            )
        
        # ตรวจสอบ required claims
        if "sub" not in payload:
            raise HTTPException(
                status_code=401,
                detail="Invalid token structure"
            )
            
        return payload
        
    except JWTError as e:
        raise HTTPException(
            status_code=401,
            detail=f"Token validation failed: {str(e)}"
        )

กรณีที่ 3: Rate Limit Exceeded - ถูกบล็อกจาก HolySheep API

# ❌ วิธีผิด - ไม่มี retry logic เมื่อถูก rate limit
response = await client.post(url, json=payload)

✅ วิธีถูก - ใช้ Exponential Backoff พร้อม Retry
import asyncio
from typing import Optional

class RateLimitHandler:
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    async def request_with_retry(
        self,
        client: httpx.AsyncClient,
        method: str,
        url: str,
        **kwargs
    ) -> httpx.Response:
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                response = await client.request(method, url, **kwargs)
                
                # ถ้าได้รับ 429 Rate Limited
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 60))
                    wait_time = retry_after if retry_after else self.base_delay * (2 ** attempt)
                    
                    print(f"Rate limited. Waiting {wait_time}s before retry...")
                    await asyncio.sleep(wait_time)
                    continue
                
                response.raise_for_status()
                return response
                
            except (httpx.TimeoutException, httpx.ConnectError) as e:
                last_exception = e
                if attempt < self.max_retries:
                    delay = self.base_delay * (2 ** attempt)
                    await asyncio.sleep(delay)
                continue
        
        raise last_exception  # ถ้า retry หมดแล้วยังไม่สำเร็จ

วิธีใช้งาน
handler = RateLimitHandler(max_retries=3, base_delay=2.0)
response = await handler.request_with_retry(
    client,
    "POST",
    f"{settings.HOLYSHEEP_BASE_URL}/chat/completions",
    json=payload,
    headers=headers
)

สรุปความสำเร็จ

หลังจากติดตั้ง Proxy Server พร้อม Authentication Middleware นี้แล้ว ผลลัพธ์ที่ได้คือ:

• ✅ ปลอดภัยขึ้น — API Key ไม่รั่วไหลไปฝั่ง Client แล้ว
• ✅ เร็วขึ้น — Latency เฉลี่ย 45ms (เมื่อใช้ HolySheep AI)
• ✅ ประหยัดขึ้น — ค่าใช้จ่ายลดลง 85%+ เพราะอัตราแลกเปลี่ยน ¥1=$1
• ✅ ควบคุมได้ — จำกัด rate limit ได้ตามต้องการ

ราคา AI API จาก HolySheep AI (อัปเดต 2026)

Model	ราคา (USD/MTok)	เทียบกับ OpenAI
GPT-4.1	$8	ถูกกว่าเดิม
Claude Sonnet 4.5	$15	ราคาดี
Gemini 2.5 Flash	$2.50	เหมาะสำหรับงานทั่วไป
DeepSeek V3.2	$0.42	ถูกที่สุด!

เมื่อใช้ API ผ่าน Proxy ของเรา คุณสามารถเปลี่ยน provider ได้ง่ายโดยไม่ต้องแก้ไขโค้ดฝั่ง Client เลย

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน