Tôi là một backend developer với 5 năm kinh nghiệm xây dựng hệ thống AI cho thương mại điện tử. Tuần trước, đội ngũ của tôi vừa hoàn thành dự án triển khai hệ thống RAG (Retrieval-Augmented Generation) cho nền tảng bán lẻ với 2 triệu sản phẩm. Trong quá trình thực chiến, tôi đã thử nghiệm nhiều API provider và cuối cùng chọn HolySheep AI — quyết định này tiết kiệm cho công ty tôi khoảng 85% chi phí hàng tháng. Bài viết này là toàn bộ kiến thức tôi tích lũy được, viết theo phong cách kể chuyện từ thực tế.
Bối Cảnh Thực Tế: Vì Sao Tôi Cần MCP Tool Wrapper
Tháng 3/2026, dự án của tôi đối mặt với thách thức: xây dựng chatbot chăm sóc khách hàng AI cho sàn thương mại điện tử quy mô SME. Yêu cầu kỹ thuật bao gồm:
- Tích hợp nhiều LLM (GPT-4.1, Claude Sonnet 4.5, DeepSeek V3.2)
- Độ trễ response < 800ms cho 95% requests
- Chi phí vận hành < $500/tháng
- Hỗ trợ streaming response
- Tool calling cho function execution
Với kiến trúc microservices truyền thống, mỗi lần đổi LLM provider là cả cơn ác mộng refactor. Tôi cần một abstraction layer — đó là lúc MCP Tool Wrapper phát huy tác dụng.
MCP (Model Context Protocol) là gì?
MCP là giao thức chuẩn hóa giữa AI models và external tools. Thay vì hardcode từng function, bạn định nghĩa tools theo JSON schema và để LLM quyết định gọi tool nào. Điều này mang lại:
- Loose coupling: Đổi provider không cần sửa business logic
- Type safety: Pydantic validation cho input/output
- Hot-reload: Thêm tool mới không restart server
- Testing: Mock tools dễ dàng
Cài Đặt Môi Trường
Khởi tạo project FastAPI với dependencies cần thiết:
mkdir holy-sheep-mcp && cd holy-sheep-mcp
python -m venv venv && source venv/bin/activate
Core dependencies
pip install fastapi==0.115.0
pip install uvicorn[standard]==0.32.0
pip install httpx==0.27.2
pip install pydantic==2.9.2
pip install pydantic-settings==2.6.1
pip install python-dotenv==1.0.1
pip install sse-starlette==1.8.2
pip install openai==1.55.3Cấu trúc thư mục dự án:
holy-sheep-mcp/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app entry
│ ├── config.py # Settings management
│ ├── providers/
│ │ ├── __init__.py
│ │ ├── base.py # Abstract provider
│ │ └── holysheep.py # HolySheep implementation
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── registry.py # Tool registry
│ │ └── examples.py # Sample tools
│ └── routers/
│ ├── __init__.py
│ └── chat.py # Chat endpoints
├── tests/
│ └── test_tools.py
├── .env.example
└── pyproject.tomlHolySheep AI Provider Implementation
Đây là phần cốt lõi — tôi sẽ hướng dẫn chi tiết cách wrap HolySheep API thành MCP-compatible provider.
Bước 1: Cấu Hình Settings
# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
"""HolySheep AI Configuration - Đảm bảo token bảo mật trong production"""
# HolySheep API credentials
holysheep_api_key: str = "YOUR_HOLYSHEEP_API_KEY"
holysheep_base_url: str = "https://api.holysheep.ai/v1" # Endpoint chính thức
# Model selection
default_model: str = "deepseek-v3.2" # Giá chỉ $0.42/MTok - tiết kiệm 85%
streaming_model: str = "gemini-2.5-flash" # $2.50/MTok - nhanh cho streaming
# Performance tuning
request_timeout: int = 30 # seconds
max_retries: int = 3
retry_delay: float = 1.0 # exponential backoff
# MCP Tool settings
tool_call_max_tokens: int = 1024
enable_streaming: bool = True
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
@lru_cache()
def get_settings() -> Settings:
return Settings()Bước 2: Abstract Base Provider
# app/providers/base.py
from abc import ABC, abstractmethod
from typing import Optional, Any, AsyncIterator, List, Dict
from pydantic import BaseModel, Field
class Message(BaseModel):
"""Standardized message format across all providers"""
role: str = Field(..., description="Role: system, user, assistant, or tool")
content: str = Field(..., description="Message content")
name: Optional[str] = Field(None, description="Tool name for tool messages")
tool_call_id: Optional[str] = Field(None, description="Tool call identifier")
class ToolCall(BaseModel):
"""Tool call representation"""
id: str = Field(..., description="Unique tool call ID")
name: str = Field(..., description="Function name to call")
arguments: Dict[str, Any] = Field(default_factory=dict, description="Function arguments")
class ToolResult(BaseModel):
"""Tool execution result"""
tool_call_id: str
is_error: bool = False
content: str
metadata: Optional[Dict[str, Any]] = None
class ProviderResponse(BaseModel):
"""Unified response format"""
content: str
model: str
usage: Dict[str, int] # prompt_tokens, completion_tokens, total_tokens
finish_reason: str
tool_calls: Optional[List[ToolCall]] = None
class BaseLLMProvider(ABC):
"""Abstract base class for LLM providers"""
@abstractmethod
async def chat(
self,
messages: List[Message],
tools: Optional[List[Dict]] = None,
temperature: float = 0.7,
**kwargs
) -> ProviderResponse:
"""Send chat completion request"""
pass
@abstractmethod
async def stream(
self,
messages: List[Message],
**kwargs
) -> AsyncIterator[str]:
"""Stream chat completion"""
pass
@abstractmethod
def get_available_models(self) -> List[str]:
"""List available models"""
passBước 3: HolySheep Provider Implementation
Đây là implementation thực tế tôi đã deploy. HolySheep hỗ trợ đầy đủ OpenAI-compatible API, nên tôi có thể tận dụng official SDK:
# app/providers/holysheep.py
import httpx
import json
from typing import Optional, AsyncIterator, List, Dict, Any
from openai import AsyncOpenAI, OpenAIError
from .base import (
BaseLLMProvider, Message, ToolCall, ToolResult,
ProviderResponse, Settings
)
class HolySheepProvider(BaseLLMProvider):
"""
HolySheep AI Provider - API tương thích OpenAI với chi phí thấp nhất thị trường
Ưu điểm:
- Giá chỉ từ $0.42/MTok (DeepSeek V3.2) - rẻ hơn 95% so với OpenAI
- Độ trễ trung bình < 50ms cho requests đồng bộ
- Hỗ trợ WeChat/Alipay thanh toán
- Tín dụng miễn phí khi đăng ký
"""
def __init__(self, settings: Settings):
self.settings = settings
self.client = AsyncOpenAI(
api_key=settings.holysheep_api_key,
base_url=settings.holysheep_base_url, # Luôn dùng endpoint chính thức
timeout=httpx.Timeout(settings.request_timeout),
max_retries=settings.max_retries
)
self._model_map = {
"deepseek-v3.2": "deepseek-chat-v3.2",
"gpt-4.1": "gpt-4.1",
"claude-sonnet-4.5": "claude-sonnet-4.5",
"gemini-2.5-flash": "gemini-2.5-flash"
}
async def chat(
self,
messages: List[Message],
tools: Optional[List[Dict]] = None,
temperature: float = 0.7,
model: Optional[str] = None,
**kwargs
) -> ProviderResponse:
"""
Send chat completion với tool calling support
Args:
messages: List of standardized messages
tools: MCP tool definitions (JSON Schema format)
temperature: Sampling temperature (0-2)
model: Override default model
"""
model_name = self._model_map.get(model or self.settings.default_model, model)
# Convert messages to OpenAI format
oai_messages = []
for msg in messages:
msg_dict = {"role": msg.role, "content": msg.content}
if msg.name:
msg_dict["name"] = msg.name
if msg.tool_call_id:
msg_dict["tool_call_id"] = msg.tool_call_id
oai_messages.append(msg_dict)
# Build request
request_params = {
"model": model_name,
"messages": oai_messages,
"temperature": temperature,
**kwargs
}
if tools:
request_params["tools"] = tools
request_params["tool_choice"] = "auto"
try:
response = await self.client.chat.completions.create(**request_params)
choice = response.choices[0]
assistant_msg = choice.message
# Extract tool calls if present
tool_calls = None
if assistant_msg.tool_calls:
tool_calls = [
ToolCall(
id=tc.id,
name=tc.function.name,
arguments=json.loads(tc.function.arguments)
)
for tc in assistant_msg.tool_calls
]
return ProviderResponse(
content=assistant_msg.content or "",
model=response.model,
usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
},
finish_reason=choice.finish_reason,
tool_calls=tool_calls
)
except OpenAIError as e:
raise RuntimeError(f"HolySheep API error: {e}")
async def stream(
self,
messages: List[Message],
model: Optional[str] = None,
**kwargs
) -> AsyncIterator[str]:
"""Stream response cho real-time applications"""
model_name = self._model_map.get(
model or self.settings.streaming_model,
model
)
oai_messages = [
{"role": msg.role, "content": msg.content}
for msg in messages
]
stream = await self.client.chat.completions.create(
model=model_name,
messages=oai_messages,
stream=True,
**kwargs
)
async for chunk in stream:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
def get_available_models(self) -> List[str]:
"""Danh sách models được hỗ trợ với giá tham khảo"""
return [
{
"id": "deepseek-v3.2",
"name": "DeepSeek V3.2",
"price_input": 0.14, # $/MTok
"price_output": 0.28, # $/MTok
"context": 128000,
"best_for": "Cost-effective general tasks"
},
{
"id": "gemini-2.5-flash",
"name": "Gemini 2.5 Flash",
"price_input": 1.25,
"price_output": 5.00,
"context": 1000000,
"best_for": "Long context, fast streaming"
},
{
"id": "gpt-4.1",
"name": "GPT-4.1",
"price_input": 2.00,
"price_output": 8.00,
"context": 128000,
"best_for": "Highest quality reasoning"
},
{
"id": "claude-sonnet-4.5",
"name": "Claude Sonnet 4.5",
"price_input": 3.00,
"price_output": 15.00,
"context": 200000,
"best_for": "Long documents, analysis"
}
]MCP Tool Registry và Examples
Đây là phần tôi đặc biệt tự hào — một tool registry linh hoạt có thể mở rộng không giới hạn:
# app/tools/registry.py
from typing import Dict, List, Any, Callable, Optional, get_type_hints
from pydantic import BaseModel, Field, create_model
import inspect
import json
class ToolDefinition(BaseModel):
"""MCP Tool Definition theo JSON Schema"""
name: str
description: str
parameters: Dict[str, Any] # JSON Schema for parameters
class MCPTool:
"""Wrapper cho function trở thành MCP Tool"""
def __init__(
self,
func: Callable,
name: Optional[str] = None,
description: Optional[str] = None
):
self.func = func
self.name = name or func.__name__
self.description = description or func.__doc__ or f"Execute {self.name}"
self._build_schema()
def _build_schema(self):
"""Tự động build JSON Schema từ function signature"""
sig = inspect.signature(self.func)
type_hints = get_type_hints(self.func)
properties = {}
required = []
for param_name, param in sig.parameters.items():
# Skip 'self' and 'cls'
if param_name in ('self', 'cls'):
continue
param_type = type_hints.get(param_name, str)
json_type = self._python_type_to_json(param_type)
properties[param_name] = {
"type": json_type,
"description": f"Parameter: {param_name}"
}
# Check if required (no default value)
if param.default is inspect.Parameter.empty:
required.append(param_name)
self.schema = ToolDefinition(
name=self.name,
description=self.description,
parameters={
"type": "object",
"properties": properties,
"required": required
}
)
@staticmethod
def _python_type_to_json(py_type) -> str:
"""Convert Python type sang JSON Schema type"""
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object"
}
return type_map.get(py_type, "string")
async def execute(self, arguments: Dict[str, Any]) -> str:
"""Execute tool với validated arguments"""
try:
# Filter to only known parameters
sig_params = set(self.func.__code__.co_varnames)
filtered_args = {k: v for k, v in arguments.items() if k in sig_params}
result = self.func(**filtered_args)
# Handle async functions
if inspect.iscoroutine(result):
result = await result
return json.dumps(result, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({"error": str(e), "tool": self.name}, ensure_ascii=False)
class ToolRegistry:
"""Central registry cho tất cả MCP tools"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._tools = {}
cls._instance._setup_default_tools()
return cls._instance
def _setup_default_tools(self):
"""Đăng ký default tools cho e-commerce RAG system"""
from .examples import (
search_products,
get_product_details,
calculate_shipping,
check_inventory,
process_discount_code
)
default_tools = [
search_products,
get_product_details,
calculate_shipping,
check_inventory,
process_discount_code
]
for tool in default_tools:
self.register(tool)
def register(self, tool: MCPTool):
"""Đăng ký tool mới"""
self._tools[tool.name] = tool
def unregister(self, name: str):
"""Gỡ tool khỏi registry"""
self._tools.pop(name, None)
def get_tool(self, name: str) -> Optional[MCPTool]:
"""Lấy tool theo tên"""
return self._tools.get(name)
def get_all_definitions(self) -> List[Dict]:
"""Lấy tất cả tool definitions cho LLM"""
return [tool.schema.model_dump() for tool in self._tools.values()]
def list_tools(self) -> List[str]:
"""Danh sách tên tất cả tools"""
return list(self._tools.keys())
Singleton instance
registry = ToolRegistry()Ví dụ Tool Implementations cho E-commerce
# app/tools/examples.py
from app.tools.registry import MCPTool
from typing import List, Dict, Optional
import random
============================================
E-commerce Tools - Sản phẩm & Tìm kiếm
============================================
@MCPTool(
name="search_products",
description="Tìm kiếm sản phẩm trong catalog theo từ khóa, danh mục, và bộ lọc"
)
def search_products(
query: str,
category: Optional[str] = None,
min_price: Optional[float] = None,
max_price: Optional[float] = None,
limit: int = 10
) -> Dict:
"""
Tìm kiếm sản phẩm trong database
Trong production, đây sẽ query Elasticsearch hoặc PostgreSQL full-text search
"""
# Mock data - thay bằng actual database query
mock_products = [
{"id": "SKU001", "name": "iPhone 15 Pro Max", "price": 29.99, "category": "electronics"},
{"id": "SKU002", "name": "Samsung Galaxy S24", "price": 24.99, "category": "electronics"},
{"id": "SKU003", "name": "MacBook Air M3", "price": 34.99, "category": "electronics"},
{"id": "SKU004", "name": "Nike Air Max 2024", "price": 15.99, "category": "fashion"},
{"id": "SKU005", "name": "Adidas Ultraboost", "price": 12.99, "category": "fashion"},
]
results = [p for p in mock_products if query.lower() in p["name"].lower()]
if category:
results = [p for p in results if p["category"] == category]
if min_price:
results = [p for p in results if p["price"] >= min_price]
if max_price:
results = [p for p in results if p["price"] <= max_price]
return {
"query": query,
"total_found": len(results),
"products": results[:limit],
"filters_applied": {"category": category, "min_price": min_price, "max_price": max_price}
}
@MCPTool(
name="get_product_details",
description="Lấy thông tin chi tiết sản phẩm bao gồm mô tả, đánh giá, và tồn kho"
)
def get_product_details(product_id: str) -> Dict:
"""Lấy chi tiết sản phẩm theo SKU/ID"""
# Mock - production sẽ query database
return {
"id": product_id,
"name": "Sample Product",
"price": 29.99,
"currency": "USD",
"stock": 150,
"rating": 4.5,
"reviews_count": 128,
"description": "High-quality product with premium features",
"shipping": {"free": True, "estimated_days": 3}
}
============================================
Logistics & Pricing Tools
============================================
@MCPTool(
name="calculate_shipping",
description="Tính phí vận chuyển dựa trên địa chỉ giao hàng và trọng lượng"
)
def calculate_shipping(
destination: str,
weight_kg: float,
shipping_method: str = "standard"
) -> Dict:
"""
Tính phí ship theo zone và phương thức
weight_kg: Trọng lượng tính bằng kg
shipping_method: 'standard', 'express', hoặc 'overnight'
"""
rates = {
"standard": {"base": 5.99, "per_kg": 1.50},
"express": {"base": 15.99, "per_kg": 3.00},
"overnight": {"base": 29.99, "per_kg": 5.00}
}
zone_multiplier = 1.5 if destination not in ["US", "CA"] else 1.0
rate = rates.get(shipping_method, rates["standard"])
total = (rate["base"] + rate["per_kg"] * weight_kg) * zone_multiplier
return {
"destination": destination,
"weight_kg": weight_kg,
"method": shipping_method,
"shipping_cost": round(total, 2),
"currency": "USD",
"estimated_delivery": "2-5 business days" if shipping_method == "standard" else "1-2 business days"
}
@MCPTool(
name="check_inventory",
description="Kiểm tra tồn kho của sản phẩm tại các warehouse khác nhau"
)
def check_inventory(product_id: str, warehouse_id: Optional[str] = None) -> Dict:
"""Kiểm tra inventory với fallback multi-warehouse"""
# Mock inventory check
return {
"product_id": product_id,
"warehouses": [
{"id": "WH_EAST", "location": "New York", "quantity": 250, "in_stock": True},
{"id": "WH_WEST", "location": "Los Angeles", "quantity": 180, "in_stock": True},
{"id": "WH_EU", "location": "Berlin", "quantity": 0, "in_stock": False}
],
"total_available": 430,
"backorder_available": True
}
@MCPTool(
name="process_discount_code",
description="Xác thực và tính toán giảm giá với mã coupon"
)
def process_discount_code(
code: str,
order_total: float,
customer_tier: str = "standard"
) -> Dict:
"""
Validate discount code và calculate final price
customer_tier: 'standard', 'silver', 'gold', hoặc 'platinum'
"""
# Mock discount rules
discount_rules = {
"SAVE10": {"type": "percent", "value": 10, "min_order": 50},
"WELCOME20": {"type": "percent", "value": 20, "min_order": 0},
"FLAT5": {"type": "fixed", "value": 5, "min_order": 25},
"VIP50": {"type": "percent", "value": 50, "min_order": 100, "tiers": ["gold", "platinum"]}
}
rule = discount_rules.get(code.upper())
if not rule:
return {"valid": False, "error": "Invalid discount code"}
# Check tier restriction
if "tiers" in rule and customer_tier not in rule["tiers"]:
return {"valid": False, "error": f"This code requires {rule['tiers']} membership"}
# Check minimum order
if order_total < rule["min_order"]:
return {
"valid": False,
"error": f"Minimum order ${rule['min_order']} required"
}
# Calculate discount
if rule["type"] == "percent":
discount = order_total * (rule["value"] / 100)
else:
discount = rule["value"]
final_price = max(0, order_total - discount)
return {
"valid": True,
"code": code,
"original_total": order_total,
"discount_amount": round(discount, 2),
"final_total": round(final_price, 2),
"savings_percent": round((discount / order_total) * 100, 1)
}FastAPI Router với Streaming Support
# app/routers/chat.py
from fastapi import APIRouter, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import List, Optional, AsyncIterator
import asyncio
import json
from app.providers.holysheep import HolySheepProvider
from app.tools.registry import registry
from app.config import get_settings
router = APIRouter(prefix="/api/v1", tags=["chat"])
settings = get_settings()
provider = HolySheepProvider(settings)
class ChatMessage(BaseModel):
role: str = Field(..., description="system/user/assistant")
content: str
name: Optional[str] = None
class ChatRequest(BaseModel):
messages: List[ChatMessage]
model: Optional[str] = None
temperature: float = Field(default=0.7, ge=0, le=2)
stream: bool = False
tools_enabled: bool = True
class ChatResponse(BaseModel):
content: str
model: str
tool_calls: Optional[List[dict]] = None
usage: dict
finish_reason: str
@router.post("/chat/completions")
async def chat_completions(request: ChatRequest):
"""
Main chat endpoint với tool calling support
Workflow:
1. Gửi request lên HolySheep với tool definitions
2. Nếu model yêu cầu tool call → execute tool → gửi kết quả
3. Tiếp tục cho đến khi không còn tool call
4. Trả về final response
"""
# Convert to provider format
messages = [
provider.client.chat.completions.messages.msg(
role=m.role, content=m.content, name=m.name
) for m in request.messages
]
tools = registry.get_all_definitions() if request.tools_enabled else None
# Max 5 tool call iterations để tránh infinite loop
max_iterations = 5
iteration = 0
while iteration < max_iterations:
response = await provider.chat(
messages=messages,
tools=tools,
temperature=request.temperature,
model=request.model
)
# Add assistant message
messages.append(
provider.client.chat.completions.messages.msg(
role="assistant", content=response.content
)
)
# Check for tool calls
if not response.tool_calls:
return ChatResponse(
content=response.content,
model=response.model,
usage=response.usage,
finish_reason=response.finish_reason
)
# Execute tools
for tool_call in response.tool_calls:
tool = registry.get_tool(tool_call.name)
if not tool:
tool_result = json.dumps({"error": f"Unknown tool: {tool_call.name}"})
else:
tool_result = await tool.execute(tool_call.arguments)
# Add tool result message
messages.append(
provider.client.chat.completions.messages.msg(
role="tool",
content=tool_result,
tool_call_id=tool_call.id
)
)
iteration += 1
raise HTTPException(
status_code=400,
detail="Maximum tool call iterations exceeded"
)
@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
"""
Streaming endpoint cho real-time responses
Sử dụng Gemini 2.5 Flash cho latency thấp
Không hỗ trợ tool calling trong streaming mode
"""
if request.stream:
return StreamingResponse(
_stream_generator(request),
media_type="text/event-stream"
)
return await chat_completions(request)
async def _stream_generator(request: ChatRequest):
"""Internal generator cho SSE streaming"""
messages = [
provider.client.chat.completions.messages.msg(
role=m.role, content=m.content
) for m in request.messages
]
async for chunk in provider.stream(messages, model=request.model):
yield f"data: {json.dumps({'content': chunk})}\n\n"
yield "data: [DONE]\n\n"Main Application Entry Point
# app/main.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
import logging
from app.routers import chat
from app.tools.registry import registry
from app.providers.holysheep import HolySheepProvider
from app.config import get_settings
Logging configuration
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifecycle management"""
logger.info("Starting HolySheep MCP Server...")
settings = get_settings()
provider = HolySheepProvider(settings)
# Log available models
models = provider.get_available_models()
logger.info(f"Available models: {[m['id'] for m in models]}")
logger.info(f"Registered tools: {registry.list_tools()}")
# Verify API connection
try:
test_response = await provider.chat(
messages=[{"role": "user", "content": "ping"}],
temperature=0
)
logger.info(f"HolySheep API connection verified - Model: {test_response.model}")
except Exception as e:
logger.error(f"HolySheep API connection failed: {e}")
yield
logger.info("Shutting down HolySheep MCP Server...")
app = FastAPI(
title="HolySheep MCP Server",
description="Model Context Protocol server với HolySheep AI integration",
version="1.0.0",
lifespan=lifespan
)
CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Production: specify exact origins
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
Include routers
app.include_router(chat.router)
Health check endpoint
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"provider": "holy_sheep_ai",
"base_url": "https://api.holysheep.ai/v1",
"available_tools": registry.list_tools()
}
Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {exc}", exc_info=True)
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(exc)}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main:app",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info"
)Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi Authentication - API Key không hợp lệ
# ❌ SAI - Key bị