import asyncio import copy import enum import importlib import inspect import io import os import random import re import secrets import shutil import subprocess import sys import threading import time import traceback import warnings from datetime import datetime, timedelta, timezone from typing import ( TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Literal, Optional, Set, Tuple, Union, cast, get_args, get_origin, get_type_hints, ) import anyio import websockets import websockets.exceptions from pydantic import BaseModel, Json from litellm._uuid import uuid from litellm.constants import ( AIOHTTP_CONNECTOR_LIMIT, AIOHTTP_CONNECTOR_LIMIT_PER_HOST, AIOHTTP_KEEPALIVE_TIMEOUT, AIOHTTP_NEEDS_CLEANUP_CLOSED, AIOHTTP_TTL_DNS_CACHE, AUDIO_SPEECH_CHUNK_SIZE, BASE_MCP_ROUTE, DEFAULT_MAX_RECURSE_DEPTH, DEFAULT_SHARED_HEALTH_CHECK_LOCK_TTL, DEFAULT_SHARED_HEALTH_CHECK_TTL, DEFAULT_SLACK_ALERTING_THRESHOLD, LITELLM_EMBEDDING_PROVIDERS_SUPPORTING_INPUT_ARRAY_OF_TOKENS, LITELLM_SETTINGS_SAFE_DB_OVERRIDES, LITELLM_UI_ALLOW_HEADERS, LITELLM_UI_SESSION_DURATION, DAILY_TAG_SPEND_BATCH_MULTIPLIER, ) from litellm.litellm_core_utils.litellm_logging import ( _init_custom_logger_compatible_class, ) from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.proxy._types import ( CallbackDelete, CallInfo, CommonProxyErrors, ConfigFieldDelete, ConfigFieldInfo, ConfigFieldUpdate, ConfigGeneralSettings, ConfigList, ConfigYAML, EnterpriseLicenseData, FieldDetail, InvitationClaim, InvitationDelete, InvitationModel, InvitationNew, InvitationUpdate, Litellm_EntityType, LiteLLM_EndUserTable, LiteLLM_JWTAuth, LiteLLM_TagTable, LiteLLM_TeamTable, LiteLLM_TeamTableCachedObj, LiteLLM_UserTable, LitellmUserRoles, PassThroughGenericEndpoint, ProxyErrorTypes, ProxyException, RoleBasedPermissions, SpecialModelNames, SupportedDBObjectType, TeamDefaultSettings, TokenCountRequest, TransformRequestBody, UI_TEAM_ID, UserAPIKeyAuth, ) from litellm.proxy.common_utils.cache_pydantic_utils import CacheCodec from litellm.proxy.common_utils.callback_utils import ( normalize_callback_names, process_callback, ) from litellm.proxy.common_utils.realtime_utils import _realtime_request_body from litellm.types.utils import ( ModelResponse, ModelResponseStream, TextCompletionResponse, TokenCountResponse, ) from litellm.utils import ( _invalidate_model_cost_lowercase_map, load_credentials_from_list, ) if TYPE_CHECKING: from aiohttp import ClientSession from opentelemetry.trace import Span as _Span from litellm.integrations.opentelemetry import OpenTelemetry Span = Union[_Span, Any] else: Span = Any OpenTelemetry = Any REALTIME_REQUEST_SCOPE_TEMPLATE: Dict[str, Any] = { "type": "http", "method": "POST", "path": "/v1/realtime", } def showwarning(message, category, filename, lineno, file=None, line=None): traceback_info = f"{filename}:{lineno}: {category.__name__}: {message}\n" if file is not None: file.write(traceback_info) warnings.showwarning = showwarning warnings.filterwarnings("default", category=UserWarning) # Your client code here messages: list = [] sys.path.insert( 0, os.path.abspath("../..") ) # Adds the parent directory to the system path - for litellm local dev try: import logging import backoff import fastapi import orjson import yaml # type: ignore from apscheduler.schedulers.asyncio import AsyncIOScheduler except ImportError as e: raise ImportError(f"Missing dependency {e}. Run `pip install 'litellm[proxy]'`") list_of_messages = [ "'The thing I wish you improved is...'", "'A feature I really want is...'", "'The worst thing about this product is...'", "'This product would be better if...'", "'I don't like how this works...'", "'It would help me if you could add...'", "'This feature doesn't meet my needs because...'", "'I get frustrated when the product...'", ] def generate_feedback_box(): box_width = 60 # Select a random message message = random.choice(list_of_messages) print() # noqa print("\033[1;37m" + "#" + "-" * box_width + "#\033[0m") # noqa print("\033[1;37m" + "#" + " " * box_width + "#\033[0m") # noqa print("\033[1;37m" + "# {:^59} #\033[0m".format(message)) # noqa print( # noqa "\033[1;37m" + "# {:^59} #\033[0m".format("https://github.com/BerriAI/litellm/issues/new") ) # noqa print("\033[1;37m" + "#" + " " * box_width + "#\033[0m") # noqa print("\033[1;37m" + "#" + "-" * box_width + "#\033[0m") # noqa print() # noqa print(" Thank you for using LiteLLM! - Krrish & Ishaan") # noqa print() # noqa print() # noqa print() # noqa print( # noqa "\033[1;31mGive Feedback / Get Help: https://github.com/BerriAI/litellm/issues/new\033[0m" ) # noqa print() # noqa print() # noqa import contextlib from collections import defaultdict from contextlib import asynccontextmanager from functools import lru_cache import litellm from litellm import Router from litellm._logging import verbose_proxy_logger, verbose_router_logger from litellm.caching.caching import DualCache, RedisCache from litellm.caching.redis_cluster_cache import RedisClusterCache from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache from litellm.constants import ( _REALTIME_BODY_CACHE_SIZE, APSCHEDULER_COALESCE, APSCHEDULER_MAX_INSTANCES, APSCHEDULER_MISFIRE_GRACE_TIME, APSCHEDULER_REPLACE_EXISTING, DAYS_IN_A_MONTH, DEFAULT_HEALTH_CHECK_INTERVAL, DEFAULT_MODEL_CREATED_AT_TIME, LITELLM_PROXY_ADMIN_NAME, PROMETHEUS_FALLBACK_STATS_SEND_TIME_HOURS, PROXY_BATCH_POLLING_ENABLED, PROXY_BATCH_POLLING_INTERVAL, PROXY_BATCH_WRITE_AT, PROXY_BUDGET_RESCHEDULER_MAX_TIME, PROXY_BUDGET_RESCHEDULER_MIN_TIME, ) from litellm.exceptions import RejectedRequestError from litellm.integrations.custom_guardrail import ModifyResponseException from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting from litellm.litellm_core_utils.core_helpers import ( _get_parent_otel_span_from_kwargs, get_litellm_metadata_from_kwargs, ) from litellm.litellm_core_utils.credential_accessor import CredentialAccessor from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler from litellm.llms.vertex_ai.vertex_llm_base import VertexBase from litellm.proxy._types import * from litellm.proxy._lazy_features import attach_lazy_features from litellm.proxy.analytics_endpoints.analytics_endpoints import ( router as analytics_router, ) from litellm.proxy.auth.auth_checks import ( ExperimentalUIJWTToken, get_team_object, log_db_metrics, ) from litellm.proxy.auth.auth_utils import check_response_size_is_safe from litellm.proxy.auth.handle_jwt import JWTHandler from litellm.proxy.auth.litellm_license import LicenseCheck from litellm.proxy.auth.model_checks import ( get_all_fallbacks, get_complete_model_list, get_key_models, get_mcp_server_ids, get_team_models, ) from litellm.proxy.auth.user_api_key_auth import ( _fetch_global_spend_with_event_coordination, user_api_key_auth, user_api_key_auth_websocket, ) from litellm.proxy.batches_endpoints.endpoints import router as batches_router ## Import All Misc routes here ## from litellm.proxy.caching_routes import router as caching_router from litellm.proxy.common_request_processing import ( ProxyBaseLLMRequestProcessing, _is_azure_model_router_request, create_response, ) from litellm.proxy.common_utils.callback_utils import initialize_callbacks_on_proxy from litellm.proxy.common_utils.debug_utils import init_verbose_loggers from litellm.proxy.common_utils.debug_utils import router as debugging_endpoints_router from litellm.proxy.common_utils.encrypt_decrypt_utils import ( decrypt_value_helper, encrypt_value_helper, ) from litellm.proxy.common_utils.html_forms.ui_login import build_ui_login_form from litellm.proxy.common_utils.http_parsing_utils import ( _read_request_body, _safe_get_request_headers, check_file_size_under_limit, get_form_data, ) from litellm.proxy.common_utils.load_config_utils import ( get_config_file_contents_from_gcs, get_file_contents_from_s3, ) from litellm.proxy.common_utils.openai_endpoint_utils import ( remove_sensitive_info_from_deployment, ) from litellm.proxy.common_utils.proxy_state import ProxyState from litellm.proxy.common_utils.reset_budget_job import ResetBudgetJob from litellm.proxy.common_utils.swagger_utils import ERROR_RESPONSES from litellm.proxy.container_endpoints.endpoints import router as container_router from litellm.proxy.credential_endpoints.endpoints import router as credential_router from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import SpendLogCleanup from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler from litellm.proxy.db.spend_counter_reseed import SpendCounterReseed from litellm.proxy.discovery_endpoints import ui_discovery_endpoints_router from litellm.proxy.fine_tuning_endpoints.endpoints import router as fine_tuning_router from litellm.proxy.fine_tuning_endpoints.endpoints import set_fine_tuning_config from litellm.proxy.google_endpoints.endpoints import router as google_router from litellm.proxy.managed_agents_endpoints import router as managed_agents_router from litellm.proxy.guardrails.init_guardrails import ( init_guardrails_v2, initialize_guardrails, ) from litellm.proxy.health_check import perform_health_check from litellm.proxy.health_endpoints._health_endpoints import router as health_router from litellm.proxy.hooks.model_max_budget_limiter import ( _PROXY_VirtualKeyModelMaxBudgetLimiter, ) from litellm.proxy.hooks.prompt_injection_detection import ( _OPTIONAL_PromptInjectionDetection, ) from litellm.proxy.hooks.proxy_track_cost_callback import _ProxyDBLogger from litellm.proxy.image_endpoints.endpoints import router as image_router from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request from litellm.proxy.management_endpoints.budget_management_endpoints import ( router as budget_management_router, ) from litellm.proxy.management_endpoints.cache_settings_endpoints import ( router as cache_settings_router, ) from litellm.proxy.management_endpoints.callback_management_endpoints import ( router as callback_management_endpoints_router, ) from litellm.proxy.management_endpoints.common_utils import ( _user_has_admin_privileges, _user_has_admin_view, admin_can_invite_user, ) from litellm.proxy.management_endpoints.cost_tracking_settings import ( router as cost_tracking_settings_router, ) from litellm.proxy.management_endpoints.customer_endpoints import ( router as customer_router, ) from litellm.proxy.management_endpoints.fallback_management_endpoints import ( router as fallback_management_router, ) from litellm.proxy.management_endpoints.internal_user_endpoints import ( router as internal_user_router, ) from litellm.proxy.management_endpoints.internal_user_endpoints import user_update from litellm.proxy.management_endpoints.key_management_endpoints import ( delete_verification_tokens, duration_in_seconds, generate_key_helper_fn, ) from litellm.proxy.management_endpoints.key_management_endpoints import ( router as key_management_router, ) from litellm.proxy.management_endpoints.model_access_group_management_endpoints import ( router as model_access_group_management_router, ) from litellm.proxy.management_endpoints.model_management_endpoints import ( _add_model_to_db, _add_team_model_to_db, _deduplicate_litellm_router_models, ) from litellm.proxy.management_endpoints.model_management_endpoints import ( router as model_management_router, ) from litellm.proxy.management_endpoints.organization_endpoints import ( router as organization_router, ) from litellm.proxy.management_endpoints.router_settings_endpoints import ( router as router_settings_router, ) from litellm.proxy.management_endpoints.tag_management_endpoints import ( router as tag_management_router, ) from litellm.proxy.management_endpoints.team_callback_endpoints import ( router as team_callback_router, ) from litellm.proxy.management_endpoints.team_endpoints import router as team_router from litellm.proxy.management_endpoints.team_endpoints import ( update_team, validate_membership, ) from litellm.proxy.management_endpoints.workflow_management_endpoints import ( router as workflow_management_router, ) from litellm.proxy.memory.memory_endpoints import router as memory_router from litellm.proxy.management_endpoints.ui_sso import ( get_disabled_non_admin_personal_key_creation, ) from litellm.proxy.management_endpoints.ui_sso import router as ui_sso_router from litellm.proxy.management_endpoints.user_agent_analytics_endpoints import ( router as user_agent_analytics_router, ) from litellm.proxy.management_helpers.audit_logs import create_audit_log_for_update from litellm.proxy.middleware.in_flight_requests_middleware import ( InFlightRequestsMiddleware, ) from litellm.proxy.middleware.prometheus_auth_middleware import PrometheusAuthMiddleware from litellm.proxy.ocr_endpoints.endpoints import router as ocr_router from litellm.proxy.openai_files_endpoints.files_endpoints import ( router as openai_files_router, ) from litellm.proxy.openai_files_endpoints.files_endpoints import set_files_config from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import ( passthrough_endpoint_router, ) from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import ( router as llm_passthrough_router, ) from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import ( vertex_ai_live_websocket_passthrough, ) from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( initialize_pass_through_endpoints, ) from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( router as pass_through_router, ) from litellm.proxy.public_endpoints import router as public_endpoints_router from litellm.proxy.rag_endpoints.endpoints import router as rag_router from litellm.proxy.rerank_endpoints.endpoints import router as rerank_router from litellm.proxy.response_api_endpoints.endpoints import router as response_router from litellm.proxy.route_llm_request import route_request from litellm.proxy.search_endpoints.endpoints import router as search_router from litellm.proxy.spend_tracking.spend_management_endpoints import ( router as spend_management_router, ) from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload from litellm.proxy.types_utils.utils import get_instance_fn from litellm.proxy.ui_crud_endpoints.proxy_setting_endpoints import ( router as ui_crud_endpoints_router, ) from litellm.proxy.utils import ( PrismaClient, ProxyLogging, ProxyUpdateSpend, _cache_user_row, _get_docs_url, _get_openapi_url, _get_projected_spend_over_limit, _get_redoc_url, _is_projected_spend_over_limit, _is_valid_team_configs, get_config_param, get_custom_url, get_error_message_str, get_server_root_path, handle_exception_on_proxy, hash_password, hash_token, invalidate_config_param, litellm_config_cache, migrate_passwords_to_scrypt_async, model_dump_with_preserved_fields, prefetch_config_params, update_spend, ) from litellm.proxy.video_endpoints.endpoints import router as video_router from litellm.router import ( AssistantsTypedDict, Deployment, LiteLLM_Params, ModelGroupInfo, ) from litellm.scheduler import FlowItem, Scheduler from litellm.secret_managers.aws_secret_manager import load_aws_kms from litellm.secret_managers.google_kms import load_google_kms from litellm.secret_managers.main import ( get_secret, get_secret_bool, get_secret_str, normalize_nonempty_secret_str, str_to_bool, ) from litellm.types.integrations.slack_alerting import SlackAlertingArgs from litellm.types.llms.anthropic import ( AnthropicMessagesRequest, AnthropicResponse, AnthropicResponseContentBlockText, AnthropicResponseUsageBlock, ) from litellm.types.llms.openai import HttpxBinaryResponseContent from litellm.types.proxy.control_plane_endpoints import WorkerRegistryEntry from litellm.types.proxy.management_endpoints.model_management_endpoints import ( ModelGroupInfoProxy, ) from litellm.types.proxy.management_endpoints.ui_sso import ( DefaultTeamSSOParams, LiteLLM_UpperboundKeyGenerateParams, ) from litellm.types.realtime import RealtimeQueryParams from litellm.types.router import DeploymentTypedDict from litellm.types.router import ModelInfo as RouterModelInfo from litellm.types.router import ( RouterGeneralSettings, SearchToolTypedDict, updateDeployment, ) from litellm.types.scheduler import DefaultPriorities from litellm.types.secret_managers.main import ( KeyManagementSettings, KeyManagementSystem, ) from litellm.types.utils import CredentialItem, CustomHuggingfaceTokenizer from litellm.types.utils import ModelInfo as ModelMapInfo from litellm.types.utils import RawRequestTypedDict, StandardLoggingPayload from litellm.utils import _add_custom_logger_callback_to_specific_event try: from litellm._version import version except Exception: version = "0.0.0" litellm.suppress_debug_info = True import json from typing import Union from fastapi import ( Depends, FastAPI, File, Form, Header, HTTPException, Path, Query, Request, Response, UploadFile, WebSocket, WebSocketDisconnect, applications, status, ) from fastapi.encoders import jsonable_encoder from fastapi.middleware.cors import CORSMiddleware from fastapi.openapi.docs import get_swagger_ui_html from fastapi.openapi.utils import get_openapi from fastapi.responses import ( FileResponse, JSONResponse, ORJSONResponse, RedirectResponse, StreamingResponse, ) from fastapi.routing import APIRouter from fastapi.security import OAuth2PasswordBearer from fastapi.security.api_key import APIKeyHeader from fastapi.staticfiles import StaticFiles from litellm.types.agents import AgentConfig # import enterprise folder enterprise_router = APIRouter() try: # when using litellm cli import litellm.proxy.enterprise as enterprise except Exception: # when using litellm docker image try: import enterprise # type: ignore except Exception: pass ################### # Import enterprise routes try: from litellm_enterprise.proxy.enterprise_routes import router as _enterprise_router from litellm_enterprise.proxy.proxy_server import EnterpriseProxyConfig enterprise_router = _enterprise_router enterprise_proxy_config: Optional[EnterpriseProxyConfig] = EnterpriseProxyConfig() except ImportError: enterprise_proxy_config = None ################### server_root_path = get_server_root_path() _license_check = LicenseCheck() premium_user: bool = _license_check.is_premium() premium_user_data: Optional["EnterpriseLicenseData"] = ( _license_check.airgapped_license_data ) global_max_parallel_request_retries_env: Optional[str] = os.getenv( "LITELLM_GLOBAL_MAX_PARALLEL_REQUEST_RETRIES" ) proxy_state = ProxyState() SENSITIVE_DATA_MASKER = SensitiveDataMasker() if global_max_parallel_request_retries_env is None: global_max_parallel_request_retries: int = 3 else: global_max_parallel_request_retries = int(global_max_parallel_request_retries_env) global_max_parallel_request_retry_timeout_env: Optional[str] = os.getenv( "LITELLM_GLOBAL_MAX_PARALLEL_REQUEST_RETRY_TIMEOUT" ) if global_max_parallel_request_retry_timeout_env is None: global_max_parallel_request_retry_timeout: float = 60.0 else: global_max_parallel_request_retry_timeout = float( global_max_parallel_request_retry_timeout_env ) ui_link = f"{server_root_path}/ui" fallback_login_link = f"{server_root_path}/fallback/login" model_hub_link = f"{server_root_path}/ui/model_hub_table" ui_message = f"šŸ‘‰ [```LiteLLM Admin Panel on /ui```]({ui_link}). Create, Edit Keys with SSO. Having issues? Try [```Fallback Login```]({fallback_login_link})" ui_message += "\n\nšŸ’ø [```LiteLLM Model Cost Map```](https://models.litellm.ai/)." ui_message += f"\n\nšŸ”Ž [```LiteLLM Model Hub```]({model_hub_link}). See available models on the proxy. [**Docs**](https://docs.litellm.ai/docs/proxy/ai_hub)" custom_swagger_message = "[**Customize Swagger Docs**](https://docs.litellm.ai/docs/proxy/enterprise#swagger-docs---custom-routes--branding)" ### CUSTOM BRANDING [ENTERPRISE FEATURE] ### _title = os.getenv("DOCS_TITLE", "LiteLLM API") if premium_user else "LiteLLM API" _description = ( os.getenv( "DOCS_DESCRIPTION", f"Enterprise Edition \n\nProxy Server to call 100+ LLMs in the OpenAI format. {custom_swagger_message}\n\n{ui_message}", ) if premium_user else f"Proxy Server to call 100+ LLMs in the OpenAI format. {custom_swagger_message}\n\n{ui_message}" ) def cleanup_router_config_variables(): global master_key, user_config_file_path, otel_logging, user_custom_auth, user_custom_auth_path, user_custom_key_generate, user_custom_key_update, user_custom_sso, user_custom_ui_sso_sign_in_handler, use_background_health_checks, use_shared_health_check, health_check_interval, health_check_concurrency, prisma_client # Set all variables to None master_key = None user_config_file_path = None otel_logging = None user_custom_auth = None user_custom_auth_path = None user_custom_key_generate = None user_custom_key_update = None user_custom_sso = None user_custom_ui_sso_sign_in_handler = None use_background_health_checks = None use_shared_health_check = None health_check_interval = None health_check_concurrency = None prisma_client = None async def proxy_shutdown_event(): global prisma_client, master_key, user_custom_auth, user_custom_key_generate, user_custom_key_update verbose_proxy_logger.info("Shutting down LiteLLM Proxy Server") if prisma_client: verbose_proxy_logger.debug("Disconnecting from Prisma") await prisma_client.disconnect() if litellm.cache is not None: await litellm.cache.disconnect() await jwt_handler.close() if db_writer_client is not None: await db_writer_client.close() # type: ignore[reportGeneralTypeIssues] # flush remaining langfuse logs if "langfuse" in litellm.success_callback: try: # flush langfuse logs on shutdow from litellm.utils import langFuseLogger if langFuseLogger is not None: langFuseLogger.Langfuse.flush() except Exception: # [DO NOT BLOCK shutdown events for this] pass ## RESET CUSTOM VARIABLES ## cleanup_router_config_variables() async def _initialize_shared_aiohttp_session(): """Initialize shared aiohttp session for connection reuse with connection limits.""" try: from aiohttp import ClientSession, TCPConnector from litellm.llms.custom_httpx.http_handler import ( _build_aiohttp_keepalive_socket_factory, ) connector_kwargs: Dict[str, Any] = { "keepalive_timeout": AIOHTTP_KEEPALIVE_TIMEOUT, "ttl_dns_cache": AIOHTTP_TTL_DNS_CACHE, } if AIOHTTP_NEEDS_CLEANUP_CLOSED: connector_kwargs["enable_cleanup_closed"] = True if AIOHTTP_CONNECTOR_LIMIT > 0: connector_kwargs["limit"] = AIOHTTP_CONNECTOR_LIMIT if AIOHTTP_CONNECTOR_LIMIT_PER_HOST > 0: connector_kwargs["limit_per_host"] = AIOHTTP_CONNECTOR_LIMIT_PER_HOST socket_factory = _build_aiohttp_keepalive_socket_factory() if socket_factory is not None: connector_kwargs["socket_factory"] = socket_factory connector = TCPConnector(**connector_kwargs) session = ClientSession(connector=connector) verbose_proxy_logger.info( f"SESSION REUSE: Created shared aiohttp session for connection pooling (ID: {id(session)}, " f"limit={AIOHTTP_CONNECTOR_LIMIT}, limit_per_host={AIOHTTP_CONNECTOR_LIMIT_PER_HOST})" ) return session except Exception as e: verbose_proxy_logger.warning( f"Failed to create shared aiohttp session: {e}. Continuing without session reuse." ) return None @asynccontextmanager async def proxy_startup_event(app: FastAPI): # noqa: PLR0915 global prisma_client, master_key, use_background_health_checks, llm_router, llm_model_list, general_settings, proxy_budget_rescheduler_min_time, proxy_budget_rescheduler_max_time, litellm_proxy_admin_name, db_writer_client, store_model_in_db, premium_user, _license_check, proxy_batch_polling_interval, shared_aiohttp_session import json init_verbose_loggers() ## RUN WORKER STARTUP HOOKS (e.g., gflags initialization) ## _startup_hooks_env = os.environ.get("LITELLM_WORKER_STARTUP_HOOKS", "") if _startup_hooks_env: for _hook_spec in _startup_hooks_env.split(","): _hook_spec = _hook_spec.strip() if not _hook_spec: continue try: if ":" not in _hook_spec: raise ValueError( f"Invalid hook spec '{_hook_spec}': expected format is 'module.path:function_name'" ) _module_path, _func_name = _hook_spec.rsplit(":", 1) _module = importlib.import_module(_module_path) _hook_fn = getattr(_module, _func_name) if inspect.iscoroutinefunction(_hook_fn): await _hook_fn() else: _hook_fn() verbose_proxy_logger.info( "Worker startup hook '%s' executed successfully", _hook_spec ) except Exception as e: verbose_proxy_logger.error( "Worker startup hook '%s' failed: %s", _hook_spec, e ) raise ## CHECK PREMIUM USER verbose_proxy_logger.debug( "litellm.proxy.proxy_server.py::startup() - CHECKING PREMIUM USER - {}".format( premium_user ) ) if premium_user is False: premium_user = _license_check.is_premium() ## CHECK MASTER KEY IN ENVIRONMENT ## master_key = get_secret_str("LITELLM_MASTER_KEY") ### LOAD CONFIG ### worker_config: Optional[Union[str, dict]] = get_secret("WORKER_CONFIG") # type: ignore env_config_yaml: Optional[str] = get_secret_str("CONFIG_FILE_PATH") verbose_proxy_logger.debug("worker_config: %s", worker_config) # check if it's a valid file path if env_config_yaml is not None: if os.path.isfile(env_config_yaml) and proxy_config.is_yaml( config_file_path=env_config_yaml ): ( llm_router, llm_model_list, general_settings, ) = await proxy_config.load_config( router=llm_router, config_file_path=env_config_yaml ) elif worker_config is not None: if ( isinstance(worker_config, str) and os.path.isfile(worker_config) and proxy_config.is_yaml(config_file_path=worker_config) ): ( llm_router, llm_model_list, general_settings, ) = await proxy_config.load_config( router=llm_router, config_file_path=worker_config ) elif os.environ.get("LITELLM_CONFIG_BUCKET_NAME") is not None and isinstance( worker_config, str ): ( llm_router, llm_model_list, general_settings, ) = await proxy_config.load_config( router=llm_router, config_file_path=worker_config ) elif isinstance(worker_config, dict): await initialize(**worker_config) else: # if not, assume it's a json string worker_config = json.loads(worker_config) if isinstance(worker_config, dict): await initialize(**worker_config) # check if DATABASE_URL in environment - load from there if prisma_client is None: _db_url: Optional[str] = get_secret("DATABASE_URL", None) # type: ignore prisma_client = await ProxyStartupEvent._setup_prisma_client( database_url=_db_url, proxy_logging_obj=proxy_logging_obj, user_api_key_cache=user_api_key_cache, ) if prisma_client is not None: async def _run_pw_migration(): try: result = await migrate_passwords_to_scrypt_async(prisma_client) verbose_proxy_logger.info(f"Password migration: {result}") except Exception as e: verbose_proxy_logger.warning(f"Password migration skipped: {e}") asyncio.create_task(_run_pw_migration()) ProxyStartupEvent._initialize_startup_logging( llm_router=llm_router, proxy_logging_obj=proxy_logging_obj, redis_usage_cache=redis_usage_cache, ) ## Validate use_redis_transaction_buffer requires Redis cache ## ProxyStartupEvent._validate_redis_transaction_buffer_config( general_settings=general_settings, redis_usage_cache=redis_usage_cache, ) ## SEMANTIC TOOL FILTER ## # Read litellm_settings from config for semantic filter initialization try: verbose_proxy_logger.debug("About to initialize semantic tool filter") _config = proxy_config.get_config_state() _litellm_settings = _config.get("litellm_settings", {}) verbose_proxy_logger.debug( f"litellm_settings keys = {list(_litellm_settings.keys())}" ) await ProxyStartupEvent._initialize_semantic_tool_filter( llm_router=llm_router, litellm_settings=_litellm_settings, ) verbose_proxy_logger.debug("After semantic tool filter initialization") except Exception as e: verbose_proxy_logger.error(f"Semantic filter init failed: {e}", exc_info=True) ## JWT AUTH ## ProxyStartupEvent._initialize_jwt_auth( general_settings=general_settings, prisma_client=prisma_client, user_api_key_cache=user_api_key_cache, ) if prompt_injection_detection_obj is not None: # [TODO] - REFACTOR THIS prompt_injection_detection_obj.update_environment(router=llm_router) verbose_proxy_logger.debug("prisma_client: %s", prisma_client) if prisma_client is not None and litellm.max_budget > 0: ProxyStartupEvent._add_proxy_budget_to_db( litellm_proxy_budget_name=litellm_proxy_admin_name ) asyncio.create_task( ProxyStartupEvent._warm_global_spend_cache( litellm_proxy_admin_name=litellm_proxy_admin_name, user_api_key_cache=user_api_key_cache, prisma_client=prisma_client, ) ) ### START BATCH WRITING DB + CHECKING NEW MODELS### if prisma_client is not None: await ProxyStartupEvent.initialize_scheduled_background_jobs( general_settings=general_settings, prisma_client=prisma_client, proxy_budget_rescheduler_min_time=proxy_budget_rescheduler_min_time, proxy_budget_rescheduler_max_time=proxy_budget_rescheduler_max_time, proxy_batch_write_at=proxy_batch_write_at, proxy_logging_obj=proxy_logging_obj, ) await ProxyStartupEvent._update_default_team_member_budget() ## SYNC UI SETTINGS ## await ProxyStartupEvent._sync_ui_settings_to_general_settings() # Start background health checks AFTER models are loaded and index is built if use_background_health_checks: asyncio.create_task( _run_background_health_check() ) # start the background health check coroutine. # Start adaptive-router queue flusher unconditionally — adaptive routers # may be added later via `/config/reload`, and the flusher is a no-op when # `llm_router.adaptive_routers` is empty. Per-router DB state is loaded # lazily by the flusher on first tick (see `_state_loaded` flag) so # hot-reloaded routers also get their persisted priors. if llm_router is not None and getattr(llm_router, "adaptive_routers", None): for _ar in llm_router.adaptive_routers.values(): await _ar.load_state_from_db(prisma_client) _ar._state_loaded = True asyncio.create_task(_adaptive_router_flusher_loop()) ## [Optional] Initialize dd tracer ProxyStartupEvent._init_dd_tracer() ## [Optional] Initialize Pyroscope continuous profiling (env: LITELLM_ENABLE_PYROSCOPE=true) ProxyStartupEvent._init_pyroscope() ## Initialize shared aiohttp session for connection reuse shared_aiohttp_session = await _initialize_shared_aiohttp_session() ## [Optional] Managed agents — Fargate orphan task reconciler managed_agents_reconciler_task: Optional[asyncio.Task] = None from litellm.proxy.managed_agents_endpoints import ( config_loader as managed_agents_config_loader, ) managed_agents_config_loader.initialize(general_settings) managed_agents_config = managed_agents_config_loader.MANAGED_AGENTS_CONFIG if ( managed_agents_config is not None and managed_agents_config.enabled and prisma_client is not None ): from litellm.proxy.managed_agents_endpoints.lifecycle import reconcile_loop cluster_name = managed_agents_config.aws.cluster or "litellm-agents" managed_agents_reconciler_task = asyncio.create_task( reconcile_loop( prisma_client=prisma_client, region=managed_agents_config.aws_region or "us-east-1", cluster=cluster_name, interval_seconds=managed_agents_config.reconcile_interval_seconds, ) ) verbose_proxy_logger.info( f"managed_agents: started Fargate orphan reconciler (cluster={cluster_name})" ) # End of startup event yield # Shutdown event - cancel managed_agents reconciler if managed_agents_reconciler_task is not None: managed_agents_reconciler_task.cancel() try: await managed_agents_reconciler_task except asyncio.CancelledError: pass except Exception as e: verbose_proxy_logger.error(f"Error stopping managed_agents reconciler: {e}") # Shutdown event - drain pooled passthrough HTTP client try: from litellm.proxy.managed_agents_endpoints.endpoints_passthrough import ( close_passthrough_http_client, ) await close_passthrough_http_client() except Exception as e: verbose_proxy_logger.error( f"Error closing managed_agents passthrough http client: {e}" ) # Shutdown event - close shared aiohttp session if shared_aiohttp_session is not None: try: await shared_aiohttp_session.close() verbose_proxy_logger.info("SESSION REUSE: Closed shared aiohttp session") except Exception as e: verbose_proxy_logger.error(f"Error closing shared aiohttp session: {e}") # Shutdown event - stop RDS IAM token refresh background task if ( prisma_client is not None and hasattr(prisma_client, "db") and hasattr(prisma_client.db, "stop_token_refresh_task") ): try: await prisma_client.db.stop_token_refresh_task() except Exception as e: verbose_proxy_logger.error(f"Error stopping token refresh task: {e}") # Shutdown event - stop Prisma DB health watchdog task if prisma_client is not None and hasattr( prisma_client, "stop_db_health_watchdog_task" ): try: await prisma_client.stop_db_health_watchdog_task() except Exception as e: verbose_proxy_logger.error(f"Error stopping DB health watchdog task: {e}") await proxy_shutdown_event() # type: ignore[reportGeneralTypeIssues] def _generate_stable_operation_id(route: Any) -> str: operation_id = re.sub(r"\W", "_", f"{route.name}{route.path_format}") route_methods = sorted(route.methods or []) if len(route_methods) == 1: operation_id = f"{operation_id}_{route_methods[0].lower()}" return operation_id _OPENAPI_HTTP_METHODS = { "delete", "get", "head", "options", "patch", "post", "put", "trace", } def _strip_operation_id_method_suffix(operation_id: str) -> str: base, separator, suffix = operation_id.rpartition("_") if separator and suffix in _OPENAPI_HTTP_METHODS: return base return operation_id def ensure_unique_openapi_operation_ids( openapi_schema: Dict[str, Any], reserved_operation_ids: Optional[Set[str]] = None, ) -> Dict[str, Any]: operation_entries = [] operation_id_counts: Dict[str, int] = {} for path_item in openapi_schema.get("paths", {}).values(): if not isinstance(path_item, dict): continue for method, operation in path_item.items(): if method not in _OPENAPI_HTTP_METHODS or not isinstance(operation, dict): continue operation_id = operation.get("operationId") if not isinstance(operation_id, str): continue operation_entries.append((method, operation, operation_id)) operation_id_counts[operation_id] = ( operation_id_counts.get(operation_id, 0) + 1 ) used_operation_ids = set(reserved_operation_ids or set()) seen_operation_ids: Set[str] = set() for method, operation, operation_id in operation_entries: should_rewrite = ( operation_id_counts[operation_id] > 1 or operation_id in used_operation_ids or operation_id in seen_operation_ids ) if not should_rewrite: seen_operation_ids.add(operation_id) used_operation_ids.add(operation_id) continue base_operation_id = _strip_operation_id_method_suffix(operation_id) new_operation_id = f"{base_operation_id}_{method}" suffix = 2 while ( new_operation_id in used_operation_ids or new_operation_id in seen_operation_ids ): new_operation_id = f"{base_operation_id}_{method}_{suffix}" suffix += 1 operation["operationId"] = new_operation_id seen_operation_ids.add(new_operation_id) used_operation_ids.add(new_operation_id) if reserved_operation_ids is not None: reserved_operation_ids.update(used_operation_ids) return openapi_schema app = FastAPI( docs_url=_get_docs_url(), redoc_url=_get_redoc_url(), openapi_url=_get_openapi_url(), title=_title, description=_description, version=version, root_path=server_root_path, lifespan=proxy_startup_event, # type: ignore[reportGeneralTypeIssues] generate_unique_id_function=_generate_stable_operation_id, ) vertex_live_passthrough_vertex_base = VertexBase() ### CUSTOM API DOCS [ENTERPRISE FEATURE] ### # Custom OpenAPI schema generator to include only selected routes from fastapi.routing import APIWebSocketRoute def get_openapi_schema(): if app.openapi_schema: return app.openapi_schema # Use compatibility wrapper for FastAPI 0.120+ schema generation from litellm.proxy.common_utils.openapi_schema_compat import ( get_openapi_schema_with_compat, ) openapi_schema = get_openapi_schema_with_compat( get_openapi_func=get_openapi, title=app.title, version=app.version, description=app.description, routes=app.routes, ) # Find all WebSocket routes websocket_routes = [ route for route in app.routes if isinstance(route, APIWebSocketRoute) ] # Add each WebSocket route to the schema for route in websocket_routes: # Get the base path without query parameters base_path = route.path.split("{")[0].rstrip("?") # Extract parameters from the route parameters = [] try: if hasattr(route, "dependant") and route.dependant is not None: # Handle both FastAPI <0.120 and >=0.120 query_params = getattr(route.dependant, "query_params", []) if query_params: for param in query_params: parameters.append( { "name": param.name, "in": "query", "required": param.required, "schema": { "type": "string" }, # You can make this more specific if needed } ) except (AttributeError, TypeError): # If we can't access query_params, continue without them pass openapi_schema["paths"][base_path] = { "get": { "summary": f"WebSocket: {route.name or base_path}", "description": "WebSocket connection endpoint", "operationId": f"websocket_{route.name or base_path.replace('/', '_')}", "parameters": parameters, "responses": {"101": {"description": "WebSocket Protocol Switched"}}, "tags": ["WebSocket"], } } # Add LLM API request schema bodies for documentation from litellm.proxy.common_utils.custom_openapi_spec import CustomOpenAPISpec openapi_schema = CustomOpenAPISpec.add_llm_api_request_schema_body(openapi_schema) # Stub unloaded lazy features so they appear as Swagger sections. from litellm.proxy._lazy_features import inject_lazy_stubs openapi_schema = inject_lazy_stubs(openapi_schema) openapi_schema = ensure_unique_openapi_operation_ids(openapi_schema) # Fix Swagger UI execute path error when server_root_path is set if server_root_path: openapi_schema["servers"] = [{"url": "/" + server_root_path.strip("/")}] app.openapi_schema = openapi_schema return app.openapi_schema def custom_openapi(): if app.openapi_schema: return app.openapi_schema openapi_schema = get_openapi_schema() # Filter routes to include only specific ones openai_routes = LiteLLMRoutes.openai_routes.value paths_to_include: dict = {} for route in openai_routes: if route in openapi_schema["paths"]: paths_to_include[route] = openapi_schema["paths"][route] openapi_schema["paths"] = paths_to_include # Add LLM API request schema bodies for documentation from litellm.proxy.common_utils.custom_openapi_spec import CustomOpenAPISpec openapi_schema = CustomOpenAPISpec.add_llm_api_request_schema_body(openapi_schema) # Stub unloaded lazy features so they appear as Swagger sections. from litellm.proxy._lazy_features import inject_lazy_stubs openapi_schema = inject_lazy_stubs(openapi_schema) openapi_schema = ensure_unique_openapi_operation_ids(openapi_schema) # Fix Swagger UI execute path error when server_root_path is set if server_root_path: openapi_schema["servers"] = [{"url": "/" + server_root_path.strip("/")}] app.openapi_schema = openapi_schema return app.openapi_schema if os.getenv("DOCS_FILTERED", "False") == "True" and premium_user: app.openapi = custom_openapi # type: ignore else: # For regular users, use get_openapi_schema to include LLM API schemas app.openapi = get_openapi_schema # type: ignore class UserAPIKeyCacheTTLEnum(enum.Enum): in_memory_cache_ttl = 60 # 1 min ttl ## configure via `general_settings::user_api_key_cache_ttl: ` @app.exception_handler(ProxyException) async def openai_exception_handler(request: Request, exc: ProxyException): # NOTE: DO NOT MODIFY THIS, its crucial to map to Openai exceptions headers = exc.headers error_dict = exc.to_dict() return JSONResponse( status_code=( int(exc.code) if exc.code else status.HTTP_500_INTERNAL_SERVER_ERROR ), content={"error": error_dict}, headers=headers, ) router = APIRouter() def _get_cors_config( cors_origins_env: Optional[str] = None, cors_credentials_env: Optional[str] = None, ): """ Compute CORS allowed origins and credentials flag from environment variables. Extracted into a function so it can be unit-tested without reloading the module. Args: cors_origins_env: Value of LITELLM_CORS_ORIGINS (defaults to os.getenv). cors_credentials_env: Value of LITELLM_CORS_ALLOW_CREDENTIALS (defaults to os.getenv). Returns: Tuple[List[str], bool]: (origins, allow_credentials) """ _origins_raw = ( cors_origins_env if cors_origins_env is not None else os.getenv("LITELLM_CORS_ORIGINS") ) if _origins_raw is None or _origins_raw.strip() == "": computed_origins = ["*"] else: computed_origins = [o.strip() for o in _origins_raw.split(",") if o.strip()] # Disable credentials by default when wildcard origins are used — combining # allow_origins=["*"] with allow_credentials=True causes Starlette to reflect # the incoming Origin header, allowing any site to make credentialed requests. # Set LITELLM_CORS_ALLOW_CREDENTIALS=true to explicitly restore the old behaviour # (e.g. for non-browser clients that relied on the Access-Control-Allow-Credentials # header being present regardless of origin). _credentials_raw = ( cors_credentials_env if cors_credentials_env is not None else os.getenv("LITELLM_CORS_ALLOW_CREDENTIALS") ) if _credentials_raw is not None: computed_credentials = _credentials_raw.strip().lower() == "true" else: computed_credentials = "*" not in computed_origins return computed_origins, computed_credentials origins, allow_cors_credentials = _get_cors_config() # get current directory try: current_dir = os.path.dirname(os.path.abspath(__file__)) packaged_ui_path = os.path.join(current_dir, "_experimental", "out") ui_path = packaged_ui_path litellm_asset_prefix = "/litellm-asset-prefix" def _dir_has_content(path: str) -> bool: try: return os.path.isdir(path) and any(os.scandir(path)) except FileNotFoundError: return False def _validate_ui_directory(ui_path: str) -> bool: """ Verify UI directory has minimum required structure. Checks for: - Directory exists - Has index.html (main entry point) - Has _next directory (Next.js assets) Returns True if UI directory appears valid and servable. """ if not os.path.isdir(ui_path): return False # Must have main index.html if not os.path.exists(os.path.join(ui_path, "index.html")): return False # Must have _next directory with Next.js assets next_dir = os.path.join(ui_path, "_next") if not os.path.isdir(next_dir): return False return True def _is_ui_pre_restructured(ui_dir: str) -> bool: """ Detect if UI directory is already pre-restructured and ready to serve. Returns True if: 1. Marker file .litellm_ui_ready exists (created by Dockerfile), OR 2. Restructuring pattern detected (subdirectories with index.html inside) This allows skipping copy/restructure operations on read-only filesystems. """ if not os.path.isdir(ui_dir): return False # Primary signal: marker file created by Dockerfile marker_file = os.path.join(ui_dir, ".litellm_ui_ready") if os.path.exists(marker_file): verbose_proxy_logger.debug(f"Found UI ready marker: {marker_file}") return True # Fallback signal: Detect restructuring pattern # After restructuring, routes exist as directories with index.html inside # (e.g., login/index.html instead of login.html) # Check for main index.html first (basic UI structure requirement) if not os.path.exists(os.path.join(ui_dir, "index.html")): return False # Look for ANY subdirectory with index.html (proves restructuring happened) # Ignore directories starting with _ (Next.js internals like _next) try: for entry in os.scandir(ui_dir): if entry.is_dir() and not entry.name.startswith("_"): index_path = os.path.join(entry.path, "index.html") if os.path.exists(index_path): # Found at least one restructured route - this proves the pattern verbose_proxy_logger.debug( f"Detected restructured UI via pattern: found {entry.name}/index.html" ) return True except (PermissionError, OSError) as e: verbose_proxy_logger.debug( f"Could not scan {ui_dir} for restructuring detection: {e}" ) return False # No restructured routes found return False def _try_populate_ui_directory( source_path: str, target_path: str ) -> tuple[bool, str]: """ Attempt to populate target UI directory from source. Returns: (success: bool, error_message: str) """ try: os.makedirs(target_path, exist_ok=True) if not _dir_has_content(target_path) and _dir_has_content(source_path): shutil.copytree( source_path, target_path, dirs_exist_ok=True, ) verbose_proxy_logger.info(f"Successfully populated UI at {target_path}") return True, "" else: return False, "Source or target directory state invalid" except (PermissionError, OSError) as e: return False, str(e) # Use a writable runtime UI directory whenever possible. # This prevents mutating the packaged UI directory (e.g. site-packages or the repo checkout) # and ensures extensionless routes like /ui/login work via /index.html. is_non_root = os.getenv("LITELLM_NON_ROOT", "").lower() == "true" # Determine runtime UI path # Priority: LITELLM_UI_PATH env var > default path based on is_non_root if is_non_root: default_runtime_ui_path = "/var/lib/litellm/ui" else: default_runtime_ui_path = packaged_ui_path runtime_ui_path = os.getenv("LITELLM_UI_PATH", default_runtime_ui_path) # Validate packaged UI before proceeding if not _validate_ui_directory(packaged_ui_path): verbose_proxy_logger.error( f"Packaged UI at {packaged_ui_path} is invalid or incomplete. " f"UI may not function correctly." ) # Decision tree for UI path selection: # 1. If runtime path == packaged path: use packaged UI directly # 2. If runtime UI exists and is pre-restructured: use it # 3. If runtime UI exists but not restructured: use it (will restructure later) # 4. If runtime UI missing: try to populate from packaged UI # 4a. If population succeeds: use runtime UI # 4b. If population fails: fall back to packaged UI should_use_runtime_path = runtime_ui_path != packaged_ui_path if should_use_runtime_path: is_pre_restructured = _is_ui_pre_restructured(runtime_ui_path) has_content = _dir_has_content(runtime_ui_path) # Case 2: Runtime UI exists and is ready if has_content and is_pre_restructured: verbose_proxy_logger.info(f"Using pre-restructured UI at {runtime_ui_path}") ui_path = runtime_ui_path # Case 3: Runtime UI exists but needs restructuring elif has_content and not is_pre_restructured: verbose_proxy_logger.warning( f"UI at {runtime_ui_path} has content but is not properly restructured. " f"Will attempt to restructure in place." ) ui_path = runtime_ui_path # Case 4: Runtime UI missing - try to populate else: verbose_proxy_logger.info( f"UI not found at {runtime_ui_path}. Attempting to populate from packaged UI." ) success, error = _try_populate_ui_directory( packaged_ui_path, runtime_ui_path ) if success: # Case 4a: Population succeeded ui_path = runtime_ui_path else: # Case 4b: Population failed - fall back to packaged UI verbose_proxy_logger.warning( f"Failed to populate UI at {runtime_ui_path}: {error}. " f"Falling back to packaged UI at {packaged_ui_path}. " f"For read-only deployments, pre-build UI in Dockerfile " f"or set LITELLM_UI_PATH to a writable emptyDir volume." ) ui_path = packaged_ui_path else: # Case 1: Using packaged UI directly (local development) verbose_proxy_logger.info(f"Using packaged UI directory: {packaged_ui_path}") ui_path = packaged_ui_path # Validate final UI path if not _validate_ui_directory(ui_path): verbose_proxy_logger.error( f"Selected UI path {ui_path} is invalid or incomplete. UI may not work correctly." ) # Only modify files if a custom server root path is set AND filesystem is writable if server_root_path and server_root_path != "/": # Check if UI path is writable is_writable = os.access(ui_path, os.W_OK) if not is_writable: verbose_proxy_logger.warning( f"Cannot apply server_root_path replacements to UI at {ui_path}: " f"path is not writable. Ensure server_root_path is '/' or pre-process " f"UI files in Dockerfile with custom server_root_path." ) else: # Iterate through files in the UI directory for root, dirs, files in os.walk(ui_path): for filename in files: file_path = os.path.join(root, filename) # Skip binary files and files that don't need path replacement if filename.endswith( ( ".png", ".jpg", ".jpeg", ".gif", ".ico", ".woff", ".woff2", ".ttf", ".eot", ) ): continue try: with open(file_path, "r", encoding="utf-8") as f: content = f.read() # Replace the asset prefix with the server root path modified_content = content.replace( f"{litellm_asset_prefix}", f"{server_root_path}", ) # Replace the /.well-known/litellm-ui-config with the server root path modified_content = modified_content.replace( "/litellm/.well-known/litellm-ui-config", f"{server_root_path}/.well-known/litellm-ui-config", ) with open(file_path, "w", encoding="utf-8") as f: f.write(modified_content) except (UnicodeDecodeError, PermissionError, OSError): # Skip binary files or files we can't write to continue # # Mount the _next directory at the root level app.mount( "/_next", StaticFiles(directory=os.path.join(ui_path, "_next")), name="next_static", ) app.mount( f"{litellm_asset_prefix}/_next", StaticFiles(directory=os.path.join(ui_path, "_next")), name="next_static", ) # print(f"mounted _next at {server_root_path}/ui/_next") app.mount("/ui", StaticFiles(directory=ui_path, html=True), name="ui") def _restructure_ui_html_files(ui_root: str) -> None: """Ensure each exported HTML route is available as /index.html.""" for current_root, _, files in os.walk(ui_root): rel_root = os.path.relpath(current_root, ui_root) first_segment = "" if rel_root == "." else rel_root.split(os.sep)[0] # Ignore Next.js asset directories if first_segment in {"_next", "litellm-asset-prefix"}: continue for filename in files: if not filename.endswith(".html") or filename == "index.html": continue file_path = os.path.join(current_root, filename) target_dir = os.path.splitext(file_path)[0] target_path = os.path.join(target_dir, "index.html") os.makedirs(target_dir, exist_ok=True) try: os.replace(file_path, target_path) except FileNotFoundError: # Another process may have already moved this file. continue # Handle HTML file restructuring # Only restructure if: # 1. UI is not already pre-restructured # 2. Filesystem is writable try: is_pre_restructured = _is_ui_pre_restructured(ui_path) is_writable = os.access(ui_path, os.W_OK) if is_pre_restructured: verbose_proxy_logger.info( f"Skipping UI restructuring: {ui_path} is already pre-restructured" ) elif not is_writable: verbose_proxy_logger.warning( f"Cannot restructure UI at {ui_path}: path is not writable. " f"UI may not work correctly for extensionless routes. " f"Pre-build and restructure UI in Dockerfile for read-only deployments." ) else: _restructure_ui_html_files(ui_path) verbose_proxy_logger.info(f"Restructured UI directory: {ui_path}") except PermissionError as e: verbose_proxy_logger.exception( f"Permission error while restructuring UI directory {ui_path}: {e}" ) except Exception as e: verbose_proxy_logger.exception( f"Error while restructuring UI directory {ui_path}: {e}" ) except Exception: pass current_dir = os.path.dirname(os.path.abspath(__file__)) # ui_path = os.path.join(current_dir, "_experimental", "out") # # Mount this test directory instead # app.mount("/ui", StaticFiles(directory=ui_path, html=True), name="ui") app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=allow_cors_credentials, allow_methods=["*"], allow_headers=["*"], expose_headers=LITELLM_UI_ALLOW_HEADERS, ) app.add_middleware(PrometheusAuthMiddleware) app.add_middleware(InFlightRequestsMiddleware) def mount_swagger_ui(): swagger_directory = os.path.join(current_dir, "swagger") swagger_path = "/" if server_root_path is None else server_root_path if not swagger_path.endswith("/"): swagger_path = swagger_path + "/" custom_root_path_swagger_path = swagger_path + "swagger" app.mount("/swagger", StaticFiles(directory=swagger_directory), name="swagger") # On dropdown expand: one-time fetch to the prefix (triggers lazy load), # then spec re-download so real routes replace the stub. Raw JS (no #