import requests
import logging
from typing import Dict, Any, Optional
class BifrostClient:
def __init__(self, base_url: str):
self.base_url = base_url
self.logger = logging.getLogger(__name__)
def chat_completion(self, payload: Dict[Any, Any]) -> Optional[Dict[Any, Any]]:
try:
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()
# Handle Bifrost errors
error_data = response.json()
if error_data.get("is_bifrost_error"):
self.log_structured_error(error_data, payload)
error_type = error_data.get("error", {}).get("type")
error_message = error_data.get("error", {}).get("message", "Unknown error")
if error_type == "authentication_error":
raise Exception(f"Authentication failed: {error_message}")
elif error_type == "rate_limit_error":
raise Exception(f"Rate limited: {error_message}")
elif error_type == "network_error":
raise Exception(f"Network error: {error_message}")
else:
raise Exception(f"Bifrost error: {error_message}")
# Handle other HTTP errors
response.raise_for_status()
except requests.exceptions.RequestException as e:
self.logger.error(f"Request failed: {e}")
raise
def log_structured_error(self, error_data: Dict[Any, Any], payload: Dict[Any, Any]):
error_info = error_data.get("error", {})
self.logger.error(
"Bifrost error occurred",
extra={
"provider": payload.get("provider"),
"model": payload.get("model"),
"error_type": error_info.get("type"),
"error_code": error_info.get("code"),
"error_message": error_info.get("message"),
"event_id": error_data.get("event_id")
}
)
# Usage
client = BifrostClient("http://localhost:8080")
try:
response = client.chat_completion({
"provider": "openai",
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Hello!"}]
})
print("Success:", response)
except Exception as e:
print("Error:", e)