การจัดการ API Key Scope Limitations เป็นหัวใจสำคัญของการ Deploy ระบบ AI ในระดับ Production โดยเฉพาะเมื่อต้องควบคุมการใช้งานของผู้ใช้หลายรายบนแพลตฟอร์มเดียว บทความนี้จะพาคุณสร้างระบบ Scope ที่ครอบคลุม ตั้งแต่พื้นฐานจนถึง Advanced Use Cases โดยใช้ HolySheep AI เป็นตัวอย่างหลัก
ทำไมต้อง Implement API Key Scope Limitations
ในการพัฒนา SaaS ที่ใช้ AI คุณต้องควบคุมหลายปัจจัย:
- การจำกัดงบประมาณ — ป้องกันการใช้งานเกินจำนวนที่กำหนด
- การจำกัดโมเดล — กำหนดว่าผู้ใช้แต่ละรายใช้ได้เฉพาะโมเดลใดบ้าง
- การติดตามการใช้งาน — วิเคราะห์ patterns และ optimize ต้นทุน
- ความปลอดภัย — ป้องกันการลักลอบใช้ API Key
Architecture ของระบบ Scope Limitations
ระบบ Scope ที่ดีต้องมี 4 ชั้นหลัก:
- Rate Limiting Layer — จำกัดจำนวน request ต่อนาที
- Quota Management — จำกัดปริมาณ token หรือ budget
- Model Access Control — กำหนดว่าใช้โมเดลใดได้บ้าง
- Audit Logging — บันทึกการใช้งานทุกรายการ
Implementation ด้วย Python + HolySheep API
ด้านล่างคือตัวอย่างการสร้าง API Key Scope Manager ที่ใช้งานได้จริง:
import httpx
import time
import hashlib
from dataclasses import dataclass, field
from typing import Optional, List, Dict
from datetime import datetime, timedelta
import json
@dataclass
class APIKeyScope:
"""โครงสร้างข้อมูล Scope สำหรับ API Key"""
key_id: str
scopes: List[str] = field(default_factory=list)
daily_limit: int = 10000 # token ต่อวัน
monthly_limit: int = 100000
allowed_models: List[str] = field(default_factory=list)
rate_limit_per_minute: int = 60
current_daily_usage: int = 0
current_monthly_usage: int = 0
last_reset: datetime = field(default_factory=datetime.now)
class HolySheepScopeManager:
"""ตัวจัดการ Scope Limitations สำหรับ HolySheep AI"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, master_key: str):
self.master_key = master_key
self.client = httpx.Client(timeout=30.0)
self.scopes: Dict[str, APIKeyScope] = {}
self.request_history: Dict[str, List[float]] = {}
def create_scoped_key(
self,
key_id: str,
allowed_models: List[str],
daily_limit: int = 10000,
monthly_limit: int = 100000
) -> APIKeyScope:
"""สร้าง API Key ใหม่พร้อมกำหนด Scope"""
scope = APIKeyScope(
key_id=key_id,
scopes=["chat:read", "chat:write"],
daily_limit=daily_limit,
monthly_limit=monthly_limit,
allowed_models=allowed_models,
rate_limit_per_minute=60
)
self.scopes[key_id] = scope
self.request_history[key_id] = []
print(f"✅ สร้าง Scoped Key: {key_id}")
print(f" โมเดลที่อนุญาต: {', '.join(allowed_models)}")
print(f" วงเงินรายวัน: {daily_limit:,} tokens")
print(f" วงเงินรายเดือน: {monthly_limit:,} tokens")
return scope
def validate_request(self, key_id: str, model: str, estimated_tokens: int) -> tuple[bool, str]:
"""ตรวจสอบคำขอก่อนส่งไปยัง API"""
if key_id not in self.scopes:
return False, "API Key ไม่ถูกต้อง"
scope = self.scopes[key_id]
# ตรวจสอบการ reset รายวัน
if datetime.now() - scope.last_reset > timedelta(days=1):
scope.current_daily_usage = 0
scope.last_reset = datetime.now()
print(f"🔄 Reset Daily Quota สำหรับ {key_id}")
# ตรวจสอบ Rate Limit
current_time = time.time()
self.request_history[key_id] = [
t for t in self.request_history[key_id]
if current_time - t < 60
]
if len(self.request_history[key_id]) >= scope.rate_limit_per_minute:
return False, f"Rate Limit: เกิน {scope.rate_limit_per_minute} requests/นาที"
# ตรวจสอบโมเดล
if model not in scope.allowed_models:
return False, f"โมเดล '{model}' ไม่อยู่ใน Scope ที่อนุญาต"
# ตรวจสอบ Quota
if scope.current_daily_usage + estimated_tokens > scope.daily_limit:
return False, f"Daily Quota เกิน ({scope.current_daily_usage}/{scope.daily_limit})"
if scope.current_monthly_usage + estimated_tokens > scope.monthly_limit:
return False, f"Monthly Quota เกิน ({scope.current_monthly_usage}/{scope.monthly_limit})"
self.request_history[key_id].append(current_time)
return True, "OK"
def execute_chat_completion(
self,
key_id: str,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict:
"""ส่ง request ไปยัง HolySheep AI พร้อม Scope Validation"""
# Estimate tokens (rough calculation)
estimated_tokens = sum(len(str(m)) for m in messages) + max_tokens
# Validate
is_valid, error_msg = self.validate_request(key_id, model, estimated_tokens)
if not is_valid:
return {
"error": True,
"message": error_msg,
"key_id": key_id,
"timestamp": datetime.now().isoformat()
}
# Execute request
headers = {
"Authorization": f"Bearer {self.master_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = self.client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
# Update usage
usage = result.get("usage", {})
tokens_used = usage.get("total_tokens", estimated_tokens)
self.scopes[key_id].current_daily_usage += tokens_used
self.scopes[key_id].current_monthly_usage += tokens_used
# Log
self._log_request(key_id, model, tokens_used)
return {
"error": False,
"data": result,
"usage": tokens_used,
"remaining_daily": self.scopes[key_id].daily_limit - self.scopes[key_id].current_daily_usage,
"remaining_monthly": self.scopes[key_id].monthly_limit - self.scopes[key_id].current_monthly_usage
}
except httpx.HTTPStatusError as e:
return {
"error": True,
"message": f"HTTP Error: {e.response.status_code}",
"detail": e.response.text
}
def _log_request(self, key_id: str, model: str, tokens: int):
"""บันทึก Log การใช้งาน"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"key_id": key_id,
"model": model,
"tokens": tokens
}
print(f"📊 {log_entry['timestamp']} | {key_id} | {model} | {tokens} tokens")
def get_scope_status(self, key_id: str) -> Dict:
"""ดึงสถานะ Scope ปัจจุบัน"""
if key_id not in self.scopes:
return {"error": "Key ไม่พบ"}
scope = self.scopes[key_id]
return {
"key_id": key_id,
"daily_usage": scope.current_daily_usage,
"daily_limit": scope.daily_limit,
"daily_remaining": scope.daily_limit - scope.current_daily_usage,
"monthly_usage": scope.current_monthly_usage,
"monthly_limit": scope.monthly_limit,
"monthly_remaining": scope.monthly_limit - scope.current_monthly_usage,
"allowed_models": scope.allowed_models,
"rate_limit_per_minute": scope.rate_limit_per_minute,
"requests_last_minute": len(self.request_history.get(key_id, []))
}
ตัวอย่างการใช้งาน
manager = HolySheepScopeManager("YOUR_HOLYSHEEP_API_KEY")
สร้าง Key สำหรับ Developer Tier
dev_scope = manager.create_scoped_key(
key_id="dev_001",
allowed_models=["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash"],
daily_limit=50000,
monthly_limit=500000
)
สร้าง Key สำหรับ Free Tier
free_scope = manager.create_scoped_key(
key_id="free_001",
allowed_models=["deepseek-v3.2", "gemini-2.5-flash"],
daily_limit=5000,
monthly_limit=50000
)
ทดสอบ request
messages = [{"role": "user", "content": "อธิบายเรื่อง API Scope"}]
result = manager.execute_chat_completion(
key_id="dev_001",
model="gpt-4.1",
messages=messages
)
แสดงสถานะ
status = manager.get_scope_status("dev_001")
print(f"\n📈 สถานะ: {json.dumps(status, indent=2, ensure_ascii=False)}")Middleware สำหรับ FastAPI
หากคุณใช้ FastAPI เป็น Backend Framework ด้านล่างคือ Middleware ที่พร้อมใช้งาน:
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional, List
import time
import jwt
from datetime import datetime, timedelta
app = FastAPI(title="AI API Gateway with Scope Control")
security = HTTPBearer()
class ScopeConfig(BaseModel):
allowed_models: List[str]
daily_token_limit: int
monthly_token_limit: int
rate_limit_rpm: int
class APIKeyPayload(BaseModel):
sub: str # user_id
scopes: ScopeConfig
exp: datetime
ฐานข้อมูล Scope (ใช้ Redis ใน Production)
scope_db = {
"user_premium_001": ScopeConfig(
allowed_models=["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"],
daily_token_limit=500000,
monthly_token_limit=5000000,
rate_limit_rpm=120
),
"user_basic_001": ScopeConfig(
allowed_models=["gemini-2.5-flash", "deepseek-v3.2"],
daily_token_limit=50000,
monthly_token_limit=500000,
rate_limit_rpm=30
),
"user_free_001": ScopeConfig(
allowed_models=["deepseek-v3.2"],
daily_token_limit=5000,
monthly_token_limit=50000,
rate_limit_rpm=10
)
}
ติดตามการใช้งาน
usage_tracker = {}
rate_limiter = {}
def verify_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""ตรวจสอบ API Key และ Decode JWT"""
token = credentials.credentials
try:
# ใน Production ใช้ JWT library ที่เหมาะสม
# ตัวอย่างนี้ใช้ decode อย่างง่าย
payload = decode_jwt_token(token)
if payload["exp"] < datetime.now().timestamp():
raise HTTPException(status_code=401, detail="Token หมดอายุ")
return payload
except Exception as e:
raise HTTPException(status_code=401, detail=f"การยืนยันตัวตนล้มเหลว: {str(e)}")
def decode_jwt_token(token: str) -> dict:
"""Decode JWT Token (ตัวอย่าง)"""
# ใน Production ใช้ PyJWT: jwt.decode(token, SECRET, algorithms=["HS256"])
parts = token.split(".")
if len(parts) != 3:
raise ValueError("Invalid Token Format")
# Decode payload (simplified)
import base64
import json
padding = 4 - len(parts[1]) % 4
if padding != 4:
parts[1] += "=" * padding
payload = json.loads(base64.urlsafe_b64decode(parts[1]))
return payload
def check_rate_limit(user_id: str, limit: int) -> bool:
"""ตรวจสอบ Rate Limit ต่อนาที"""
current_time = time.time()
if user_id not in rate_limiter:
rate_limiter[user_id] = []
# ลบ request ที่เก่ากว่า 60 วินาที
rate_limiter[user_id] = [
t for t in rate_limiter[user_id]
if current_time - t < 60
]
if len(rate_limiter[user_id]) >= limit:
return False
rate_limiter[user_id].append(current_time)
return True
def check_quota(user_id: str, requested_tokens: int, scope: ScopeConfig) -> tuple[bool, str]:
"""ตรวจสอบ Quota รายวันและรายเดือน"""
today = datetime.now().strftime("%Y-%m-%d")
this_month = datetime.now().strftime("%Y-%m")
if user_id not in usage_tracker:
usage_tracker[user_id] = {"daily": {}, "monthly": {}}
# Initialize if not exists
if today not in usage_tracker[user_id]["daily"]:
usage_tracker[user_id]["daily"][today] = 0
if this_month not in usage_tracker[user_id]["monthly"]:
usage_tracker[user_id]["monthly"][this_month] = 0
daily_usage = usage_tracker[user_id]["daily"][today]
monthly_usage = usage_tracker[user_id]["monthly"][this_month]
if daily_usage + requested_tokens > scope.daily_token_limit:
return False, f"เกินวงเงินรายวัน ({daily_usage}/{scope.daily_token_limit})"
if monthly_usage + requested_tokens > scope.monthly_token_limit:
return False, f"เกินวงเงินรายเดือน ({monthly_usage}/{scope.monthly_token_limit})"
return True, "OK"
@app.post("/v1/chat/completions")
async def chat_completions(
request: Request,
payload: dict,
auth: dict = Depends(verify_api_key)
):
"""Endpoint หลักสำหรับ Chat Completions"""
user_id = auth["sub"]
model = payload.get("model", "")
messages = payload.get("messages", [])
# ตรวจสอบ Scope
if user_id not in scope_db:
raise HTTPException(status_code=403, detail="ไม่พบ Scope สำหรับผู้ใช้นี้")
scope = scope_db[user_id]
# 1. ตรวจสอบ Rate Limit
if not check_rate_limit(user_id, scope.rate_limit_rpm):
raise HTTPException(
status_code=429,
detail=f"Rate Limit: เกิน {scope.rate_limit_rpm} requests/นาที"
)
# 2. ตรวจสอบโมเดลที่อนุญาต
if model not in scope.allowed_models:
raise HTTPException(
status_code=403,
detail=f"โมเดล '{model}' ไม่อยู่ใน Scope ที่อนุญาต. อนุญาต: {scope.allowed_models}"
)
# 3. ประมาณการ tokens
estimated_tokens = sum(len(str(m.get("content", ""))) for m in messages)
estimated_tokens += payload.get("max_tokens", 2048)
# 4. ตรวจสอบ Quota
is_valid, msg = check_quota(user_id, estimated_tokens, scope)
if not is_valid:
raise HTTPException(status_code=402, detail=msg)
# 5. ส่งต่อไปยัง HolySheep API
import httpx
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json=payload
)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
result = response.json()
# 6. อัปเดต Usage
actual_tokens = result.get("usage", {}).get("total_tokens", estimated_tokens)
today = datetime.now().strftime("%Y-%m-%d")
this_month = datetime.now().strftime("%Y-%m")
usage_tracker[user_id]["daily"][today] += actual_tokens
usage_tracker[user_id]["monthly"][this_month] += actual_tokens
return result
@app.get("/v1/scope/status")
async def get_scope_status(auth: dict = Depends(verify_api_key)):
"""ดึงสถานะ Scope ของผู้ใช้ปัจจุบัน"""
user_id = auth["sub"]
if user_id not in scope_db:
raise HTTPException(status_code=404, detail="ไม่พบ Scope")
scope = scope_db[user_id]
today = datetime.now().strftime("%Y-%m-%d")
this_month = datetime.now().strftime("%Y-%m")
daily_usage = usage_tracker.get(user_id, {}).get("daily", {}).get(today, 0)
monthly_usage = usage_tracker.get(user_id, {}).get("monthly", {}).get(this_month, 0)
return {
"user_id": user_id,
"allowed_models": scope.allowed_models,
"daily": {
"used": daily_usage,
"limit": scope.daily_token_limit,
"remaining": scope.daily_token_limit - daily_usage
},
"monthly": {
"used": monthly_usage,
"limit": scope.monthly_token_limit,
"remaining": scope.monthly_token_limit - monthly_usage
},
"rate_limit": {
"per_minute": scope.rate_limit_rpm,
"current_count": len(rate_limiter.get(user_id, []))
}
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)การ Monitoring และ Alerting
ระบบ Scope ที่ดีต้องมี Dashboard สำหรับติดตามการใช้งานแบบ Real-time:
import asyncio
from datetime import datetime
from collections import defaultdict
import json
class ScopeMonitor:
"""ระบบ Monitoring สำหรับ API Key Scopes"""
def __init__(self):
self.metrics = defaultdict(lambda: {
"total_requests": 0,
"total_tokens": 0,
"successful_requests": 0,
"failed_requests": 0,
"cost_by_model": defaultdict(float),
"hourly_usage": defaultdict(int),
"error_types": defaultdict(int)
})
# ราคาต่อ 1M tokens (อ้างอิงจาก HolySheep 2026)
self.pricing = {
"gpt-4.1": 8.0, # $8/M tokens
"claude-sonnet-4.5": 15.0, # $15/M tokens
"gemini-2.5-flash": 2.50, # $2.50/M tokens
"deepseek-v3.2": 0.42 # $0.42/M tokens
}
def record_request(
self,
key_id: str,
model: str,
tokens: int,
success: bool,
error_type: str = None
):
"""บันทึก metrics ของ request"""
m = self.metrics[key_id]
m["total_requests"] += 1
m["total_tokens"] += tokens
m["hourly_usage"][datetime.now().strftime("%Y-%m-%d %H:00")] += tokens
if success:
m["successful_requests"] += 1
# คำนวณค่าใช้จ่าย
cost = (tokens / 1_000_000) * self.pricing.get(model, 0)
m["cost_by_model"][model] += cost
else:
m["failed_requests"] += 1
if error_type:
m["error_types"][error_type] += 1
def get_dashboard_data(self, key_id: str) -> dict:
"""สร้างข้อมูลสำหรับ Dashboard"""
m = self.metrics[key_id]
total_cost = sum(m["cost_by_model"].values())
success_rate = (
m["successful_requests"] / m["total_requests"] * 100
if m["total_requests"] > 0 else 0
)
# หา peak hour
peak_hour = max(
m["hourly_usage"].items(),
key=lambda x: x[1],
default=("N/A", 0)
)
return {
"key_id": key_id,
"summary": {
"total_requests": m["total_requests"],
"total_tokens": m["total_tokens"],
"total_cost_usd": round(total_cost, 4),
"success_rate": round(success_rate, 2)
},
"cost_by_model": dict(m["cost_by_model"]),
"hourly_usage": dict(m["hourly_usage"]),
"error_summary": dict(m["error_types"]),
"peak_hour": {"time": peak_hour[0], "tokens": peak_hour[1]}
}
def check_anomalies(self, key_id: str, threshold_tokens_per_minute: int = 10000):
"""ตรวจจับความผิดปกติ"""
m = self.metrics[key_id]
anomalies = []
# ตรวจจับการใช้งานสูงผิดปกติ
current_hour = datetime.now().strftime("%Y-%m-%d %H:00")
if current_hour in m["hourly_usage"]:
hourly_tokens = m["hourly_usage"][current_hour]
if hourly_tokens > threshold_tokens_per_minute * 60:
anomalies.append({
"type": "HIGH_USAGE",
"message": f"ใช้งานสูงผิดปกติ: {hourly_tokens} tokens ในชั่วโมงนี้",
"severity": "WARNING"
})
# ตรวจจับอัตราความล้มเหลวสูง
if m["total_requests"] > 10:
failure_rate = m["failed_requests"] / m["total_requests"]
if failure_rate > 0.2: # > 20%
anomalies.append({
"type": "HIGH_FAILURE_RATE",
"message": f"อัตราความล้มเหลว: {failure_rate*100:.1f}%",
"severity": "CRITICAL"
})
return anomalies
def generate_report(self, key_id: str) -> str:
"""สร้างรายงานแบบ Text"""
data = self.get_dashboard_data(key_id)
anomalies = self.check_anomalies(key_id)
report = f"""
╔══════════════════════════════════════════════════════════════╗
║ API KEY SCOPE REPORT - {key_id:<20} ║
╠══════════════════════════════════════════════════════════════╣
║ Total Requests: {data['summary']['total_requests']:>10,} ║
║ Total Tokens: {data['summary']['total_tokens']:>10,} ║
║ Total Cost (USD): ${data['summary']['total_cost_usd']:>10.4f} ║
║ Success Rate: {data['summary']['success_rate']:>10.1f}% ║
╠══════════════════════════════════════════════════════════════╣
║ COST BY MODEL ║"""
for model, cost in data["cost_by_model"].items():
report += f"\n║ {model:<20} ${cost:>10.4f} ║"
report += """
╠══════════════════════════════════════════════════════════════╣
║ PEAK USAGE ║
║ Peak Hour: {time:<20} Tokens: {tokens:>10,} ║
╚══════════════════════════════════════════════════════════════╝""".format(
time=data["peak_hour"]["time"],
tokens=data["peak_hour"]["tokens"]
)
if anomalies:
report += "\n\n⚠️ ANOMALIES DETECTED:"
for a in anomalies:
report += f"\n [{a['severity']}] {a['message']}"
return report
ตัวอย่างการใช้งาน
monitor = ScopeMonitor()
จำลองข้อมูล
for i in range(100):
monitor.record_request(
key_id="dev_001",
model=["gpt-4.1", "gemini-2.5-flash", "deepseek-v3.2"][i % 3],
tokens=500 + (i * 10),
success=(i % 10 != 0),
error_type="TIMEOUT" if i % 20 == 0 else None
)
แสดงรายงาน
print(monitor.generate_report("dev_001"))
print("\n" + "="*60)
Dashboard JSON
dashboard = monitor.get_dashboard_data("dev_001")
print(json.dumps(dashboard, indent=2, ensure_ascii=False))ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. Error 401: Authentication Failed
สาเหตุ: API Key ไม่ถูกต้องหรือหมดอายุ
# ❌ วิธีผิด - Hardcode API Key ในโค้ด
response = httpx.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": "Bearer sk-1234567890"}
)
✅ วิธีถูก - ใช้ Environment Variable
import os
API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
if not API_KEY:
raise ValueError("HOLYSHEEP_API_KEY environment variable ไม่ได้กำหนดค่า")
response = httpx.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"}
)2. Error 429: Rate Limit Exceeded
สาเหตุ: ส่ง Request เกินจำนวนที่กำหนดต่อนาที
import time
import asyncio
✅ วิธีแก้ไข - Implement Retry with Exponential Backoff
async def call_with_retry(prompt: str, max_retries: int = 3):
for attempt in range(max_retries):
try:
response = await httpx.AsyncClient().post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"model": "gpt-4.1", "messages": [{"role": "user", "content": prompt}]}
)
if response.status_code == 429:
wait_time = 2 ** attempt # 1, 2, 4 วินาที
print(f"Rate Limited. รอ {wait_time} วินาที...")
await asyncio