Là một kỹ sư backend đã triển khai hệ thống AI gateway cho 3 dự án enterprise quy mô lớn, tôi nhận ra một vấn đề nan giải: khi doanh nghiệp cần sử dụng đồng thời GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash và DeepSeek V3.2, việc quản lý riêng lẻ từng API không chỉ tốn kém mà còn phức tạp về mặt vận hành. Bài viết này là kinh nghiệm thực chiến của tôi khi xây dựng hệ thống MCP (Model Context Protocol) gateway tập trung, giúp tiết kiệm 85%+ chi phí và giảm 70% độ phức tạp code.

Thực Trạng Chi Phí AI API 2026 — Dữ Liệu Đã Xác Minh

Trước khi đi vào technical implementation, hãy cùng tôi phân tích chi phí thực tế khi vận hành multi-model system. Đây là bảng giá output token tôi đã xác minh trực tiếp từ các nhà cung cấp (cập nhật tháng 4/2026):

So Sánh Chi Phí Cho 10 Triệu Token/Tháng

ModelGiá/MTok10M TokensVới HolySheep (85% tiết kiệm)
GPT-4.1$8.00$80.00$12.00
Claude Sonnet 4.5$15.00$150.00$22.50
Gemini 2.5 Flash$2.50$25.00$3.75
DeepSeek V3.2$0.42$4.20$0.63

Con số này cho thấy rằng nếu team của bạn sử dụng đều 4 model với 2.5M tokens/model/tháng, chi phí gốc là $259.20/tháng. Với HolySheep AI và tỷ giá ¥1=$1, con số này chỉ còn $38.88/tháng — tiết kiệm được $220 mỗi tháng, tương đương $2,640/năm.

MCP Protocol Là Gì và Tại Sao Cần Aggregation Gateway

MCP (Model Context Protocol) là protocol chuẩn công nghiệp cho phép AI models giao tiếp với external tools một cách có cấu trúc. Thay vì mỗi model có cách gọi tools riêng, MCP tạo ra một abstraction layer thống nhất. Khi tôi triển khai hệ thống cho dự án thứ 2, việc không có gateway tập trung dẫn đến:

Architecture Design — Từ Zero Đến Production

High-Level Architecture


┌─────────────────────────────────────────────────────────────────┐
│                        Client Applications                       │
│                    (Web App / Mobile / CLI Tool)                │
└─────────────────────────────────────────────────────────────────┘
                                 │
                                 ▼
┌─────────────────────────────────────────────────────────────────┐
│                    MCP Aggregation Gateway                       │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │   Router     │  │  Load        │  │   Cost       │          │
│  │   Layer      │──│  Balancer    │──│  Optimizer   │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
│         │                 │                  │                  │
│  ┌──────────────────────────────────────────────────┐          │
│  │              Tool Registry (MCP Registry)         │          │
│  │  - file_system: GPT-4.1, Claude                   │          │
│  │  - web_search: Gemini 2.5 Flash, DeepSeek         │          │
│  │  - code_executor: Claude Sonnet 4.5               │          │
│  └──────────────────────────────────────────────────┘          │
└─────────────────────────────────────────────────────────────────┘
         │                 │                  │
         ▼                 ▼                  ▼
┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│ HolySheep   │  │ HolySheep   │  │ HolySheep    │  │ HolySheep   │
│ GPT-4.1     │  │ Claude S.45 │  │ Gemini 2.5   │  │ DeepSeek V3 │
│ $8/MTok     │  │ $15/MTok    │  │ $2.50/MTok   │  │ $0.42/MTok  │
└─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘
```


Điểm mấu chốt: Tất cả requests đều đi qua một endpoint duy nhất của HolySheep, nơi tôi có thể route đến bất kỳ model nào với cùng một API key và cùng một authentication flow.



Implementation — Code Thực Chiến



1. MCP Gateway Server Core



// mcp_gateway.py - Core Gateway Implementation
// Base URL: https://api.holysheep.ai/v1 (KHÔNG dùng api.openai.com)

import asyncio
import hashlib
import time
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
import httpx
from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.responses import StreamingResponse
import json


============================================================================


CẤU HÌNH HOLYSHEEP - ĐÂY LÀ API GATEWAY DUY NHẤT


============================================================================

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Thay bằng key thực tế

class ModelType(Enum):
    GPT_4_1 = "gpt-4.1"
    CLAUDE_SONNET_45 = "claude-sonnet-4.5"
    GEMINI_FLASH_25 = "gemini-2.5-flash"
    DEEPSEEK_V32 = "deepseek-v3.2"

@dataclass
class ModelPricing:
    name: str
    input_cost: float  # $/MTok
    output_cost: float  # $/MTok
    latency_p50: float  # ms
    best_for: List[str]

MODEL_CATALOG: Dict[str, ModelPricing] = {
    "gpt-4.1": ModelPricing(
        name="GPT-4.1",
        input_cost=2.00,
        output_cost=8.00,
        latency_p50=850,
        best_for=["complex_reasoning", "long_context", "code_generation"]
    ),
    "claude-sonnet-4.5": ModelPricing(
        name="Claude Sonnet 4.5",
        input_cost=3.50,
        output_cost=15.00,
        latency_p50=920,
        best_for=["coding", "analysis", "creative_writing"]
    ),
    "gemini-2.5-flash": ModelPricing(
        name="Gemini 2.5 Flash",
        input_cost=0.35,
        output_cost=2.50,
        latency_p50=320,
        best_for=["fast_responses", "high_volume", "summarization"]
    ),
    "deepseek-v3.2": ModelPricing(
        name="DeepSeek V3.2",
        input_cost=0.14,
        output_cost=0.42,
        latency_p50=580,
        best_for=["cost_optimization", "reasoning", "multilingual"]
    ),
}

@dataclass
class MCPTool:
    name: str
    description: str
    input_schema: Dict[str, Any]
    recommended_models: List[str]
    priority_routing: bool = False

class MCPAggregationGateway:
    """
    Gateway tập trung quản lý multi-model MCP tool calls.
    Đặc điểm:
    - Single endpoint cho tất cả models
    - Automatic cost optimization
    - Unified error handling
    - Request queuing với priority
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = HOLYSHEEP_BASE_URL
        self.tools_registry: Dict[str, MCPTool] = {}
        self.usage_stats: Dict[str, Dict] = {}
        
    def register_tool(self, tool: MCPTool):
        """Đăng ký tool vào registry"""
        self.tools_registry[tool.name] = tool
        print(f"[MCP Gateway] Registered tool: {tool.name}")
        
    async def route_request(
        self,
        messages: List[Dict],
        tools: Optional[List[str]] = None,
        model: Optional[str] = None,
        auto_optimize: bool = True,
        budget_limit: Optional[float] = None
    ) -> Dict[str, Any]:
        """
        Route request đến model phù hợp nhất.
        Nếu auto_optimize=True, sẽ tự động chọn model rẻ nhất phù hợp với task.
        """
        
        # Bước 1: Xác định model tối ưu
        target_model = model
        if auto_optimize and not model:
            target_model = self._optimize_model_selection(messages, tools, budget_limit)
        
        # Bước 2: Build request payload
        payload = {
            "model": target_model,
            "messages": messages,
            "temperature": 0.7,
            "max_tokens": 4096
        }
        
        if tools:
            mcp_tools = [self._build_mcp_tool(tool_name) for tool_name in tools]
            payload["tools"] = mcp_tools
        
        # Bước 3: Gửi request đến HolySheep
        start_time = time.time()
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json=payload
            )
        
        elapsed_ms = (time.time() - start_time) * 1000
        
        if response.status_code != 200:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"Gateway Error: {response.text}"
            )
        
        result = response.json()
        
        # Bước 4: Update usage stats
        self._update_usage_stats(target_model, result, elapsed_ms)
        
        return {
            "model": target_model,
            "response": result,
            "latency_ms": round(elapsed_ms, 2),
            "estimated_cost": self._calculate_cost(target_model, result)
        }
    
    def _optimize_model_selection(
        self,
        messages: List[Dict],
        tools: Optional[List[str]],
        budget_limit: Optional[float]
    ) -> str:
        """
        Auto-select model based on task complexity và budget.
        Đây là logic tối ưu chi phí - kinh nghiệm thực chiến của tôi.
        """
        
        # Phân tích nội dung message
        content = " ".join([m.get("content", "") for m in messages])
        content_length = len(content)
        
        # Xác định task type từ content analysis
        task_keywords = {
            "coding": ["code", "function", "class", "debug", "implement", "algorithm"],
            "reasoning": ["analyze", "think", "reason", "logic", "solve", "compare"],
            "creative": ["write", "story", "creative", "poem", "narrative"],
            "fast": ["quick", "summary", "brief", "simple", "what is"]
        }
        
        detected_task = "general"
        for task, keywords in task_keywords.items():
            if any(kw.lower() in content.lower() for kw in keywords):
                detected_task = task
                break
        
        # Routing logic dựa trên task và budget
        if tools:
            # Nếu có tools, cần model hỗ trợ function calling tốt
            if budget_limit and budget_limit < 0.50:
                return "deepseek-v3.2"
            return "claude-sonnet-4.5"  # Claude hỗ trợ tools tốt nhất
            
        if detected_task == "coding":
            return "claude-sonnet-4.5"
        elif detected_task == "reasoning":
            if content_length > 5000:
                return "gpt-4.1"  # Long context reasoning
            return "deepseek-v3.2"  # Tiết kiệm cho reasoning ngắn
        elif detected_task == "creative":
            return "gpt-4.1"
        elif detected_task == "fast":
            return "gemini-2.5-flash"  # Nhanh nhất, rẻ nhất
        
        # Default: cân bằng giữa cost và quality
        return "gemini-2.5-flash"
    
    def _build_mcp_tool(self, tool_name: str) -> Dict:
        """Build MCP tool schema theo chuẩn protocol"""
        tool_schemas = {
            "file_system": {
                "type": "function",
                "function": {
                    "name": "file_system",
                    "description": "Read, write, or list files in the filesystem",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "operation": {
                                "type": "string",
                                "enum": ["read", "write", "list", "delete"]
                            },
                            "path": {"type": "string"},
                            "content": {"type": "string"}
                        }
                    }
                }
            },
            "web_search": {
                "type": "function",
                "function": {
                    "name": "web_search",
                    "description": "Search the web for current information",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string"},
                            "num_results": {"type": "integer", "default": 5}
                        }
                    }
                }
            },
            "code_executor": {
                "type": "function",
                "function": {
                    "name": "code_executor",
                    "description": "Execute code in sandbox environment",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "language": {"type": "string"},
                            "code": {"type": "string"},
                            "timeout": {"type": "integer", "default": 30}
                        }
                    }
                }
            }
        }
        return tool_schemas.get(tool_name, {})
    
    def _update_usage_stats(self, model: str, response: Dict, latency: float):
        """Track usage để optimize future requests"""
        if model not in self.usage_stats:
            self.usage_stats[model] = {
                "total_requests": 0,
                "total_tokens": 0,
                "total_cost": 0.0,
                "avg_latency": 0.0
            }
        
        stats = self.usage_stats[model]
        stats["total_requests"] += 1
        
        # Estimate tokens từ response
        usage = response.get("usage", {})
        tokens = usage.get("total_tokens", 0)
        stats["total_tokens"] += tokens
        
        # Tính cost
        model_info = MODEL_CATALOG.get(model)
        if model_info:
            input_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * model_info.input_cost
            output_cost = (usage.get("completion_tokens", 0) / 1_000_000) * model_info.output_cost
            stats["total_cost"] += (input_cost + output_cost)
        
        # Update latency
        stats["avg_latency"] = (
            (stats["avg_latency"] * (stats["total_requests"] - 1) + latency) 
            / stats["total_requests"]
        )
    
    def _calculate_cost(self, model: str, response: Dict) -> Dict[str, float]:
        """Tính chi phí thực tế của request"""
        model_info = MODEL_CATALOG.get(model)
        if not model_info:
            return {"total": 0.0}
        
        usage = response.get("usage", {})
        input_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * model_info.input_cost
        output_cost = (usage.get("completion_tokens", 0) / 1_000_000) * model_info.output_cost
        
        return {
            "input_cost": round(input_cost, 6),
            "output_cost": round(output_cost, 6),
            "total": round(input_cost + output_cost, 6)
        }
    
    def get_cost_report(self) -> Dict[str, Any]:
        """Generate báo cáo chi phí chi tiết"""
        total_cost = sum(s["total_cost"] for s in self.usage_stats.values())
        total_tokens = sum(s["total_tokens"] for s in self.usage_stats.values())
        
        return {
            "summary": {
                "total_spend": round(total_cost, 4),
                "total_tokens": total_tokens,
                "cost_per_1m_tokens": round(
                    (total_cost / total_tokens * 1_000_000) if total_tokens > 0 else 0, 
                    2
                )
            },
            "by_model": {
                model: {
                    "requests": stats["total_requests"],
                    "tokens": stats["total_tokens"],
                    "cost": round(stats["total_cost"], 4),
                    "avg_latency_ms": round(stats["avg_latency"], 2)
                }
                for model, stats in self.usage_stats.items()
            }
        }


============================================================================


KHỞI TẠO GATEWAY


============================================================================

gateway = MCPAggregationGateway(api_key=HOLYSHEEP_API_KEY)


Đăng ký default MCP tools

gateway.register_tool(MCPTool(
    name="file_system",
    description="File system operations",
    input_schema={},
    recommended_models=["gpt-4.1", "claude-sonnet-4.5"]
))

gateway.register_tool(MCPTool(
    name="web_search",
    description="Web search capability",
    input_schema={},
    recommended_models=["gemini-2.5-flash", "deepseek-v3.2"]
))

gateway.register_tool(MCPTool(
    name="code_executor",
    description="Execute code in sandbox",
    input_schema={},
    recommended_models=["claude-sonnet-4.5"],
    priority_routing=True
))




2. FastAPI Application Với Streaming Support



// main.py - FastAPI Application Entry Point
// Sử dụng HolySheep API: https://api.holysheep.ai/v1

from fastapi import FastAPI, HTTPException, Header, Body
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
import asyncio
import uvicorn

from mcp_gateway import gateway, ModelType, HOLYSHEEP_BASE_URL

app = FastAPI(
    title="MCP Aggregation Gateway",
    description="Unified Multi-Model MCP Gateway với Cost Optimization",
    version="1.0.0"
)


CORS Configuration

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


============================================================================


REQUEST/RESPONSE MODELS


============================================================================


class Message(BaseModel):
    role: str = Field(..., description="Role: system, user, hoặc assistant")
    content: str = Field(..., description="Nội dung message")

class MCPRequest(BaseModel):
    messages: List[Message]
    model: Optional[str] = Field(None, description="Model cụ thể, hoặc None để auto-optimize")
    tools: Optional[List[str]] = Field(None, description="Danh sách MCP tools cần sử dụng")
    auto_optimize: bool = Field(True, description="Tự động tối ưu model theo task")
    budget_limit: Optional[float] = Field(None, description="Giới hạn budget cho request này")
    temperature: float = Field(0.7, ge=0, le=2.0)
    max_tokens: int = Field(4096, ge=1, le=128000)
    stream: bool = Field(False, description="Enable streaming response")

class CostReport(BaseModel):
    summary: Dict[str, Any]
    by_model: Dict[str, Any]


============================================================================


API ENDPOINTS


============================================================================


@app.post("/v1/chat/completions", tags=["Chat Completions"])
async def chat_completions(request: MCPRequest):
    """
    Endpoint chính cho chat completions.
    Tự động route đến model tối ưu nếu không chỉ định model.
    """
    try:
        messages_dict = [msg.dict() for msg in request.messages]
        
        result = await gateway.route_request(
            messages=messages_dict,
            tools=request.tools,
            model=request.model,
            auto_optimize=request.auto_optimize,
            budget_limit=request.budget_limit
        )
        
        return result
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/v1/chat/completions/stream", tags=["Chat Completions"])
async def chat_completions_stream(
    request: MCPRequest = Body(...)
):
    """
    Streaming endpoint cho real-time responses.
    Sử dụng SSE (Server-Sent Events) protocol.
    """
    import httpx
    import json
    
    async def event_generator():
        messages_dict = [msg.dict() for msg in request.messages]
        
        payload = {
            "model": request.model or "gemini-2.5-flash",
            "messages": messages_dict,
            "temperature": request.temperature,
            "max_tokens": request.max_tokens,
            "stream": True
        }
        
        async with httpx.AsyncClient(timeout=120.0) as client:
            async with client.stream(
                "POST",
                f"{HOLYSHEEP_BASE_URL}/chat/completions",
                headers={
                    "Authorization": f"Bearer {gateway.api_key}",
                    "Content-Type": "application/json"
                },
                json=payload
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        if line.strip() == "data: [DONE]":
                            yield "data: [DONE]\n\n"
                        else:
                            yield f"{line}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"
        }
    )

@app.get("/v1/models", tags=["Models"])
async def list_models():
    """
    Liệt kê tất cả models có sẵn với pricing info.
    """
    from mcp_gateway import MODEL_CATALOG
    
    return {
        "object": "list",
        "data": [
            {
                "id": model_id,
                "object": "model",
                "created": 1700000000,
                "owned_by": "holysheep",
                "input_cost_per_mtok": info.input_cost,
                "output_cost_per_mtok": info.output_cost,
                "best_for": info.best_for,
                "latency_p50_ms": info.latency_p50
            }
            for model_id, info in MODEL_CATALOG.items()
        ]
    }

@app.get("/v1/costs/report", tags=["Analytics"])
async def get_cost_report():
    """Lấy báo cáo chi phí chi tiết"""
    return gateway.get_cost_report()

@app.get("/v1/tools", tags=["MCP Tools"])
async def list_tools():
    """Liệt kê tất cả registered MCP tools"""
    return {
        "object": "list",
        "data": [
            {
                "name": name,
                "description": tool.description,
                "recommended_models": tool.recommended_models,
                "priority_routing": tool.priority_routing
            }
            for name, tool in gateway.tools_registry.items()
        ]
    }

@app.get("/health", tags=["System"])
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "gateway": "MCP Aggregation Gateway",
        "version": "1.0.0",
        "registered_tools": len(gateway.tools_registry),
        "holy_sheep_endpoint": HOLYSHEEP_BASE_URL
    }

@app.get("/v1/estimate-cost", tags=["Utilities"])
async def estimate_cost(
    model: str,
    estimated_input_tokens: int,
    estimated_output_tokens: int
):
    """Ước tính chi phí trước khi gửi request"""
    from mcp_gateway import MODEL_CATALOG
    
    model_info = MODEL_CATALOG.get(model)
    if not model_info:
        raise HTTPException(status_code=400, detail=f"Unknown model: {model}")
    
    input_cost = (estimated_input_tokens / 1_000_000) * model_info.input_cost
    output_cost = (estimated_output_tokens / 1_000_000) * model_info.output_cost
    
    return {
        "model": model,
        "estimated_input_cost": round(input_cost, 6),
        "estimated_output_cost": round(output_cost, 6),
        "estimated_total": round(input_cost + output_cost, 6),
        "currency": "USD"
    }


============================================================================


STARTUP


============================================================================


@app.on_event("startup")
async def startup_event():
    print(f"""
╔══════════════════════════════════════════════════════════════╗
║           MCP Aggregation Gateway Started                     ║
║  HolySheep Endpoint: {HOLYSHEEP_BASE_URL}                ║
║  Models Available: 4 (GPT-4.1, Claude S.45, Gemini, DeepSeek)║
║  MCP Tools Registered: {len(gateway.tools_registry):<3}                                  ║
╚══════════════════════════════════════════════════════════════╝
    """)

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )




3. Python Client Với Retry Logic Và Circuit Breaker



// mcp_client.py - Python Client Với Advanced Features
// Kết nối đến HolySheep Gateway

import asyncio
import time
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass
from enum import Enum
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    recovery_timeout: float = 30.0
    half_open_max_calls: int = 3

class CircuitBreaker:
    """Circuit Breaker pattern implementation để handle API failures"""
    
    def __init__(self, config: CircuitBreakerConfig):
        self.config = config
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None
        self.half_open_calls = 0
    
    def record_success(self):
        self.failure_count = 0
        self.success_count += 1
        if self.state == CircuitState.HALF_OPEN:
            if self.success_count >= self.config.half_open_max_calls:
                self.state = CircuitState.CLOSED
                self.success_count = 0
    
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
        elif self.failure_count >= self.config.failure_threshold:
            self.state = CircuitState.OPEN
    
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.config.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                return True
            return False
        
        if self.state == CircuitState.HALF_OPEN:
            return self.half_open_calls < self.config.half_open_max_calls
        
        return False

class MCPClient:
    """
    Production-ready MCP Client với:
    - Automatic retry với exponential backoff
    - Circuit breaker pattern
    - Request/Response logging
    - Cost tracking
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "http://localhost:8000",
        timeout: float = 60.0
    ):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.circuit_breaker = CircuitBreaker(CircuitBreakerConfig())
        self.request_count = 0
        self.total_cost = 0.0
        
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    async def chat(
        self,
        messages: List[Dict[str, str]],
        model: Optional[str] = None,
        tools: Optional[List[str]] = None,
        auto_optimize: bool = True,
        temperature: float = 0.7,
        max_tokens: int = 4096
    ) -> Dict[str, Any]:
        """
        Send chat request với automatic retry và circuit breaker.
        """
        
        if not self.circuit_breaker.can_execute():
            raise Exception("Circuit breaker is OPEN. Service temporarily unavailable.")
        
        payload = {
            "messages": messages,
            "model": model,
            "tools": tools,
            "auto_optimize": auto_optimize,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": False
        }
        
        # Remove None values
        payload = {k: v for k, v in payload.items() if v is not None}
        
        start_time = time.time()
        
        try:
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                response = await client.post(
                    f"{self.base_url}/v1/chat/completions",
                    json=payload,
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json"
                    }
                )
                
                response.raise_for_status()
                result = response.json()
                
                # Record success
                self.circuit_breaker.record_success()
                
                # Update cost tracking
                cost = result.get("estimated_cost", {}).get("total", 0)
                self.total_cost += cost
                self.request_count += 1
                
                return result
                
        except httpx.HTTPStatusError as e:
            self.circuit_breaker.record_failure()
            raise Exception(f"HTTP Error {e.response.status_code}: {e.response.text}")
            
        except Exception as e:
            self.circuit_breaker.record_failure()
            raise
    
    async def stream_chat(
        self,

Tài nguyên liên quan
Bài viết liên quan