from typing import Any, Dict from fastapi import Request from litellm._logging import verbose_proxy_logger from litellm.proxy._types import UserAPIKeyAuth async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth: """ Handle request from oauth2 proxy. """ from litellm.proxy.proxy_server import general_settings verbose_proxy_logger.debug("Handling oauth2 proxy request") # Define the OAuth2 config mappings oauth2_config_mappings: Dict[str, str] = general_settings.get( "oauth2_config_mappings", None ) verbose_proxy_logger.debug(f"Oauth2 config mappings: {oauth2_config_mappings}") if not oauth2_config_mappings: raise ValueError("Oauth2 config mappings not found in general_settings") # Initialize a dictionary to store the mapped values auth_data: Dict[str, Any] = {} # Extract values from headers based on the mappings for key, header in oauth2_config_mappings.items(): value = request.headers.get(header) if value: # Convert max_budget to float if present if key == "max_budget": auth_data[key] = float(value) # Convert models to list if present elif key == "models": auth_data[key] = [model.strip() for model in value.split(",")] else: auth_data[key] = value verbose_proxy_logger.debug( f"Auth data before creating UserAPIKeyAuth object: {auth_data}" ) user_api_key_auth = UserAPIKeyAuth(**auth_data) verbose_proxy_logger.debug(f"UserAPIKeyAuth object created: {user_api_key_auth}") # Create and return UserAPIKeyAuth object return user_api_key_auth