from fastapi import FastAPI, Request, HTTPException import uvicorn import logging import json from pydantic import BaseModel, Field, field_validator, model_validator from typing import List, Dict, Any, Optional, Union, Literal import httpx import os from fastapi.responses import JSONResponse, StreamingResponse import litellm import uuid import time from dotenv import load_dotenv import re from datetime import datetime import sys # Load environment variables from .env file load_dotenv() # Configure logging logging.basicConfig( level=logging.INFO, # Changed from DEBUG to INFO format='%(asctime)s - %(levelname)s - %(message)s', ) logger = logging.getLogger(__name__) # Configure uvicorn to be quieter import uvicorn # Tell uvicorn's loggers to be quiet logging.getLogger("uvicorn").setLevel(logging.WARNING) logging.getLogger("uvicorn.access").setLevel(logging.WARNING) logging.getLogger("uvicorn.error").setLevel(logging.WARNING) # Create a filter to block any log messages containing specific strings class MessageFilter(logging.Filter): def filter(self, record): # Block messages containing these strings blocked_phrases = [ "LiteLLM completion()", "HTTP Request:", "selected model name for cost calculation", "utils.py", "cost_calculator" ] if hasattr(record, 'msg') and isinstance(record.msg, str): for phrase in blocked_phrases: if phrase in record.msg: return False return True # Apply the filter to the root logger to catch all messages root_logger = logging.getLogger() root_logger.addFilter(MessageFilter()) # Custom formatter for model mapping logs class ColorizedFormatter(logging.Formatter): """Custom formatter to highlight model mappings""" BLUE = "\033[94m" GREEN = "\033[92m" YELLOW = "\033[93m" RED = "\033[91m" RESET = "\033[0m" BOLD = "\033[1m" def format(self, record): return super().format(record) # Apply custom formatter to console handler for handler in logger.handlers: if isinstance(handler, logging.StreamHandler): handler.setFormatter(ColorizedFormatter('%(asctime)s - %(levelname)s - %(message)s')) app = FastAPI() # Get API keys from environment ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY") OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY") # auth for client request AUTH_API_KEY = os.environ.get("AUTH_API_KEY") # Get preferred provider (default to openai) PREFERRED_PROVIDER = os.environ.get("PREFERRED_PROVIDER", "openai").lower() # Get custom OpenAI API base URL for LiteLLM proxy support OPENAI_API_BASE = os.environ.get("OPENAI_API_BASE", None) # Get model mapping configuration from environment # Default to latest OpenAI models if not set BIG_MODEL = os.environ.get("BIG_MODEL", "gpt-4.1") SMALL_MODEL = os.environ.get("SMALL_MODEL", "gpt-4.1-mini") # List of OpenAI models OPENAI_MODELS = [ "o3-mini", "o1", "o1-mini", "o1-pro", "gpt-4.5-preview", "gpt-4o", "gpt-4o-audio-preview", "chatgpt-4o-latest", "gpt-4o-mini", "gpt-4o-mini-audio-preview", "gpt-4.1", # Added default big model "gpt-4.1-mini" # Added default small model ] # List of Gemini models GEMINI_MODELS = [ "gemini-2.5-pro-preview-03-25", "gemini-2.0-flash", "gemini-2.5-flash-preview-05-20", # Added user's models "gemini-2.5-pro-preview-05-06" ] # List of Anthropic models (for LiteLLM anthropic/ prefix) ANTHROPIC_MODELS = [ "claude-3-5-sonnet-20241022", "claude-3-5-haiku-20241022", "claude-3-5-sonnet-20240620", "claude-3-5-haiku-20240307", "claude-3-opus-20240229", "claude-sonnet-4-20250514", # User's models "claude-3-5-haiku-latest", "claude-3-haiku-20240307", "claude-3-sonnet-20240229", "claude-2.1", "claude-2.0", "claude-instant-1.2" ] # Helper function to clean schema for Gemini def clean_gemini_schema(schema: Any) -> Any: """Recursively removes unsupported fields from a JSON schema for Gemini.""" if isinstance(schema, dict): # Remove specific keys unsupported by Gemini tool parameters schema.pop("additionalProperties", None) schema.pop("default", None) # Check for unsupported 'format' in string types if schema.get("type") == "string" and "format" in schema: allowed_formats = {"enum", "date-time"} if schema["format"] not in allowed_formats: schema.pop("format") # Recursively clean nested schemas (properties, items, etc.) for key, value in list(schema.items()): # Use list() to allow modification during iteration schema[key] = clean_gemini_schema(value) elif isinstance(schema, list): # Recursively clean items in a list return [clean_gemini_schema(item) for item in schema] return schema # Models for Anthropic API requests class ContentBlockText(BaseModel): type: Literal["text"] text: str class ContentBlockImage(BaseModel): type: Literal["image"] source: Dict[str, Any] class ContentBlockToolUse(BaseModel): type: Literal["tool_use"] id: str name: str input: Dict[str, Any] class ContentBlockToolResult(BaseModel): type: Literal["tool_result"] tool_use_id: str content: Union[str, List[Dict[str, Any]], Dict[str, Any], List[Any], Any] class SystemContent(BaseModel): type: Literal["text"] text: str class Message(BaseModel): role: Literal["user", "assistant"] content: Union[str, List[Union[ContentBlockText, ContentBlockImage, ContentBlockToolUse, ContentBlockToolResult]]] class Tool(BaseModel): name: str description: Optional[str] = None input_schema: Dict[str, Any] class ThinkingConfig(BaseModel): enabled: bool class MessagesRequest(BaseModel): model: str max_tokens: int messages: List[Message] system: Optional[Union[str, List[SystemContent]]] = None stop_sequences: Optional[List[str]] = None stream: Optional[bool] = False temperature: Optional[float] = 1.0 top_p: Optional[float] = None top_k: Optional[int] = None metadata: Optional[Dict[str, Any]] = None tools: Optional[List[Tool]] = None tool_choice: Optional[Dict[str, Any]] = None thinking: Optional[ThinkingConfig] = None original_model: Optional[str] = None # Will store the original model name @model_validator(mode='before') def preserve_original_model(cls, values): """Preserve the original model name before any mapping occurs""" if isinstance(values, dict) and 'model' in values: values['original_model'] = values['model'] return values @field_validator('model') def validate_model_field(cls, v, info): # Renamed to avoid conflict original_model = v new_model = v # Default to original value # Remove provider prefixes for easier matching clean_v = v if clean_v.startswith('anthropic/'): clean_v = clean_v[10:] elif clean_v.startswith('openai/'): clean_v = clean_v[7:] elif clean_v.startswith('gemini/'): clean_v = clean_v[7:] # --- Mapping Logic --- START --- mapped = False # Map Haiku to SMALL_MODEL based on provider preference if 'haiku' in clean_v.lower(): # Check what type of model SMALL_MODEL is (with or without prefix) small_model_clean = SMALL_MODEL.replace('gemini/', '').replace('openai/', '').replace('anthropic/', '') if PREFERRED_PROVIDER == "google" and small_model_clean in GEMINI_MODELS: # Use SMALL_MODEL as-is if it already has prefix, otherwise add gemini/ new_model = SMALL_MODEL if SMALL_MODEL.startswith('gemini/') else f"gemini/{SMALL_MODEL}" mapped = True elif small_model_clean in ANTHROPIC_MODELS: # Use SMALL_MODEL as-is if it already has prefix, otherwise add anthropic/ new_model = SMALL_MODEL if SMALL_MODEL.startswith('anthropic/') else f"anthropic/{SMALL_MODEL}" mapped = True else: # Use SMALL_MODEL as-is if it already has prefix, otherwise add openai/ new_model = SMALL_MODEL if SMALL_MODEL.startswith('openai/') else f"openai/{SMALL_MODEL}" mapped = True # Map Sonnet to BIG_MODEL based on provider preference elif 'sonnet' in clean_v.lower(): # Check what type of model BIG_MODEL is (with or without prefix) big_model_clean = BIG_MODEL.replace('gemini/', '').replace('openai/', '').replace('anthropic/', '') if PREFERRED_PROVIDER == "google" and big_model_clean in GEMINI_MODELS: # Use BIG_MODEL as-is if it already has prefix, otherwise add gemini/ new_model = BIG_MODEL if BIG_MODEL.startswith('gemini/') else f"gemini/{BIG_MODEL}" mapped = True elif big_model_clean in ANTHROPIC_MODELS: # Use BIG_MODEL as-is if it already has prefix, otherwise add anthropic/ new_model = BIG_MODEL if BIG_MODEL.startswith('anthropic/') else f"anthropic/{BIG_MODEL}" mapped = True else: # Use BIG_MODEL as-is if it already has prefix, otherwise add openai/ new_model = BIG_MODEL if BIG_MODEL.startswith('openai/') else f"openai/{BIG_MODEL}" mapped = True # Add prefixes to non-mapped models if they match known lists elif not mapped: if clean_v in ANTHROPIC_MODELS and not v.startswith('anthropic/'): new_model = f"anthropic/{clean_v}" mapped = True # Technically mapped to add prefix elif clean_v in GEMINI_MODELS and not v.startswith('gemini/'): new_model = f"gemini/{clean_v}" mapped = True # Technically mapped to add prefix elif clean_v in OPENAI_MODELS and not v.startswith('openai/'): new_model = f"openai/{clean_v}" mapped = True # Technically mapped to add prefix # --- Mapping Logic --- END --- if not mapped: # If no mapping occurred and no prefix exists, log warning or decide default if not v.startswith(('openai/', 'gemini/', 'anthropic/')): logger.warning(f"⚠️ No prefix or mapping rule for model: '{original_model}'. Using as is.") new_model = v # Ensure we return the original if no rule applied return new_model class TokenCountRequest(BaseModel): model: str messages: List[Message] system: Optional[Union[str, List[SystemContent]]] = None tools: Optional[List[Tool]] = None thinking: Optional[ThinkingConfig] = None tool_choice: Optional[Dict[str, Any]] = None original_model: Optional[str] = None # Will store the original model name @model_validator(mode='before') def preserve_original_model(cls, values): """Preserve the original model name before any mapping occurs""" if isinstance(values, dict) and 'model' in values: values['original_model'] = values['model'] return values @field_validator('model') def validate_model_token_count(cls, v, info): # Renamed to avoid conflict # Use the same logic as MessagesRequest validator # NOTE: Pydantic validators might not share state easily if not class methods # Re-implementing the logic here for clarity, could be refactored original_model = v new_model = v # Default to original value # Remove provider prefixes for easier matching clean_v = v if clean_v.startswith('anthropic/'): clean_v = clean_v[10:] elif clean_v.startswith('openai/'): clean_v = clean_v[7:] elif clean_v.startswith('gemini/'): clean_v = clean_v[7:] # --- Mapping Logic --- START --- mapped = False # Map Haiku to SMALL_MODEL based on provider preference if 'haiku' in clean_v.lower(): # Check what type of model SMALL_MODEL is (with or without prefix) small_model_clean = SMALL_MODEL.replace('gemini/', '').replace('openai/', '').replace('anthropic/', '') if PREFERRED_PROVIDER == "google" and small_model_clean in GEMINI_MODELS: # Use SMALL_MODEL as-is if it already has prefix, otherwise add gemini/ new_model = SMALL_MODEL if SMALL_MODEL.startswith('gemini/') else f"gemini/{SMALL_MODEL}" mapped = True elif small_model_clean in ANTHROPIC_MODELS: # Use SMALL_MODEL as-is if it already has prefix, otherwise add anthropic/ new_model = SMALL_MODEL if SMALL_MODEL.startswith('anthropic/') else f"anthropic/{SMALL_MODEL}" mapped = True else: # Use SMALL_MODEL as-is if it already has prefix, otherwise add openai/ new_model = SMALL_MODEL if SMALL_MODEL.startswith('openai/') else f"openai/{SMALL_MODEL}" mapped = True # Map Sonnet to BIG_MODEL based on provider preference elif 'sonnet' in clean_v.lower(): # Check what type of model BIG_MODEL is (with or without prefix) big_model_clean = BIG_MODEL.replace('gemini/', '').replace('openai/', '').replace('anthropic/', '') if PREFERRED_PROVIDER == "google" and big_model_clean in GEMINI_MODELS: # Use BIG_MODEL as-is if it already has prefix, otherwise add gemini/ new_model = BIG_MODEL if BIG_MODEL.startswith('gemini/') else f"gemini/{BIG_MODEL}" mapped = True elif big_model_clean in ANTHROPIC_MODELS: # Use BIG_MODEL as-is if it already has prefix, otherwise add anthropic/ new_model = BIG_MODEL if BIG_MODEL.startswith('anthropic/') else f"anthropic/{BIG_MODEL}" mapped = True else: # Use BIG_MODEL as-is if it already has prefix, otherwise add openai/ new_model = BIG_MODEL if BIG_MODEL.startswith('openai/') else f"openai/{BIG_MODEL}" mapped = True # Add prefixes to non-mapped models if they match known lists elif not mapped: if clean_v in ANTHROPIC_MODELS and not v.startswith('anthropic/'): new_model = f"anthropic/{clean_v}" mapped = True # Technically mapped to add prefix elif clean_v in GEMINI_MODELS and not v.startswith('gemini/'): new_model = f"gemini/{clean_v}" mapped = True # Technically mapped to add prefix elif clean_v in OPENAI_MODELS and not v.startswith('openai/'): new_model = f"openai/{clean_v}" mapped = True # Technically mapped to add prefix # --- Mapping Logic --- END --- if not mapped: if not v.startswith(('openai/', 'gemini/', 'anthropic/')): logger.warning(f"⚠️ No prefix or mapping rule for token count model: '{original_model}'. Using as is.") new_model = v # Ensure we return the original if no rule applied # Store the original model in the values dictionary values = info.data if isinstance(values, dict): values['original_model'] = original_model return new_model class TokenCountResponse(BaseModel): input_tokens: int class Usage(BaseModel): input_tokens: int output_tokens: int cache_creation_input_tokens: int = 0 cache_read_input_tokens: int = 0 class MessagesResponse(BaseModel): id: str model: str role: Literal["assistant"] = "assistant" content: List[Union[ContentBlockText, ContentBlockToolUse]] type: Literal["message"] = "message" stop_reason: Optional[Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"]] = None stop_sequence: Optional[str] = None usage: Usage @app.middleware("http") async def log_requests(request: Request, call_next): # Get request details method = request.method path = request.url.path # Log only basic request details at debug level # Process the request and get the response response = await call_next(request) return response # Not using validation function as we're using the environment API key def parse_tool_result_content(content): """Helper function to properly parse and normalize tool result content.""" if content is None: return "No content provided" if isinstance(content, str): return content if isinstance(content, list): result = "" for item in content: if isinstance(item, dict) and item.get("type") == "text": result += item.get("text", "") + "\n" elif isinstance(item, str): result += item + "\n" elif isinstance(item, dict): if "text" in item: result += item.get("text", "") + "\n" else: try: result += json.dumps(item) + "\n" except: result += str(item) + "\n" else: try: result += str(item) + "\n" except: result += "Unparseable content\n" return result.strip() if isinstance(content, dict): if content.get("type") == "text": return content.get("text", "") try: return json.dumps(content) except: return str(content) # Fallback for any other type try: return str(content) except: return "Unparseable content" def convert_anthropic_to_litellm(anthropic_request: MessagesRequest) -> Dict[str, Any]: """Convert Anthropic API request format to LiteLLM format (which follows OpenAI).""" # LiteLLM already handles Anthropic models when using the format model="anthropic/claude-3-opus-20240229" # So we just need to convert our Pydantic model to a dict in the expected format messages = [] # Add system message if present if anthropic_request.system: # Handle different formats of system messages if isinstance(anthropic_request.system, str): # Simple string format messages.append({"role": "system", "content": anthropic_request.system}) elif isinstance(anthropic_request.system, list): # List of content blocks system_text = "" for block in anthropic_request.system: if hasattr(block, 'type') and block.type == "text": system_text += block.text + "\n\n" elif isinstance(block, dict) and block.get("type") == "text": system_text += block.get("text", "") + "\n\n" if system_text: messages.append({"role": "system", "content": system_text.strip()}) # Add conversation messages for idx, msg in enumerate(anthropic_request.messages): content = msg.content if isinstance(content, str): messages.append({"role": msg.role, "content": content}) else: # Special handling for tool_result in user messages # OpenAI/LiteLLM format expects the assistant to call the tool, # and the user's next message to include the result as plain text if msg.role == "user" and any(block.type == "tool_result" for block in content if hasattr(block, "type")): # For user messages with tool_result, split into separate messages text_content = "" # Extract all text parts and concatenate them for block in content: if hasattr(block, "type"): if block.type == "text": text_content += block.text + "\n" elif block.type == "tool_result": # Add tool result as a message by itself - simulate the normal flow tool_id = block.tool_use_id if hasattr(block, "tool_use_id") else "" # Handle different formats of tool result content result_content = "" if hasattr(block, "content"): if isinstance(block.content, str): result_content = block.content elif isinstance(block.content, list): # If content is a list of blocks, extract text from each for content_block in block.content: if hasattr(content_block, "type") and content_block.type == "text": result_content += content_block.text + "\n" elif isinstance(content_block, dict) and content_block.get("type") == "text": result_content += content_block.get("text", "") + "\n" elif isinstance(content_block, dict): # Handle any dict by trying to extract text or convert to JSON if "text" in content_block: result_content += content_block.get("text", "") + "\n" else: try: result_content += json.dumps(content_block) + "\n" except: result_content += str(content_block) + "\n" elif isinstance(block.content, dict): # Handle dictionary content if block.content.get("type") == "text": result_content = block.content.get("text", "") else: try: result_content = json.dumps(block.content) except: result_content = str(block.content) else: # Handle any other type by converting to string try: result_content = str(block.content) except: result_content = "Unparseable content" # In OpenAI format, tool results come from the user (rather than being content blocks) text_content += f"Tool result for {tool_id}:\n{result_content}\n" # Add as a single user message with all the content messages.append({"role": "user", "content": text_content.strip()}) else: # Regular handling for other message types processed_content = [] for block in content: if hasattr(block, "type"): if block.type == "text": processed_content.append({"type": "text", "text": block.text}) elif block.type == "image": processed_content.append({"type": "image", "source": block.source}) elif block.type == "tool_use": # Handle tool use blocks if needed processed_content.append({ "type": "tool_use", "id": block.id, "name": block.name, "input": block.input }) elif block.type == "tool_result": # Handle different formats of tool result content processed_content_block = { "type": "tool_result", "tool_use_id": block.tool_use_id if hasattr(block, "tool_use_id") else "" } # Process the content field properly if hasattr(block, "content"): if isinstance(block.content, str): # If it's a simple string, create a text block for it processed_content_block["content"] = [{"type": "text", "text": block.content}] elif isinstance(block.content, list): # If it's already a list of blocks, keep it processed_content_block["content"] = block.content else: # Default fallback processed_content_block["content"] = [{"type": "text", "text": str(block.content)}] else: # Default empty content processed_content_block["content"] = [{"type": "text", "text": ""}] processed_content.append(processed_content_block) messages.append({"role": msg.role, "content": processed_content}) # Cap max_tokens for OpenAI models to their limit of 16384 max_tokens = anthropic_request.max_tokens if anthropic_request.model.startswith("openai/") or anthropic_request.model.startswith("gemini/"): max_tokens = min(max_tokens, 16384) # Create LiteLLM request dict litellm_request = { "model": anthropic_request.model, # t understands "anthropic/claude-x" format "messages": messages, "max_tokens": max_tokens, "temperature": anthropic_request.temperature, "stream": anthropic_request.stream, } # Add optional parameters if present if anthropic_request.stop_sequences: litellm_request["stop"] = anthropic_request.stop_sequences if anthropic_request.top_p: litellm_request["top_p"] = anthropic_request.top_p if anthropic_request.top_k: litellm_request["top_k"] = anthropic_request.top_k # Convert tools to OpenAI format if anthropic_request.tools: openai_tools = [] is_gemini_model = anthropic_request.model.startswith("gemini/") for tool in anthropic_request.tools: # Convert to dict if it's a pydantic model if hasattr(tool, 'dict'): tool_dict = tool.dict() else: # Ensure tool_dict is a dictionary, handle potential errors if 'tool' isn't dict-like try: tool_dict = dict(tool) if not isinstance(tool, dict) else tool except (TypeError, ValueError): logger.error(f"Could not convert tool to dict: {tool}") continue # Skip this tool if conversion fails # Clean the schema if targeting a Gemini model input_schema = tool_dict.get("input_schema", {}) if is_gemini_model: input_schema = clean_gemini_schema(input_schema) # Create OpenAI-compatible function tool openai_tool = { "type": "function", "function": { "name": tool_dict["name"], "description": tool_dict.get("description", ""), "parameters": input_schema # Use potentially cleaned schema } } openai_tools.append(openai_tool) litellm_request["tools"] = openai_tools # Convert tool_choice to OpenAI format if present if anthropic_request.tool_choice: if hasattr(anthropic_request.tool_choice, 'dict'): tool_choice_dict = anthropic_request.tool_choice.dict() else: tool_choice_dict = anthropic_request.tool_choice # Handle Anthropic's tool_choice format choice_type = tool_choice_dict.get("type") if choice_type == "auto": litellm_request["tool_choice"] = "auto" elif choice_type == "any": litellm_request["tool_choice"] = "any" elif choice_type == "tool" and "name" in tool_choice_dict: litellm_request["tool_choice"] = { "type": "function", "function": {"name": tool_choice_dict["name"]} } else: # Default to auto if we can't determine litellm_request["tool_choice"] = "auto" return litellm_request def convert_litellm_to_anthropic(litellm_response: Union[Dict[str, Any], Any], original_request: MessagesRequest) -> MessagesResponse: """Convert LiteLLM (OpenAI format) response to Anthropic API response format.""" # Enhanced response extraction with better error handling try: # Get the original model name (before mapping) to check capabilities # This determines if the client expects Claude-style tool_use blocks original_model = getattr(original_request, 'original_model', None) or original_request.model clean_original_model = original_model if clean_original_model.startswith("anthropic/"): clean_original_model = clean_original_model[len("anthropic/"):] elif clean_original_model.startswith("openai/"): clean_original_model = clean_original_model[len("openai/"):] elif clean_original_model.startswith("gemini/"): clean_original_model = clean_original_model[len("gemini/"):] # Check if this should be treated as a Claude model (which supports tool_use blocks) # This includes: # 1. Original requests for Claude models (claude-3-5-sonnet, etc.) # 2. Original requests that start with anthropic/ # 3. Models mapped to anthropic/ providers through LiteLLM # Check original request model is_claude_request = clean_original_model.startswith("claude-") or "claude" in clean_original_model.lower() # Check if the ORIGINAL request was for an anthropic/ model is_anthropic_request = original_model.startswith("anthropic/") # Check if the MAPPED model is an anthropic/ model (through LiteLLM) mapped_model = original_request.model # This is the final mapped model is_anthropic_mapped = mapped_model.startswith("anthropic/") # A model should get Claude-style tool_use blocks if: # - The original request was for a Claude model, OR # - The original request was for an anthropic/ model, OR # - The mapped model is using anthropic/ provider is_claude_model = is_claude_request or is_anthropic_request or is_anthropic_mapped # Handle ModelResponse object from LiteLLM if hasattr(litellm_response, 'choices') and hasattr(litellm_response, 'usage'): # Extract data from ModelResponse object directly choices = litellm_response.choices message = choices[0].message if choices and len(choices) > 0 else None content_text = message.content if message and hasattr(message, 'content') else "" tool_calls = message.tool_calls if message and hasattr(message, 'tool_calls') else None finish_reason = choices[0].finish_reason if choices and len(choices) > 0 else "stop" usage_info = litellm_response.usage response_id = getattr(litellm_response, 'id', f"msg_{uuid.uuid4()}") else: # For backward compatibility - handle dict responses # If response is a dict, use it, otherwise try to convert to dict try: response_dict = litellm_response if isinstance(litellm_response, dict) else litellm_response.dict() except AttributeError: # If .dict() fails, try to use model_dump or __dict__ try: response_dict = litellm_response.model_dump() if hasattr(litellm_response, 'model_dump') else litellm_response.__dict__ except AttributeError: # Fallback - manually extract attributes response_dict = { "id": getattr(litellm_response, 'id', f"msg_{uuid.uuid4()}"), "choices": getattr(litellm_response, 'choices', [{}]), "usage": getattr(litellm_response, 'usage', {}) } # Extract the content from the response dict choices = response_dict.get("choices", [{}]) message = choices[0].get("message", {}) if choices and len(choices) > 0 else {} content_text = message.get("content", "") tool_calls = message.get("tool_calls", None) finish_reason = choices[0].get("finish_reason", "stop") if choices and len(choices) > 0 else "stop" usage_info = response_dict.get("usage", {}) response_id = response_dict.get("id", f"msg_{uuid.uuid4()}") # Create content list for Anthropic format content = [] # Add text content block if present (text might be None or empty for pure tool call responses) if content_text is not None and content_text != "": content.append({"type": "text", "text": content_text}) # Add tool calls if present (tool_use in Anthropic format) - only for Claude models if tool_calls and is_claude_model: # Convert to list if it's not already if not isinstance(tool_calls, list): tool_calls = [tool_calls] for idx, tool_call in enumerate(tool_calls): # Extract function data based on whether it's a dict or object if isinstance(tool_call, dict): function = tool_call.get("function", {}) tool_id = tool_call.get("id", f"tool_{uuid.uuid4()}") name = function.get("name", "") arguments = function.get("arguments", "{}") else: function = getattr(tool_call, "function", None) tool_id = getattr(tool_call, "id", f"tool_{uuid.uuid4()}") name = getattr(function, "name", "") if function else "" arguments = getattr(function, "arguments", "{}") if function else "{}" # Convert string arguments to dict if needed if isinstance(arguments, str): try: arguments = json.loads(arguments) except json.JSONDecodeError: logger.warning(f"Failed to parse tool arguments as JSON: {arguments}") arguments = {"raw": arguments} content.append({ "type": "tool_use", "id": tool_id, "name": name, "input": arguments }) elif tool_calls and not is_claude_model: # For non-Claude models, convert tool calls to text format # We'll append tool info to the text content tool_text = "\n\nTool usage:\n" # Convert to list if it's not already if not isinstance(tool_calls, list): tool_calls = [tool_calls] for idx, tool_call in enumerate(tool_calls): # Extract function data based on whether it's a dict or object if isinstance(tool_call, dict): function = tool_call.get("function", {}) tool_id = tool_call.get("id", f"tool_{uuid.uuid4()}") name = function.get("name", "") arguments = function.get("arguments", "{}") else: function = getattr(tool_call, "function", None) tool_id = getattr(tool_call, "id", f"tool_{uuid.uuid4()}") name = getattr(function, "name", "") if function else "" arguments = getattr(function, "arguments", "{}") if function else "{}" # Convert string arguments to dict if needed if isinstance(arguments, str): try: args_dict = json.loads(arguments) arguments_str = json.dumps(args_dict, indent=2) except json.JSONDecodeError: arguments_str = arguments else: arguments_str = json.dumps(arguments, indent=2) tool_text += f"Tool: {name}\nArguments: {arguments_str}\n\n" # Add or append tool text to content if content and content[0]["type"] == "text": content[0]["text"] += tool_text else: content.append({"type": "text", "text": tool_text}) # Get usage information - extract values safely from object or dict if isinstance(usage_info, dict): prompt_tokens = usage_info.get("prompt_tokens", 0) completion_tokens = usage_info.get("completion_tokens", 0) else: prompt_tokens = getattr(usage_info, "prompt_tokens", 0) completion_tokens = getattr(usage_info, "completion_tokens", 0) # Map OpenAI finish_reason to Anthropic stop_reason stop_reason = None if finish_reason == "stop": stop_reason = "end_turn" elif finish_reason == "length": stop_reason = "max_tokens" elif finish_reason == "tool_calls": stop_reason = "tool_use" else: stop_reason = "end_turn" # Default # Make sure content is never empty if not content: content.append({"type": "text", "text": ""}) # Create Anthropic-style response anthropic_response = MessagesResponse( id=response_id, model=original_request.model, role="assistant", content=content, stop_reason=stop_reason, stop_sequence=None, usage=Usage( input_tokens=prompt_tokens, output_tokens=completion_tokens ) ) return anthropic_response except Exception as e: import traceback error_traceback = traceback.format_exc() error_message = f"Error converting response: {str(e)}\n\nFull traceback:\n{error_traceback}" logger.error(error_message) # In case of any error, create a fallback response return MessagesResponse( id=f"msg_{uuid.uuid4()}", model=original_request.model, role="assistant", content=[{"type": "text", "text": f"Error converting response: {str(e)}. Please check server logs."}], stop_reason="end_turn", usage=Usage(input_tokens=0, output_tokens=0) ) async def handle_streaming(response_generator, original_request: MessagesRequest): """Handle streaming responses from LiteLLM and convert to Anthropic format.""" try: # Send message_start event message_id = f"msg_{uuid.uuid4().hex[:24]}" # Format similar to Anthropic's IDs message_data = { 'type': 'message_start', 'message': { 'id': message_id, 'type': 'message', 'role': 'assistant', 'model': original_request.model, 'content': [], 'stop_reason': None, 'stop_sequence': None, 'usage': { 'input_tokens': 0, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'output_tokens': 0 } } } yield f"event: message_start\ndata: {json.dumps(message_data)}\n\n" # Content block index for the first text block yield f"event: content_block_start\ndata: {json.dumps({'type': 'content_block_start', 'index': 0, 'content_block': {'type': 'text', 'text': ''}})}\n\n" # Send a ping to keep the connection alive (Anthropic does this) yield f"event: ping\ndata: {json.dumps({'type': 'ping'})}\n\n" tool_index = None current_tool_call = None tool_content = "" accumulated_text = "" # Track accumulated text content text_sent = False # Track if we've sent any text content text_block_closed = False # Track if text block is closed input_tokens = 0 output_tokens = 0 has_sent_stop_reason = False last_tool_index = 0 last_ping = time.time() # Track last ping time PING_INTERVAL = 30 # Send ping every 30 seconds # Process each chunk async for chunk in response_generator: try: # Send periodic keep-alive pings to prevent timeout if time.time() - last_ping > PING_INTERVAL: yield f"event: ping\ndata: {json.dumps({'type': 'ping'})}\n\n" last_ping = time.time() # Check if this is the end of the response with usage data if hasattr(chunk, 'usage') and chunk.usage is not None: if hasattr(chunk.usage, 'prompt_tokens'): input_tokens = chunk.usage.prompt_tokens if hasattr(chunk.usage, 'completion_tokens'): output_tokens = chunk.usage.completion_tokens # Handle text content if hasattr(chunk, 'choices') and len(chunk.choices) > 0: choice = chunk.choices[0] # Get the delta from the choice if hasattr(choice, 'delta'): delta = choice.delta else: # If no delta, try to get message delta = getattr(choice, 'message', {}) # Check for finish_reason to know when we're done finish_reason = getattr(choice, 'finish_reason', None) # Process text content delta_content = None # Handle different formats of delta content if hasattr(delta, 'content'): delta_content = delta.content elif isinstance(delta, dict) and 'content' in delta: delta_content = delta['content'] # Accumulate text content if delta_content is not None and delta_content != "": accumulated_text += delta_content # Always emit text deltas if no tool calls started if tool_index is None and not text_block_closed: text_sent = True yield f"event: content_block_delta\ndata: {json.dumps({'type': 'content_block_delta', 'index': 0, 'delta': {'type': 'text_delta', 'text': delta_content}})}\n\n" # Process tool calls delta_tool_calls = None # Handle different formats of tool calls if hasattr(delta, 'tool_calls'): delta_tool_calls = delta.tool_calls elif isinstance(delta, dict) and 'tool_calls' in delta: delta_tool_calls = delta['tool_calls'] # Process tool calls if any if delta_tool_calls: # First tool call we've seen - need to handle text properly if tool_index is None: # If we've been streaming text, close that text block if text_sent and not text_block_closed: text_block_closed = True yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': 0})}\n\n" # If we've accumulated text but not sent it, we need to emit it now # This handles the case where the first delta has both text and a tool call elif accumulated_text and not text_sent and not text_block_closed: # Send the accumulated text text_sent = True yield f"event: content_block_delta\ndata: {json.dumps({'type': 'content_block_delta', 'index': 0, 'delta': {'type': 'text_delta', 'text': accumulated_text}})}\n\n" # Close the text block text_block_closed = True yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': 0})}\n\n" # Close text block even if we haven't sent anything - models sometimes emit empty text blocks elif not text_block_closed: text_block_closed = True yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': 0})}\n\n" # Convert to list if it's not already if not isinstance(delta_tool_calls, list): delta_tool_calls = [delta_tool_calls] for tool_call in delta_tool_calls: # Get the index of this tool call (for multiple tools) current_index = None if isinstance(tool_call, dict) and 'index' in tool_call: current_index = tool_call['index'] elif hasattr(tool_call, 'index'): current_index = tool_call.index else: current_index = 0 # Check if this is a new tool or a continuation if tool_index is None or current_index != tool_index: # New tool call - create a new tool_use block tool_index = current_index last_tool_index += 1 anthropic_tool_index = last_tool_index # Extract function info if isinstance(tool_call, dict): function = tool_call.get('function', {}) name = function.get('name', '') if isinstance(function, dict) else "" tool_id = tool_call.get('id', f"toolu_{uuid.uuid4().hex[:24]}") else: function = getattr(tool_call, 'function', None) name = getattr(function, 'name', '') if function else '' tool_id = getattr(tool_call, 'id', f"toolu_{uuid.uuid4().hex[:24]}") # Start a new tool_use block yield f"event: content_block_start\ndata: {json.dumps({'type': 'content_block_start', 'index': anthropic_tool_index, 'content_block': {'type': 'tool_use', 'id': tool_id, 'name': name, 'input': {}}})}\n\n" current_tool_call = tool_call tool_content = "" # Extract function arguments arguments = None if isinstance(tool_call, dict) and 'function' in tool_call: function = tool_call.get('function', {}) arguments = function.get('arguments', '') if isinstance(function, dict) else '' elif hasattr(tool_call, 'function'): function = getattr(tool_call, 'function', None) arguments = getattr(function, 'arguments', '') if function else '' # If we have arguments, send them as a delta if arguments: # Try to detect if arguments are valid JSON or just a fragment try: # If it's already a dict, use it if isinstance(arguments, dict): args_json = json.dumps(arguments) else: # Otherwise, try to parse it json.loads(arguments) args_json = arguments except (json.JSONDecodeError, TypeError): # If it's a fragment, treat it as a string args_json = arguments # Add to accumulated tool content tool_content += args_json if isinstance(args_json, str) else "" # Send the update yield f"event: content_block_delta\ndata: {json.dumps({'type': 'content_block_delta', 'index': anthropic_tool_index, 'delta': {'type': 'input_json_delta', 'partial_json': args_json}})}\n\n" # Process finish_reason - end the streaming response if finish_reason and not has_sent_stop_reason: has_sent_stop_reason = True # Close any open tool call blocks if tool_index is not None: for i in range(1, last_tool_index + 1): yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': i})}\n\n" # If we accumulated text but never sent or closed text block, do it now if not text_block_closed: if accumulated_text and not text_sent: # Send the accumulated text yield f"event: content_block_delta\ndata: {json.dumps({'type': 'content_block_delta', 'index': 0, 'delta': {'type': 'text_delta', 'text': accumulated_text}})}\n\n" # Close the text block yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': 0})}\n\n" # Map OpenAI finish_reason to Anthropic stop_reason stop_reason = "end_turn" if finish_reason == "length": stop_reason = "max_tokens" elif finish_reason == "tool_calls": stop_reason = "tool_use" elif finish_reason == "stop": stop_reason = "end_turn" # Send message_delta with stop reason and usage usage = {"output_tokens": output_tokens} yield f"event: message_delta\ndata: {json.dumps({'type': 'message_delta', 'delta': {'stop_reason': stop_reason, 'stop_sequence': None}, 'usage': usage})}\n\n" # Send message_stop event yield f"event: message_stop\ndata: {json.dumps({'type': 'message_stop'})}\n\n" # Send final [DONE] marker to match Anthropic's behavior yield "data: [DONE]\n\n" return except Exception as e: # Log error but continue processing other chunks logger.error(f"Error processing chunk: {str(e)}") continue # If we didn't get a finish reason, close any open blocks if not has_sent_stop_reason: # Close any open tool call blocks if tool_index is not None: for i in range(1, last_tool_index + 1): yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': i})}\n\n" # Close the text content block yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': 0})}\n\n" # Send final message_delta with usage usage = {"output_tokens": output_tokens} yield f"event: message_delta\ndata: {json.dumps({'type': 'message_delta', 'delta': {'stop_reason': 'end_turn', 'stop_sequence': None}, 'usage': usage})}\n\n" # Send message_stop event yield f"event: message_stop\ndata: {json.dumps({'type': 'message_stop'})}\n\n" # Send final [DONE] marker to match Anthropic's behavior yield "data: [DONE]\n\n" except Exception as e: import traceback error_traceback = traceback.format_exc() error_message = f"Error in streaming: {str(e)}\n\nFull traceback:\n{error_traceback}" logger.error(error_message) # Send error message_delta yield f"event: message_delta\ndata: {json.dumps({'type': 'message_delta', 'delta': {'stop_reason': 'error', 'stop_sequence': None}, 'usage': {'output_tokens': 0}})}\n\n" # Send message_stop event yield f"event: message_stop\ndata: {json.dumps({'type': 'message_stop'})}\n\n" # Send final [DONE] marker yield "data: [DONE]\n\n" @app.post("/v1/messages") async def create_message( request: MessagesRequest, raw_request: Request ): try: # Add API Key authentication if AUTH_API_KEY: provided_api_key = raw_request.headers.get("x-api-key") or raw_request.headers.get("authorization") if provided_api_key and provided_api_key.startswith("Bearer "): provided_api_key = provided_api_key[7:] if provided_api_key != AUTH_API_KEY: logger.error(f"Unauthorized API Key provided: {provided_api_key}") raise HTTPException(status_code=401, detail="Unauthorized") # Debug: print the body to see what Claude Code is sending body = await raw_request.body() # Parse the raw body as JSON since it's bytes body_json = json.loads(body.decode('utf-8')) original_model = body_json.get("model", "unknown") # Get the display name for logging, just the model name without provider prefix display_model = original_model if "/" in display_model: display_model = display_model.split("/")[-1] # Clean model name for capability check clean_model = request.model if clean_model.startswith("anthropic/"): clean_model = clean_model[len("anthropic/"):] elif clean_model.startswith("openai/"): clean_model = clean_model[len("openai/"):] # Convert Anthropic request to LiteLLM format litellm_request = convert_anthropic_to_litellm(request) # Add timeout configuration for long conversations litellm_request["timeout"] = 600 # 10 minutes litellm_request["request_timeout"] = 600 # Determine which API key to use based on the model if request.model.startswith("openai/"): litellm_request["api_key"] = OPENAI_API_KEY # Add custom API base URL if provided if OPENAI_API_BASE: litellm_request["api_base"] = OPENAI_API_BASE elif request.model.startswith("gemini/"): litellm_request["api_key"] = GEMINI_API_KEY elif request.model.startswith("anthropic/"): # For Anthropic models routed through litellm proxy, use OpenAI credentials litellm_request["api_key"] = OPENAI_API_KEY if OPENAI_API_BASE: litellm_request["api_base"] = OPENAI_API_BASE # Force the provider to be openai so litellm uses OpenAI-compatible endpoints litellm_request["custom_llm_provider"] = "openai" else: litellm_request["api_key"] = ANTHROPIC_API_KEY # For OpenAI models - modify request format to work with limitations if "openai" in litellm_request["model"] and "messages" in litellm_request: # For OpenAI models, we need to convert content blocks to simple strings # and handle other requirements for i, msg in enumerate(litellm_request["messages"]): # Special case - handle message content directly when it's a list of tool_result # This is a specific case we're seeing in the error if "content" in msg and isinstance(msg["content"], list): is_only_tool_result = True for block in msg["content"]: if not isinstance(block, dict) or block.get("type") != "tool_result": is_only_tool_result = False break if is_only_tool_result and len(msg["content"]) > 0: logger.warning(f"Found message with only tool_result content - special handling required") # Extract the content from all tool_result blocks all_text = "" for block in msg["content"]: all_text += "Tool Result:\n" result_content = block.get("content", []) # Handle different formats of content if isinstance(result_content, list): for item in result_content: if isinstance(item, dict) and item.get("type") == "text": all_text += item.get("text", "") + "\n" elif isinstance(item, dict): # Fall back to string representation of any dict try: item_text = item.get("text", json.dumps(item)) all_text += item_text + "\n" except: all_text += str(item) + "\n" elif isinstance(result_content, str): all_text += result_content + "\n" else: try: all_text += json.dumps(result_content) + "\n" except: all_text += str(result_content) + "\n" # Replace the list with extracted text litellm_request["messages"][i]["content"] = all_text.strip() or "..." logger.warning(f"Converted tool_result to plain text: {all_text.strip()[:200]}...") continue # Skip normal processing for this message # 1. Handle content field - normal case if "content" in msg: # Check if content is a list (content blocks) if isinstance(msg["content"], list): # Convert complex content blocks to simple string text_content = "" for block in msg["content"]: if isinstance(block, dict): # Handle different content block types if block.get("type") == "text": text_content += block.get("text", "") + "\n" # Handle tool_result content blocks - extract nested text elif block.get("type") == "tool_result": tool_id = block.get("tool_use_id", "unknown") text_content += f"[Tool Result ID: {tool_id}]\n" # Extract text from the tool_result content result_content = block.get("content", []) if isinstance(result_content, list): for item in result_content: if isinstance(item, dict) and item.get("type") == "text": text_content += item.get("text", "") + "\n" elif isinstance(item, dict): # Handle any dict by trying to extract text or convert to JSON if "text" in item: text_content += item.get("text", "") + "\n" else: try: text_content += json.dumps(item) + "\n" except: text_content += str(item) + "\n" elif isinstance(result_content, dict): # Handle dictionary content if result_content.get("type") == "text": text_content += result_content.get("text", "") + "\n" else: try: text_content += json.dumps(result_content) + "\n" except: text_content += str(result_content) + "\n" elif isinstance(result_content, str): text_content += result_content + "\n" else: try: text_content += json.dumps(result_content) + "\n" except: text_content += str(result_content) + "\n" # Handle tool_use content blocks elif block.get("type") == "tool_use": tool_name = block.get("name", "unknown") tool_id = block.get("id", "unknown") tool_input = json.dumps(block.get("input", {})) text_content += f"[Tool: {tool_name} (ID: {tool_id})]\nInput: {tool_input}\n\n" # Handle image content blocks elif block.get("type") == "image": text_content += "[Image content - not displayed in text format]\n" # Make sure content is never empty for OpenAI models if not text_content.strip(): text_content = "..." litellm_request["messages"][i]["content"] = text_content.strip() # Also check for None or empty string content elif msg["content"] is None: litellm_request["messages"][i]["content"] = "..." # Empty content not allowed # 2. Remove any fields OpenAI doesn't support in messages for key in list(msg.keys()): if key not in ["role", "content", "name", "tool_call_id", "tool_calls"]: logger.warning(f"Removing unsupported field from message: {key}") del msg[key] # 3. Final validation - check for any remaining invalid values and dump full message details for i, msg in enumerate(litellm_request["messages"]): # If content is still a list or None, replace with placeholder if isinstance(msg.get("content"), list): logger.warning(f"CRITICAL: Message {i} still has list content after processing: {json.dumps(msg.get('content'))}") # Last resort - stringify the entire content as JSON litellm_request["messages"][i]["content"] = f"Content as JSON: {json.dumps(msg.get('content'))}" elif msg.get("content") is None: logger.warning(f"Message {i} has None content - replacing with placeholder") litellm_request["messages"][i]["content"] = "..." # Fallback placeholder # Handle streaming mode if request.stream: # Use LiteLLM for streaming num_tools = len(request.tools) if request.tools else 0 log_request_beautifully( "POST", raw_request.url.path, display_model, litellm_request.get('model'), len(litellm_request['messages']), num_tools, 200 # Assuming success at this point ) # Ensure we use the async version for streaming response_generator = await litellm.acompletion(**litellm_request) # Add streaming-specific headers to prevent buffering headers = { "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" # Disable nginx buffering } return StreamingResponse( handle_streaming(response_generator, request), media_type="text/event-stream", headers=headers ) else: # Use LiteLLM for regular completion num_tools = len(request.tools) if request.tools else 0 log_request_beautifully( "POST", raw_request.url.path, display_model, litellm_request.get('model'), len(litellm_request['messages']), num_tools, 200 # Assuming success at this point ) start_time = time.time() litellm_response = litellm.completion(**litellm_request) # Convert LiteLLM response to Anthropic format anthropic_response = convert_litellm_to_anthropic(litellm_response, request) return anthropic_response except Exception as e: import traceback error_traceback = traceback.format_exc() # Capture as much info as possible about the error error_details = { "error": str(e), "type": type(e).__name__, "traceback": error_traceback } # Check for LiteLLM-specific attributes with safe serialization for attr in ['message', 'status_code', 'response', 'llm_provider', 'model']: if hasattr(e, attr): attr_value = getattr(e, attr) # Safely convert to string if not JSON serializable try: json.dumps(attr_value) error_details[attr] = attr_value except (TypeError, ValueError): error_details[attr] = str(attr_value) # Check for additional exception details in dictionaries if hasattr(e, '__dict__'): for key, value in e.__dict__.items(): if key not in error_details and key not in ['args', '__traceback__']: # Always convert to string for safety try: json.dumps(value) error_details[key] = value except (TypeError, ValueError): error_details[key] = str(value) # Log all error details (now safe for JSON serialization) try: logger.error(f"Error processing request: {json.dumps(error_details, indent=2)}") except (TypeError, ValueError) as json_error: # Fallback if there are still serialization issues logger.error(f"Error processing request (JSON serialization failed): {error_details}") logger.error(f"JSON error: {json_error}") # Format error for response error_message = f"Error: {str(e)}" if 'message' in error_details and error_details['message']: error_message += f"\nMessage: {error_details['message']}" if 'response' in error_details and error_details['response']: error_message += f"\nResponse: {error_details['response']}" # Return detailed error status_code = error_details.get('status_code', 500) raise HTTPException(status_code=status_code, detail=error_message) @app.post("/v1/messages/count_tokens") async def count_tokens( request: TokenCountRequest, raw_request: Request ): try: # Log the incoming token count request original_model = request.original_model or request.model # Get the display name for logging, just the model name without provider prefix display_model = original_model if "/" in display_model: display_model = display_model.split("/")[-1] # Clean model name for capability check clean_model = request.model if clean_model.startswith("anthropic/"): clean_model = clean_model[len("anthropic/"):] elif clean_model.startswith("openai/"): clean_model = clean_model[len("openai/"):] # Convert the messages to a format LiteLLM can understand converted_request = convert_anthropic_to_litellm( MessagesRequest( model=request.model, max_tokens=100, # Arbitrary value not used for token counting messages=request.messages, system=request.system, tools=request.tools, tool_choice=request.tool_choice, thinking=request.thinking ) ) # Use LiteLLM's token_counter function try: # Import token_counter function from litellm import token_counter # Log the request beautifully num_tools = len(request.tools) if request.tools else 0 log_request_beautifully( "POST", raw_request.url.path, display_model, converted_request.get('model'), len(converted_request['messages']), num_tools, 200 # Assuming success at this point ) # Count tokens token_count = token_counter( model=converted_request["model"], messages=converted_request["messages"], ) # Return Anthropic-style response return TokenCountResponse(input_tokens=token_count) except ImportError: logger.error("Could not import token_counter from litellm") # Fallback to a simple approximation return TokenCountResponse(input_tokens=1000) # Default fallback except Exception as e: import traceback error_traceback = traceback.format_exc() logger.error(f"Error counting tokens: {str(e)}\n{error_traceback}") raise HTTPException(status_code=500, detail=f"Error counting tokens: {str(e)}") @app.get("/") async def root(): return {"message": "Anthropic Proxy for LiteLLM"} # Define ANSI color codes for terminal output class Colors: CYAN = "\033[96m" BLUE = "\033[94m" GREEN = "\033[92m" YELLOW = "\033[93m" RED = "\033[91m" MAGENTA = "\033[95m" RESET = "\033[0m" BOLD = "\033[1m" UNDERLINE = "\033[4m" DIM = "\033[2m" def log_request_beautifully(method, path, claude_model, openai_model, num_messages, num_tools, status_code): """Log requests in a beautiful, twitter-friendly format showing Claude to OpenAI mapping.""" # Format the Claude model name nicely claude_display = f"{Colors.CYAN}{claude_model}{Colors.RESET}" # Extract endpoint name endpoint = path if "?" in endpoint: endpoint = endpoint.split("?")[0] # Extract just the OpenAI model name without provider prefix openai_display = openai_model if "/" in openai_display: openai_display = openai_display.split("/")[-1] openai_display = f"{Colors.GREEN}{openai_display}{Colors.RESET}" # Format tools and messages tools_str = f"{Colors.MAGENTA}{num_tools} tools{Colors.RESET}" messages_str = f"{Colors.BLUE}{num_messages} messages{Colors.RESET}" # Format status code status_str = f"{Colors.GREEN}✓ {status_code} OK{Colors.RESET}" if status_code == 200 else f"{Colors.RED}✗ {status_code}{Colors.RESET}" # Put it all together in a clear, beautiful format log_line = f"{Colors.BOLD}{method} {endpoint}{Colors.RESET} {status_str}" model_line = f"{claude_display} → {openai_display} {tools_str} {messages_str}" # Print to console print(log_line) print(model_line) sys.stdout.flush() if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "--help": print("Run with: uvicorn server:app --reload --host 0.0.0.0 --port 8082") sys.exit(0) # Configure uvicorn to run with minimal logs and streaming timeouts uvicorn.run( app, host="0.0.0.0", port=8082, log_level="error", timeout_keep_alive=300, # 5 minutes keep-alive for long conversations timeout_graceful_shutdown=30, limit_max_requests=1000 )