Khi hệ thống AI production chạy 24/7, một sự cố nhỏ có thể gây ra thiệt hại lớn. Bài viết này sẽ hướng dẫn bạn xây dựng cơ chế alert toàn diện cho nền tảng Dify, kết hợp với HolySheep AI để đảm bảo hệ thống luôn ổn định.
Câu Chuyện Thực Tế: Startup AI Ở Hà Nội
Một startup AI tại Hà Nội chuyên cung cấp chatbot hỗ trợ khách hàng cho các sàn thương mại điện tử đã gặp sự cố nghiêm trọng. Vào một ngày đông tháng 12, toàn bộ hệ thống Dify của họ ngừng hoạt động trong 4 tiếng đồng hồ mà không ai hay biết. Kết quả: 2,400 khách hàng không được phản hồi, thiệt hại ước tính 150 triệu VNĐ.
Bối cảnh: Hệ thống xử lý 50,000 request/ngày, sử dụng Dify self-hosted trên AWS tại Singapore.
Điểm đau: Nhà cung cấp cũ có độ trễ trung bình 850ms, server thường xuyên timeout vào giờ cao điểm. Mỗi lần API key hết hạn, kỹ sư phải thức đêm để fix.
Giải pháp: Di chuyển sang HolySheep AI với base_url https://api.holysheep.ai/v1, đồng thời xây dựng hệ thống alert tự động.
Kết quả sau 30 ngày:
- Độ trễ trung bình: 850ms → 180ms (giảm 79%)
- Hóa đơn hàng tháng: $4,200 → $680 (tiết kiệm 84%)
- Tỷ lệ uptime: 94% → 99.7%
- Tỷ giá chỉ ¥1 = $1, tiết kiệm thêm 85% chi phí
Tại Sao Dify Cần Alert System?
Dify là nền tảng RAG & Flow Engine mạnh mẽ, nhưng khi kết nối với LLM provider, có rất nhiều điểm có thể xảy ra sự cố:
- API Timeout: Request treo không phản hồi
- Rate Limit: Vượt quota của provider
- Invalid API Key: Key hết hạn hoặc bị revoke
- Model Unavailable: Model tạm thời không khả dụng
- Network Latency: Độ trễ cao bất thường
Cài Đặt Alert机制 Trong Dify
Bước 1: Cấu Hình Health Check Endpoint
Tạo script Python để monitor Dify health status:
#!/usr/bin/env python3
"""
Dify Health Monitor - HolySheep AI Integration
Kiểm tra health status mỗi 30 giây
"""
import requests
import time
import json
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
Cấu hình HolySheep AI
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
Cấu hình Dify
DIFY_API_URL = "https://your-dify-instance.com"
DIFY_API_KEY = "your-dify-api-key"
Cấu hình alert
ALERT_THRESHOLDS = {
"latency_ms": 500,
"error_rate_percent": 5,
"timeout_count": 3,
"consecutive_failures": 3
}
class DifyAlertSystem:
def __init__(self):
self.last_check_time = time.time()
self.error_count = 0
self.alert_history = []
def check_dify_health(self):
"""Kiểm tra health của Dify instance"""
try:
response = requests.get(
f"{DIFY_API_URL}/health",
timeout=5
)
return response.status_code == 200
except requests.exceptions.Timeout:
print(f"[{datetime.now()}] Dify health check TIMEOUT")
return False
except Exception as e:
print(f"[{datetime.now()}] Dify health check ERROR: {e}")
return False
def test_llm_connection(self):
"""Test kết nối LLM qua HolySheep"""
try:
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "ping"}],
"max_tokens": 10
}
start_time = time.time()
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=10
)
latency = (time.time() - start_time) * 1000
if response.status_code == 200:
return {"status": "ok", "latency_ms": latency}
else:
return {"status": "error", "error": response.text}
except requests.exceptions.Timeout:
return {"status": "timeout", "latency_ms": 10000}
except Exception as e:
return {"status": "error", "error": str(e)}
def send_alert(self, alert_type, message, severity="high"):
"""Gửi thông báo alert qua nhiều kênh"""
alert = {
"timestamp": datetime.now().isoformat(),
"type": alert_type,
"message": message,
"severity": severity
}
self.alert_history.append(alert)
print(f"[ALERT] [{severity.upper()}] {alert_type}: {message}")
# Gửi email
self._send_email_alert(alert)
# Gửi webhook (Slack, Discord, etc.)
self._send_webhook_alert(alert)
def _send_email_alert(self, alert):
"""Gửi alert qua email"""
try:
msg = MIMEText(f"""
Dify Alert Notification
=======================
Type: {alert['type']}
Severity: {alert['severity']}
Message: {alert['message']}
Time: {alert['timestamp']}
Auto-generated by Dify Alert System
Powered by HolySheep AI
""")
msg['Subject'] = f"[{alert['severity'].upper()}] Dify Alert: {alert['type']}"
msg['From'] = "[email protected]"
msg['To'] = "[email protected]"
# Cấu hình SMTP
with smtplib.SMTP('smtp.gmail.com', 587) as server:
server.starttls()
server.login('[email protected]', 'your-app-password')
server.send_message(msg)
except Exception as e:
print(f"Email alert failed: {e}")
def _send_webhook_alert(self, alert):
"""Gửi alert qua webhook"""
try:
webhook_url = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
payload = {
"text": f":warning: *Dify Alert*\n*Type:* {alert['type']}\n*Message:* {alert['message']}",
"username": "Dify Alert Bot"
}
requests.post(webhook_url, json=payload, timeout=5)
except Exception as e:
print(f"Webhook alert failed: {e}")
def run_monitoring(self, interval=30):
"""Chạy monitoring loop"""
print(f"[{datetime.now()}] Starting Dify Alert System...")
print(f"Monitoring interval: {interval} seconds")
while True:
# Check Dify health
dify_healthy = self.check_dify_health()
# Test LLM connection
llm_result = self.test_llm_connection()
# Xử lý alerts
if not dify_healthy:
self.error_count += 1
if self.error_count >= ALERT_THRESHOLDS["consecutive_failures"]:
self.send_alert(
"DIFY_DOWN",
f"Dify instance unavailable for {self.error_count} consecutive checks",
"critical"
)
else:
self.error_count = 0
# Check latency
if llm_result["status"] == "ok":
if llm_result["latency_ms"] > ALERT_THRESHOLDS["latency_ms"]:
self.send_alert(
"HIGH_LATENCY",
f"LLM latency {llm_result['latency_ms']:.0f}ms exceeds threshold {ALERT_THRESHOLDS['latency_ms']}ms",
"warning"
)
elif llm_result["status"] == "timeout":
self.send_alert(
"LLM_TIMEOUT",
"LLM request timeout - possible network or provider issue",
"high"
)
time.sleep(interval)
if __name__ == "__main__":
monitor = DifyAlertSystem()
monitor.run_monitoring(interval=30)Bước 2: Cấu Hình Dify Webhook cho Real-time Alerts
Thêm webhook endpoint vào Dify để nhận thông báo real-time:
#!/usr/bin/env python3
"""
Dify Webhook Receiver - Nhận và xử lý alerts từ Dify
Triển khai trên Flask/FastAPI
"""
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import asyncio
import logging
from datetime import datetime
app = FastAPI(title="Dify Alert Webhook")
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
Cấu hình HolySheep AI cho retry logic
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class DifyWebhookPayload(BaseModel):
event: str
type: str
conversation_id: Optional[str] = None
message_id: Optional[str] = None
error: Optional[str] = None
duration: Optional[float] = None
metadata: Optional[dict] = {}
class AlertConfig:
ALERT_CHANNELS = {
"email": ["[email protected]", "[email protected]"],
"slack": "https://hooks.slack.com/services/YOUR/WEBHOOK",
"telegram": "https://api.telegram.org/botYOUR_BOT_TOKEN/sendMessage",
"telegram_chat_id": "-100123456789"
}
AUTO_RETRY = {
"enabled": True,
"max_retries": 3,
"retry_delay_seconds": 5
}
class AlertDispatcher:
"""Xử lý và gửi alerts đến các kênh"""
def __init__(self):
self.config = AlertConfig()
self.alert_queue = asyncio.Queue()
async def process_alert(self, payload: DifyWebhookPayload):
"""Xử lý alert từ Dify webhook"""
logger.info(f"Processing alert: {payload.event} - {payload.type}")
alert_data = {
"event": payload.event,
"type": payload.type,
"conversation_id": payload.conversation_id,
"message_id": payload.message_id,
"error": payload.error,
"duration_ms": payload.duration * 1000 if payload.duration else 0,
"timestamp": datetime.now().isoformat()
}
# Phân loại severity
severity = self._determine_severity(payload)
alert_data["severity"] = severity
# Kiểm tra và xử lý tự động
if self.config.AUTO_RETRY["enabled"]:
await self._auto_recovery_check(alert_data)
# Gửi đến các kênh
await self._dispatch_to_channels(alert_data)
# Log cho monitoring
await self._log_to_monitoring_system(alert_data)
return alert_data
def _determine_severity(self, payload: DifyWebhookPayload) -> str:
"""Xác định mức độ nghiêm trọng của alert"""
if payload.event in ["workflow.error", "app.error"]:
return "critical"
elif payload.type in ["timeout", "rate_limit_exceeded"]:
return "high"
elif payload.type == "slow_response":
return "warning"
else:
return "info"
async def _auto_recovery_check(self, alert_data: dict):
"""Tự động kiểm tra và khôi phục"""
if alert_data["type"] == "timeout":
logger.info("Detected timeout - checking HolySheep AI status...")
# Check HolySheep AI health
import aiohttp
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"}
try:
async with session.get(
f"{HOLYSHEEP_BASE_URL}/models",
headers=headers,
timeout=aiohttp.ClientTimeout(total=5)
) as resp:
if resp.status == 200:
logger.info("HolySheep AI is healthy - issue may be resolved")
else:
logger.warning(f"HolySheep AI returned status {resp.status}")
except Exception as e:
logger.error(f"Failed to check HolySheep AI: {e}")
async def _dispatch_to_channels(self, alert_data: dict):
"""Gửi alert đến tất cả các kênh đã cấu hình"""
channels = self.config.ALERT_CHANNELS
tasks = []
# Email
for email in channels["email"]:
tasks.append(self._send_email(email, alert_data))
# Slack
tasks.append(self._send_slack(alert_data))
# Telegram
tasks.append(self._send_telegram(alert_data))
await asyncio.gather(*tasks, return_exceptions=True)
async def _send_email(self, recipient: str, alert_data: dict):
"""Gửi email alert"""
logger.info(f"Sending email alert to {recipient}")
# Implement email sending logic here
pass
async def _send_slack(self, alert_data: dict):
"""Gửi Slack alert với format đẹp"""
import aiohttp
severity_emoji = {
"critical": ":rotating_light:",
"high": ":warning:",
"warning": ":large_yellow_circle:",
"info": ":information_source:"
}
payload = {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{severity_emoji.get(alert_data['severity'], ':bell:')} Dify Alert"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Event:*\n{alert_data['event']}"},
{"type": "mrkdwn", "text": f"*Type:*\n{alert_data['type']}"},
{"type": "mrkdwn", "text": f"*Severity:*\n{alert_data['severity']}"},
{"type": "mrkdwn", "text": f"*Duration:*\n{alert_data['duration_ms']:.0f}ms"}
]
},
{
"type": "context",
"elements": [
{"type": "mrkdwn", "text": f"Timestamp: {alert_data['timestamp']}"}
]
}
]
}
try:
async with aiohttp.ClientSession() as session:
await session.post(
self.config.ALERT_CHANNELS["slack"],
json=payload
)
except Exception as e:
logger.error(f"Failed to send Slack alert: {e}")
async def _send_telegram(self, alert_data: dict):
"""Gửi Telegram alert"""
import aiohttp
message = f"""
🔔 *Dify Alert*
📌 *Event:* {alert_data['event']}
⚠️ *Type:* {alert_data['type']}
🔴 *Severity:* {alert_data['severity']}
⏱️ *Duration:* {alert_data['duration_ms']:.0f}ms
🕐 *Time:* {alert_data['timestamp']}
"""
payload = {
"chat_id": self.config.ALERT_CHANNELS["telegram_chat_id"],
"text": message,
"parse_mode": "Markdown"
}
try:
async with aiohttp.ClientSession() as session:
await session.post(
self.config.ALERT_CHANNELS["telegram"],
json=payload
)
except Exception as e:
logger.error(f"Failed to send Telegram alert: {e}")
async def _log_to_monitoring_system(self, alert_data: dict):
"""Log alert vào hệ thống monitoring (Prometheus, Grafana, etc.)"""
# Prometheus metrics endpoint
logger.info(f"Alert logged: {alert_data}")
Initialize dispatcher
dispatcher = AlertDispatcher()
@app.post("/webhook/dify")
async def receive_dify_webhook(payload: DifyWebhookPayload):
"""Webhook endpoint nhận alerts từ Dify"""
try:
result = await dispatcher.process_alert(payload)
return {"status": "processed", "alert_id": result.get("message_id")}
except Exception as e:
logger.error(f"Failed to process webhook: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""Health check endpoint cho webhook service"""
return {"status": "healthy", "service": "dify-alert-webhook"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8080)Bước 3: Tích Hợp Prometheus Metrics
Xuất metrics để visualize trên Grafana:
#!/usr/bin/env python3
"""
Prometheus Metrics Exporter cho Dify Alert System
Thu thập và export metrics về performance và errors
"""
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import random
import time
from datetime import datetime
import requests
Prometheus metrics
DIFY_REQUEST_COUNT = Counter(
'dify_requests_total',
'Total Dify requests',
['status', 'model']
)
DIFY_LATENCY = Histogram(
'dify_request_latency_seconds',
'Dify request latency',
['model', 'endpoint']
)
DIFY_ERROR_COUNT = Counter(
'dify_errors_total',
'Total Dify errors',
['error_type', 'severity']
)
DIFY_ALERT_COUNT = Counter(
'dify_alerts_total',
'Total Dify alerts sent',
['alert_type', 'severity']
)
HOLYSHEEP_COST = Counter(
'holysheep_api_cost_dollars',
'Total cost in USD via HolySheep',
['model']
)
HOLYSHEEP_LATENCY = Histogram(
'holysheep_request_latency_ms',
'HolySheep API request latency in ms',
['model']
)
ACTIVE_SESSIONS = Gauge(
'dify_active_sessions',
'Number of active Dify sessions'
)
class MetricsCollector:
"""Thu thập metrics từ Dify và HolySheep"""
def __init__(self):
self.holysheep_base_url = "https://api.holysheep.ai/v1"
self.holysheep_api_key = "YOUR_HOLYSHEEP_API_KEY"
self.dify_api_url = "https://your-dify-instance.com"
# Model pricing (USD per 1M tokens) - HolySheep 2026
self.model_pricing = {
"gpt-4.1": 8.0, # GPT-4.1
"claude-sonnet-4.5": 15.0, # Claude Sonnet 4.5
"gemini-2.5-flash": 2.50, # Gemini 2.5 Flash
"deepseek-v3.2": 0.42 # DeepSeek V3.2
}
def simulate_request(self, model: str, tokens_used: int):
"""Simulate một request để thu thập metrics"""
# Random latency để simulate
latency_ms = random.uniform(100, 500)
# Record latency
DIFY_LATENCY.labels(model=model, endpoint="chat").observe(latency_ms / 1000)
HOLYSHEEP_LATENCY.labels(model=model).observe(latency_ms)
# Calculate cost
input_tokens = tokens_used // 2
output_tokens = tokens_used // 2
cost = (input_tokens + output_tokens) / 1_000_000 * self.model_pricing.get(model, 8.0)
HOLYSHEEP_COST.labels(model=model).inc(cost)
# Random error simulation (5% error rate)
if random.random() < 0.05:
error_types = ["timeout", "rate_limit", "invalid_key", "model_unavailable"]
error_type = random.choice(error_types)
severities = {"timeout": "high", "rate_limit": "medium", "invalid_key": "critical", "model_unavailable": "high"}
DIFY_ERROR_COUNT.labels(error_type=error_type, severity=severities[error_type]).inc()
DIFY_ALERT_COUNT.labels(alert_type=f"error_{error_type}", severity=severities[error_type]).inc()
# Success
DIFY_REQUEST_COUNT.labels(status="success", model=model).inc()
def record_alert(self, alert_type: str, severity: str):
"""Record một alert"""
DIFY_ALERT_COUNT.labels(alert_type=alert_type, severity=severity).inc()
def update_active_sessions(self, count: int):
"""Update số lượng active sessions"""
ACTIVE_SESSIONS.set(count)
def run_simulation(self, duration_seconds=60):
"""Chạy simulation để generate metrics"""
models = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
print(f"Starting metrics collection for {duration_seconds} seconds...")
start_time = time.time()
while time.time() - start_time < duration_seconds:
# Random requests
for _ in range(random.randint(1, 10)):
model = random.choice(models)
tokens = random.randint(100, 2000)
self.simulate_request(model, tokens)
# Update sessions
self.update_active_sessions(random.randint(50, 200))
# Random alerts
if random.random() < 0.1:
alert_types = ["high_latency", "error_spike", "rate_limit_warning"]
severities = ["low", "medium", "high"]
self.record_alert(random.choice(alert_types), random.choice(severities))
time.sleep(5)
print("Metrics collection completed")
if __name__ == "__main__":
# Start Prometheus server on port 9090
start_http_server(9090)
print("Prometheus metrics server started on port 9090")
# Run metrics collector
collector = MetricsCollector()
collector.run_simulation(duration_seconds=60)Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi "Connection Timeout" Khi Gọi HolySheep API
Mô tả: Request đến https://api.holysheep.ai/v1 bị timeout sau 30 giây.
Nguyên nhân:
- Network firewall chặn port 443
- Proxy server không được cấu hình đúng
- SSL certificate không hợp lệ
Mã khắc phục:
#!/usr/bin/env python3
"""
Khắc phục lỗi Connection Timeout
"""
import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
def create_timeout_resilient_session():
"""Tạo session với retry logic và timeout handling"""
session = requests.Session()
# Cấu hình retry strategy
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
def call_holysheep_with_timeout_handling():
"""Gọi HolySheep API với timeout handling đúng cách"""
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
session = create_timeout_resilient_session()
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Hello"}],
"max_tokens": 100
}
try:
# Sử dụng timeout hợp lý
response = session.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=(10, 30) # (connect_timeout, read_timeout)
)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
print("❌ Request timed out - HolySheep API không phản hồi")
# Fallback: Thử lại sau hoặc sử dụng backup endpoint
return fallback_to_backup()
except requests.exceptions.ConnectionError as e:
print(f"❌ Connection error: {e}")
# Kiểm tra DNS resolution
import socket
try:
socket.gethostbyname("api.holysheep.ai")
print("✓ DNS resolution OK")
except socket.gaierror:
print("❌ DNS resolution failed - Kiểm tra network config")
except requests.exceptions.SSLError:
print("❌ SSL Error - Cập nhật certificates hoặc disable SSL verification tạm thời")
# Tạm thời disable SSL verification (chỉ dùng cho dev)
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
response = session.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30,
verify=False # CHỈ DÙNG CHO DEV
)
return response.json()
def fallback_to_backup():
"""Fallback strategy khi HolySheep API fail"""
print("Trying fallback strategy...")
# Thử với model khác
alternative_models = ["deepseek-v3.2", "gemini-2.5-flash"]
for model in alternative_models:
try:
response = requests.post(
f"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={
"model": model,
"messages": [{"role": "user", "content": "Hello"}],
"max_tokens": 100
},
timeout=15
)
if response.status_code == 200:
print(f"✓ Fallback thành công với model: {model}")
return response.json()
except:
continue
return {"error": "All fallback attempts failed"}
if __name__ == "__main__":
result = call_holysheep_with_timeout_handling()
print(f"Result: {result}")2. Lỗi "401 Unauthorized" - Invalid API Key
Mô tả: Nhận được response {"error": {"code": "invalid_api_key", "message": "Invalid API key"}}
Nguyên nhân:
- API key bị sai hoặc đã bị revoke
- Key bị copy thiếu ký tự
- Sử dụng key từ provider khác thay vì HolySheep
Mã khắc phục:
#!/usr/bin/env python3
"""
Khắc phục lỗi 401 Unauthorized
"""
import os
from dotenv import load_dotenv
def validate_api_key():
"""Validate HolySheep API key"""
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
# Load từ environment
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
# Validate format
if not api_key:
print("❌ HOLYSHEEP_API_KEY không được set")
print("Vui lòng đăng ký tại: https://www.holysheep.ai/register")
return False
# Check format (HolySheep key thường bắt đầu bằng "sk-" hoặc "hs-")
valid_prefixes = ["sk-holysheep-", "hs-", "sk-"]
if not any(api_key.startswith(prefix) for prefix in valid_prefixes):
print(f"⚠️ API key format có thể không đúng")
print(f"Key hiện tại: {api_key[:10]}...")
# Test API key bằng cách gọi /models endpoint
import requests
try:
response = requests.get(
f"{HOLYSHEEP_BASE_URL}/models",
headers={"Authorization": f"Bearer {api_key}"},
timeout=10
)
if response.status_code == 200:
print("✅ API key hợp lệ")
models = response.json().get("data", [])
print(f"Tìm thấy {len(models)} models khả dụng:")
for model in models[:5]:
print(f" - {model.get('id', 'unknown')}")
return True
elif response.status_code == 401:
print("❌ API key không hợp lệ hoặc đã bị revoke")
print("\nHướng dẫn:")
print("1. Truy cập https://www.holysheep.ai/register")
print("2. Tạo API key mới trong dashboard")
print("3. Copy key mới vào file .env")
return False
elif response.status_code == 429:
print("⚠️ Rate limit exceeded - đang thử refresh...")
return None
else:
print(f"❌ Lỗi không xác định: {response.status_code}")
print(f"Response: {response.text}")
return False
except Exception as e:
print(f"❌ Error validating API key: {e}")
return False
def rotate_api_key():
"""Tự động rotate API key khi cần"""
import requests
# Call HolySheep API để revoke old key và tạo key mới
# Lưu ý: Cần có quyền admin
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
OLD_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
try:
# Revoke old key
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/api-keys/revoke",
headers={
"Authorization": f"Bearer {OLD_API_KEY}",
"Content-Type": "application/json"
},
json={"key_id": "old_key_id"},
timeout=10
)
# Create new key
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/api-keys/create",
headers={
"Authorization": f"Bearer {OLD_API_KEY}",
"Content-Type": "application/json"
},
json={
"name": "production_key_v2",
"scopes": ["chat", "completions", "models"]
},
timeout=10
)
if response.status_code == 200:
new_key = response.json().get("key")
# Save to .env
with open(".env", "r") as f:
content = f.read()
content = content.replace(f"HOLYSHEEP_API_KEY={OLD_API_KEY}",
f"HOLYSHEEP_API_KEY={new_key}")
with open(".env", "w") as f:
f.write(content)
print("✅ API key đã được rotate thành công")
return new