litellm/litellm/proxy/proxy_cli.py

# ruff: noqa: T201
import functools
import importlib.metadata
import json
import logging.config
import os
import random
import re
import subprocess
import sys
import urllib.parse as urlparse
from collections.abc import Callable, Iterable, Mapping, Sequence
from pathlib import Path
from types import MappingProxyType
from typing import TYPE_CHECKING, Annotated, Any, Final

import click
import httpx
from click.core import ParameterSource
from dotenv import load_dotenv
from pydantic import BeforeValidator, ConfigDict, TypeAdapter
from typing_extensions import ReadOnly, TypedDict

import litellm
from litellm._logging import StartupOutputSettings, print_startup_message, print_startup_warning, terminal_text
from litellm.constants import DEFAULT_NUM_WORKERS_LITELLM_PROXY
from litellm.proxy.db.pgbouncer import (
    PgBouncerError,
    PgBouncerSettings,
    export_pooled_database_url,
    start_in_container_pgbouncer,
)
from litellm.proxy.db.query_engine_reaper import start_query_engine_reaper
from litellm.types.llms.base import LiteLLMBaseModel

if TYPE_CHECKING:
    from fastapi import FastAPI
else:
    FastAPI = Any


def _drop_script_dir_from_sys_path() -> None:
    """Stop ``litellm/proxy`` modules from shadowing installed packages.

    Running this file as a script puts its own directory at ``sys.path[0]``, so
    ``import a2a`` resolves to ``litellm/proxy/a2a`` instead of the ``a2a`` SDK
    and ``proxy_server`` resolves to a second copy of
    ``litellm.proxy.proxy_server``. No-op under the ``litellm`` console script.
    """
    script_dir: Final = os.path.dirname(os.path.abspath(__file__))
    if sys.path and os.path.abspath(sys.path[0]) == script_dir:
        sys.path.pop(0)


_drop_script_dir_from_sys_path()
sys.path.append(os.getcwd())

config_filename: Final = "litellm.secrets"

litellm_mode: Final = os.getenv("LITELLM_MODE", "DEV")  # "PRODUCTION", "DEV"
if litellm_mode == "DEV":
    load_dotenv()
from enum import Enum


class LiteLLMDatabaseConnectionPool(Enum):
    database_connection_pool_limit = 10
    database_connection_pool_timeout = 60


def _build_db_connection_url_params(
    connection_limit: int,
    pool_timeout: float | None,
    connect_timeout: float | None = None,
    socket_timeout: float | None = None,
    disable_prepared_statements: bool = False,
    extra_params: dict | None = None,
) -> dict:
    """Build the Prisma DATABASE_URL query params controlling connection pool behavior.

    `connect_timeout` / `socket_timeout` map to the Prisma URL params of the same
    name (https://www.prisma.io/docs/orm/overview/databases/postgresql) and are
    omitted when None so Prisma's defaults apply. `disable_prepared_statements`
    sets `pgbouncer=true`, which makes Prisma stop using server-side prepared
    statements (pgbouncer transaction-pool compatible; also sidesteps the
    "cached plan must not change result type" error during rolling migrations).
    `extra_params` is an untyped passthrough — keys it provides win over the
    named arguments above, so it can be used to override any default we set here.
    """
    params: Final[dict] = {
        "connection_limit": connection_limit,
    }
    if pool_timeout is not None:
        params["pool_timeout"] = pool_timeout
    if connect_timeout is not None:
        params["connect_timeout"] = connect_timeout
    if socket_timeout is not None:
        params["socket_timeout"] = socket_timeout
    if disable_prepared_statements:
        params["pgbouncer"] = "true"
    if extra_params:
        params.update(extra_params)
    return params


class DatabaseTimeoutSettings(LiteLLMBaseModel):
    """The `general_settings` keys that bound how long a statement may hold locks.

    Validated at the boundary so a mistyped value fails at startup with a clear
    pydantic error instead of a TypeError deep inside URL assembly.
    """

    model_config = ConfigDict(extra="ignore")

    database_statement_timeout: float | None = None
    database_lock_timeout: float | None = None
    database_idle_in_transaction_session_timeout: float | None = None


def _pg_options_with_timeouts(
    existing_options: str,
    statement_timeout: float | None,
    lock_timeout: float | None,
    idle_in_transaction_session_timeout: float | None = None,
) -> str:
    """Return the Postgres ``options`` value carrying the configured timeouts.

    Prisma has no URL parameter for `statement_timeout` / `lock_timeout`; they are
    server settings delivered through the standard libpq ``options`` parameter as
    ``-c <name>=<value>``. Values arrive in seconds (matching the sibling
    ``database_*_timeout`` settings) and are emitted as integer milliseconds,
    which is the unit Postgres assumes for a unit-less value.

    Both settings are bounds on how long a single statement may hold its locks.
    Without them a batch that outlives the Prisma client's HTTP read timeout keeps
    running server side, holding row locks for as long as the database takes,
    while the client has already given up; every later flush queues behind it.
    The query engine's own transaction timeout cannot end that wait, because it
    cannot interrupt a statement that is already executing.

    Any ``options`` the operator already pinned on the URL is preserved, and a
    setting they pinned there wins over the configured one. Postgres applies the
    last occurrence of a setting, so a pinned one is honoured by not appending
    ours at all; it is matched in every spelling the backend accepts
    (``-c name=``, ``-cname=``, ``--name=``). The result is applied to
    ``DATABASE_URL`` only, never ``DIRECT_URL``: that one serves migrations,
    which legitimately run long and must not be cancelled mid-way.
    """
    configured: Final = tuple(
        f"-c {name}={int(seconds * 1000)}"
        for name, seconds in (
            ("statement_timeout", statement_timeout),
            ("lock_timeout", lock_timeout),
            ("idle_in_transaction_session_timeout", idle_in_transaction_session_timeout),
        )
        if seconds is not None and not re.search(rf"(?:-c\s*|--){re.escape(name)}=", existing_options)
    )
    return " ".join(part for part in (existing_options, *configured) if part)


def _url_query_value(url: str | None, key: str) -> str:
    """Return a single query-param value already present on ``url``, else ""."""
    if not isinstance(url, str) or url == "":
        return ""
    return next(iter(urlparse.parse_qs(urlparse.urlparse(url).query).get(key) or ()), "")


def _with_query_value(url: str, key: str, value: str) -> str:
    """Return ``url`` with a single query param replaced by ``value``."""
    parsed: Final = urlparse.urlparse(url)
    pairs: Final = tuple((k, v) for k, v in urlparse.parse_qsl(parsed.query) if k != key) + ((key, value),)
    return urlparse.urlunparse(parsed._replace(query=urlparse.urlencode(pairs)))


def append_query_params(url: str | None, params: dict) -> str:
    from litellm._logging import verbose_proxy_logger

    verbose_proxy_logger.debug("url: %s", url)
    verbose_proxy_logger.debug("params: %s", params)
    if not isinstance(url, str) or url == "":
        # Preserve previous startup behavior when DATABASE_URL is absent.
        # Returning an empty string avoids urlparse type errors in test/dev flows.
        verbose_proxy_logger.warning("append_query_params received empty or non-string URL, returning empty string")
        return ""
    parsed_url: Final = urlparse.urlparse(url)
    parsed_query: Final = urlparse.parse_qs(parsed_url.query)
    parsed_query.update(params)
    encoded_query: Final = urlparse.urlencode(parsed_query, doseq=True)
    modified_url: Final = urlparse.urlunparse(parsed_url._replace(query=encoded_query))
    return modified_url


def resolve_v2_migration_resolver(*, use_legacy_flag: bool, env_value: str | None) -> bool:
    from litellm_proxy_extras.utils import str_to_bool

    if use_legacy_flag:
        return False
    if env_value is None:
        return True
    return bool(str_to_bool(env_value))


def deprecated_v2_flag_passed_on_cli() -> bool:
    ctx: Final = click.get_current_context(silent=True)
    if ctx is None:
        return False
    return ctx.get_parameter_source("use_v2_migration_resolver") is ParameterSource.COMMANDLINE


def _print_enforce_prisma_migration_check_notice() -> None:
    print_startup_warning(
        "\033[1;33mLiteLLM Proxy: --enforce_prisma_migration_check is "
        "deprecated and has no effect, because the proxy always exits "
        "when database setup fails at startup. You can safely remove it.\033[0m",
    )


def _print_or_hold_enforce_prisma_migration_check_notice(
    *, reads_config_file: bool, returns_before_config: bool
) -> Callable[[], None] | None:
    """Print the notice now, or, while a config file can still switch json_logs on, return a printer
    to call once it is read; a boot that fails before that prints the held notice on the way out."""
    if not reads_config_file or litellm.json_logs or returns_before_config:
        _print_enforce_prisma_migration_check_notice()
        return None
    print_once: Final = functools.cache(_print_enforce_prisma_migration_check_notice)
    ctx: Final = click.get_current_context(silent=True)
    if ctx is not None:
        ctx.call_on_close(print_once)
    return print_once


def _empty_as_none(value: object) -> object:
    return value or None


class _ConfigStartupOutputSettings(TypedDict, total=False):
    litellm_settings: ReadOnly[Annotated[StartupOutputSettings | None, BeforeValidator(_empty_as_none)]]


_CONFIG_STARTUP_OUTPUT_SETTINGS: Final = TypeAdapter(_ConfigStartupOutputSettings)


def _apply_config_suppress_startup_output(litellm_settings: StartupOutputSettings | None) -> None:
    if litellm_settings is None or litellm_settings.get("suppress_startup_output") is not True:
        return
    litellm.suppress_startup_output = True
    os.environ["LITELLM_SUPPRESS_STARTUP_OUTPUT"] = "true"


def _apply_config_logging_settings(litellm_settings: StartupOutputSettings | None) -> None:
    """Turn on the config's json_logs and suppress_startup_output before any worker starts. Workers that
    uvicorn or granian spawn import litellm afresh and read both from the env at import, so they log JSON
    and stay quiet from their first line instead of from when load_config reaches the key."""
    _apply_config_suppress_startup_output(litellm_settings)
    if litellm_settings is None or litellm_settings.get("json_logs") is not True:
        return
    litellm.json_logs = True
    os.environ["JSON_LOGS"] = "true"
    litellm.turn_on_json()


WORKER_APP: Final = "litellm_cli.main:proxy_worker_app"
WORKER_LOG_CONFIG_ENV: Final = "LITELLM_WORKER_LOG_CONFIG"
_LOG_CONFIG: Final = TypeAdapter[str | Mapping[str, object]](str | Mapping[str, object])
_LOG_CONFIG_FILE: Final = TypeAdapter[Mapping[str, object]](Mapping[str, object])


class ProxyInitializationHelpers:
    @staticmethod
    def echo_litellm_version():
        pkg_version: Final = importlib.metadata.version("litellm")
        click.echo(f"\nLiteLLM: Current Version = {pkg_version}\n")

    _echo_litellm_version = echo_litellm_version

    @staticmethod
    def run_health_check(host, port):
        print("\nLiteLLM: Health Testing models in config")
        response: Final = httpx.get(url=f"http://{host}:{port}/health")
        print(json.dumps(response.json(), indent=4))

    _run_health_check = run_health_check

    @staticmethod
    def run_migration_data_check(use_migrate: bool) -> int:
        """`--check_migrations_against_existing_data`: the pending migrations applied to DATABASE_URL inside one
        transaction that is rolled back, so Postgres reports what the upgrade would hit. 0 when they all apply,
        2 with Postgres's error when one does not."""
        if os.getenv("DATABASE_URL") is None:
            raise click.UsageError("--check_migrations_against_existing_data requires DATABASE_URL")
        from litellm.proxy.db.prisma_client import PrismaManager

        ok, summary = PrismaManager.migration_dry_run_summary(use_migrate=use_migrate)
        if ok:
            print(f"LiteLLM Proxy: {summary}")
            return 0
        print(
            f"\033[1;31mLiteLLM Proxy: Database migration cannot proceed. {summary}\033[0m", file=sys.stderr, flush=True
        )
        return 2

    @staticmethod
    def run_config_validation(config: str | None) -> None:
        if config is None:
            raise click.UsageError("--validate_config requires --config <path>")
        import asyncio

        from litellm.proxy.proxy_server import ProxyConfig

        async def _load() -> int:
            _, model_list, _ = await ProxyConfig().load_config(router=None, config_file_path=config)
            return len(model_list)

        try:
            model_count: Final = asyncio.run(_load())
        except Exception as error:
            click.echo(f"LiteLLM: config validation failed: {error}", err=True)
            raise click.exceptions.Exit(1) from error
        click.echo(f"LiteLLM: config OK ({model_count} models)")

    _run_config_validation = run_config_validation

    @staticmethod
    def run_test_chat_completion(
        host: str,
        port: int,
        model: str,
        test: bool | str,
    ):
        request_model: Final = model or "gpt-3.5-turbo"
        click.echo(f"\nLiteLLM: Making a test ChatCompletions request to your proxy. Model={request_model}")
        import openai

        api_base = f"http://{host}:{port}"
        if isinstance(test, str):
            api_base = test
        else:
            raise ValueError("Invalid test value")
        client: Final = openai.OpenAI(api_key="My API Key", base_url=api_base)

        response: Final = client.chat.completions.create(
            model=request_model,
            messages=[
                {
                    "role": "user",
                    "content": "this is a test request, write a short poem",
                }
            ],
            max_tokens=256,
        )
        click.echo(f"\nLiteLLM: response from proxy {response}")

        print(f"\n LiteLLM: Making a test ChatCompletions + streaming r equest to proxy. Model={request_model}")

        stream_response: Final = client.chat.completions.create(
            model=request_model,
            messages=[
                {
                    "role": "user",
                    "content": "this is a test request, write a short poem",
                }
            ],
            stream=True,
        )
        for chunk in stream_response:
            click.echo(f"LiteLLM: streaming response from proxy {chunk}")
        print("\n making completion request to proxy")
        completion_response: Final = client.completions.create(
            model=request_model, prompt="this is a test request, write a short poem"
        )
        print(completion_response)

    _run_test_chat_completion = run_test_chat_completion

    @staticmethod
    def apply_log_config(log_config: str | Mapping[str, object]) -> None:
        if not isinstance(log_config, str):
            logging.config.dictConfig(dict(log_config))
        elif log_config.endswith(".json"):
            logging.config.dictConfig(dict(_LOG_CONFIG_FILE.validate_json(Path(log_config).read_bytes())))
        elif log_config.endswith((".yaml", ".yml")):
            import yaml

            logging.config.dictConfig(
                dict(_LOG_CONFIG_FILE.validate_python(yaml.safe_load(Path(log_config).read_text())))
            )
        else:
            logging.config.fileConfig(log_config, disable_existing_loggers=False)

    @staticmethod
    def child_process_uvicorn_overrides(log_config: object) -> Mapping[str, object]:
        if log_config is None:
            return MappingProxyType({"app": WORKER_APP, "factory": True})
        ProxyInitializationHelpers.apply_log_config(_LOG_CONFIG.validate_python(log_config))
        os.environ[WORKER_LOG_CONFIG_ENV] = json.dumps(log_config)
        return MappingProxyType({"app": WORKER_APP, "factory": True, "log_config": None})

    @staticmethod
    def apply_worker_log_config(environ: Mapping[str, str]) -> None:
        log_config: Final = environ.get(WORKER_LOG_CONFIG_ENV)
        if log_config is not None:
            ProxyInitializationHelpers.apply_log_config(_LOG_CONFIG.validate_json(log_config))

    @staticmethod
    def get_default_unvicorn_init_args(
        host: str,
        port: int,
        log_config: str | None = None,
        keepalive_timeout: int | None = None,
        timeout_worker_healthcheck: int | None = None,
    ) -> dict:
        """
        Get the arguments for `uvicorn` worker
        """
        import inspect

        import uvicorn

        import litellm
        from litellm._logging import _get_uvicorn_json_log_config, resolve_log_level
        from litellm.proxy.shutdown.graceful_shutdown_manager import GracefulShutdownManager

        uvicorn_args: Final = {
            "app": "litellm.proxy.proxy_server:app",
            "host": host,
            "port": port,
            "server_header": False,
            "timeout_graceful_shutdown": GracefulShutdownManager.get_timeout(),
        }
        if log_config is not None:
            print_startup_message(f"Using log_config: {log_config}")
            uvicorn_args["log_config"] = log_config
        elif litellm.json_logs:
            # Use JSON log config for uvicorn to ensure all logs (including exceptions) are JSON
            uvicorn_args["log_config"] = _get_uvicorn_json_log_config()
        elif litellm_log := os.environ.get("LITELLM_LOG"):
            uvicorn_args["log_level"] = resolve_log_level(litellm_log)
        if keepalive_timeout is not None:
            uvicorn_args["timeout_keep_alive"] = keepalive_timeout
        if timeout_worker_healthcheck is not None:
            if "timeout_worker_healthcheck" in inspect.signature(uvicorn.Config.__init__).parameters:
                uvicorn_args["timeout_worker_healthcheck"] = timeout_worker_healthcheck
            else:
                print_startup_warning(
                    f"\033[1;33mLiteLLM Proxy: --timeout_worker_healthcheck "
                    f"requires uvicorn>=0.37.0, but installed uvicorn=={uvicorn.__version__}. "
                    f"Ignoring the flag.\033[0m",
                )
        return uvicorn_args

    _get_default_unvicorn_init_args = get_default_unvicorn_init_args

    @staticmethod
    def apply_uvicorn_max_requests_jitter(
        uvicorn_args: dict,
        max_requests_before_restart: int | None,
        jitter: int,
    ) -> None:
        """
        Stagger uvicorn worker restarts via limit_max_requests_jitter (uvicorn>=0.41.0).
        """
        import inspect

        import uvicorn

        if max_requests_before_restart is None:
            print_startup_warning(
                "\033[1;33mLiteLLM Proxy: --max_requests_before_restart_jitter "
                "has no effect without --max_requests_before_restart\033[0m\n",
            )
            return
        if "limit_max_requests_jitter" in inspect.signature(uvicorn.Config.__init__).parameters:
            uvicorn_args["limit_max_requests_jitter"] = jitter
        else:
            print_startup_warning(
                f"\033[1;33mLiteLLM Proxy: --max_requests_before_restart_jitter "
                f"requires uvicorn>=0.41.0, but installed uvicorn=={uvicorn.__version__}. "
                f"Ignoring the flag.\033[0m",
            )

    _apply_uvicorn_max_requests_jitter = apply_uvicorn_max_requests_jitter

    @staticmethod
    def single_worker_max_requests_warning(
        num_workers: int,
        max_requests_before_restart: int | None,
        run_gunicorn: bool,
    ) -> str | None:
        if max_requests_before_restart is None or num_workers > 1:
            return None
        outage: Final = (
            "gunicorn accepts no connections until the replacement worker has booted"
            if run_gunicorn
            else "uvicorn exits the process, so the container restarts"
        )
        return (
            f"LiteLLM Proxy: --max_requests_before_restart {max_requests_before_restart} with --num_workers "
            f"{num_workers} does not recycle a worker. After {max_requests_before_restart} requests, health "
            f"probes included, {outage}. Use --num_workers 2 or more with --max_requests_before_restart_jitter, "
            "or rely on pod replacement instead."
        )

    @staticmethod
    def _get_reload_options(config_path: str | None) -> Mapping[str, object]:
        """Build uvicorn reload kwargs so --reload also reacts to .env and YAML edits."""
        cwd: Final = os.path.abspath(os.getcwd())
        reload_dirs: Final = [cwd]
        # Must be basenames, not absolute paths: uvicorn's
        # resolve_reload_patterns() calls pathlib.Path.glob(), which raises
        # NotImplementedError on absolute patterns (uvicorn discussion #2156).
        reload_includes: Final = ["*.py", ".env"]
        if config_path:
            config_abs: Final = os.path.abspath(config_path)
            config_dir: Final = os.path.dirname(config_abs)
            if config_dir and config_dir != cwd:
                reload_dirs.append(config_dir)
            reload_includes.append(os.path.basename(config_abs))
        return {
            "reload": True,
            "reload_dirs": reload_dirs,
            "reload_includes": reload_includes,
        }

    @staticmethod
    def _patch_statreload_extra_paths(paths: Iterable[str | None]) -> bool:
        """Make uvicorn's StatReload reloader notice non-Python dev files
        (the --config YAML and .env).

        Uvicorn uses WatchFilesReload when the optional `watchfiles` package
        is installed, otherwise StatReload. StatReload hard-codes `*.py` in
        `iter_py_files()` and silently ignores `reload_includes`, so the
        kwargs from `_get_reload_options` alone don't trigger reloads on those
        files. We monkey-patch `iter_py_files` to also yield the given paths.

        Idempotent across calls and a no-op for the WatchFilesReload path.
        """
        try:
            from uvicorn.supervisors.statreload import StatReload
        except ImportError:  # pragma: no cover - uvicorn is a hard dep
            return False

        from pathlib import Path

        resolved: Final = {Path(p).resolve() for p in paths if p}
        if not resolved:
            return False

        patched_paths = getattr(StatReload, "_litellm_patched_config_paths", None)
        if patched_paths is None:
            original_iter: Final = StatReload.iter_py_files
            patched_paths = set()

            def _iter_with_extra(self: StatReload):
                yield from original_iter(self)
                for path in StatReload._litellm_patched_config_paths:
                    if path.exists():
                        yield path

            StatReload.iter_py_files = _iter_with_extra
            StatReload._litellm_patched_config_paths = patched_paths

        patched_paths.update(resolved)
        return True

    @staticmethod
    def configure_dev_reload(uvicorn_args: dict, config_path: str | None) -> None:
        """Wire up --reload (dev only): watch *.py, the --config YAML, and .env,
        and signal reloaded workers to re-read .env with override so edits to
        existing keys actually take effect rather than staying masked by the
        value inherited from the reloader process."""
        from litellm._logging import verbose_proxy_logger

        uvicorn_args.update(ProxyInitializationHelpers._get_reload_options(config_path))
        os.environ["LITELLM_DEV_ENV_HOT_RELOAD"] = "True"
        env_path: Final = os.path.join(os.getcwd(), ".env")
        ProxyInitializationHelpers._patch_statreload_extra_paths([config_path, env_path])
        verbose_proxy_logger.warning(
            "LiteLLM --reload: worker processes re-read .env with override, so .env "
            "values win over shell-exported environment variables. Unset a key in .env "
            "to let a shell-exported value take precedence."
        )

    _configure_dev_reload = configure_dev_reload

    @staticmethod
    def init_hypercorn_server(
        app: FastAPI,
        host: str,
        port: int,
        ssl_certfile_path: str,
        ssl_keyfile_path: str,
        ciphers: str | None = None,
    ):
        """
        Initialize litellm with `hypercorn`
        """
        import asyncio

        from hypercorn.asyncio import serve
        from hypercorn.config import Config

        print_startup_message(f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} using Hypercorn\033[0m\n")
        config: Final = Config()
        config.bind = [f"{host}:{port}"]

        if ssl_certfile_path is not None and ssl_keyfile_path is not None:
            print_startup_message(
                f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
            )
            config.certfile = ssl_certfile_path
            config.keyfile = ssl_keyfile_path
            if ciphers is not None:
                config.ciphers = ciphers

        # hypercorn serve raises a type warning when passing a fast api app - even though fast API is a valid type
        asyncio.run(serve(app, config))

    _init_hypercorn_server = init_hypercorn_server

    @staticmethod
    def init_granian_server(
        host: str,
        port: int,
        num_workers: int,
        ssl_certfile_path: str | None,
        ssl_keyfile_path: str | None,
        max_requests_before_restart: int | None,
        ciphers: str | None,
        granian_runtime_threads: int | None = None,
    ) -> None:
        """
        Run the proxy with Granian (Rust-backed ASGI server, HTTP/1 + HTTP/2).

        Uses a string import path so workers load ``litellm.proxy.proxy_server:app``
        the same way as uvicorn's ``app=`` string target.
        """
        from granian import Granian
        from granian.constants import Interfaces

        print_startup_message(f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} using Granian\033[0m\n")
        if max_requests_before_restart is not None:
            print_startup_warning(
                "\033[1;33mLiteLLM: --max_requests_before_restart is not supported by Granian "
                "(Granian uses workers_lifetime in seconds, not a per-request limit).\033[0m\n",
            )
        if ciphers is not None:
            print_startup_warning("\033[1;33mLiteLLM: --ciphers is not applied when using --run_granian.\033[0m\n")

        kwargs: Final[dict[str, object]] = {
            "target": "litellm.proxy.proxy_server:app",
            "address": host,
            "port": port,
            "workers": max(1, num_workers),
            "interface": Interfaces.ASGI,
            "websockets": True,
        }
        if granian_runtime_threads is not None:
            kwargs["runtime_threads"] = granian_runtime_threads
        if ssl_certfile_path is not None and ssl_keyfile_path is not None:
            print_startup_message(
                f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
            )
            kwargs["ssl_cert"] = Path(ssl_certfile_path)
            kwargs["ssl_key"] = Path(ssl_keyfile_path)
        elif ssl_certfile_path is not None or ssl_keyfile_path is not None:
            raise click.ClickException("Both --ssl_certfile_path and --ssl_keyfile_path are required for SSL.")

        Granian(**kwargs).serve()

    _init_granian_server = init_granian_server

    @staticmethod
    def run_gunicorn_server(
        host: str,
        port: int,
        app: FastAPI,
        num_workers: int,
        ssl_certfile_path: str,
        ssl_keyfile_path: str,
        max_requests_before_restart: int | None = None,
        max_requests_before_restart_jitter: int | None = None,
    ):
        """
        Run litellm with `gunicorn`
        """
        import math

        from litellm.constants import GRACEFUL_SHUTDOWN_TEARDOWN_HEADROOM_SECONDS
        from litellm.proxy.shutdown.graceful_shutdown_manager import GracefulShutdownManager
        from litellm.proxy.shutdown.graceful_uvicorn_worker import build_graceful_shutdown_uvicorn_worker

        if os.name == "nt":
            pass
        else:
            import gunicorn.app.base

        # Gunicorn Application Class
        class StandaloneApplication(gunicorn.app.base.BaseApplication):
            def __init__(self, app, options=None):
                self.options = options or {}  # gunicorn options
                self.application = app  # FastAPI app
                super().__init__()

                _endpoint_str: Final = f"curl --location 'http://0.0.0.0:{port}/chat/completions' \\"
                curl_command: Final = (
                    _endpoint_str
                    + """
                --header 'Content-Type: application/json' \\
                --data ' {
                "model": "gpt-3.5-turbo",
                "messages": [
                    {
                    "role": "user",
                    "content": "what llm are you"
                    }
                ]
                }'
                \n
                """
                )
                print_startup_message("")
                print_startup_message(
                    '\033[1;34mLiteLLM: Test your local proxy with: "litellm --test" This runs an openai.ChatCompletion request to your proxy [In a new terminal tab]\033[0m\n'
                )
                print_startup_message(
                    f"\033[1;34mLiteLLM: Curl Command Test for your local proxy\n {curl_command} \033[0m\n"
                )
                print_startup_message("\033[1;34mDocs: https://docs.litellm.ai/docs/simple_proxy\033[0m\n")
                print_startup_message(f"\033[1;34mSee all Router/Swagger docs on http://0.0.0.0:{port} \033[0m\n")

            def load_config(self):
                # note: This Loads the gunicorn config - has nothing to do with LiteLLM Proxy config
                if self.cfg is not None:
                    config = {
                        key: value
                        for key, value in self.options.items()
                        if key in self.cfg.settings and value is not None
                    }
                else:
                    config = {}
                for key, value in config.items():
                    if self.cfg is not None:
                        self.cfg.set(key.lower(), value)

            def load(self):
                # gunicorn app function
                return self.application

        print_startup_message(
            f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} with {num_workers} workers\033[0m\n"
        )
        graceful_shutdown_timeout: Final = GracefulShutdownManager.get_timeout()
        gunicorn_options: Final = {
            "bind": f"{host}:{port}",
            "workers": num_workers,  # default is 1
            "worker_class": build_graceful_shutdown_uvicorn_worker(graceful_shutdown_timeout),
            "graceful_timeout": math.ceil(graceful_shutdown_timeout) + GRACEFUL_SHUTDOWN_TEARDOWN_HEADROOM_SECONDS,
            "preload": True,  # Add the preload flag,
            "accesslog": "-",  # Log to stdout
            "timeout": 600,  # default to very high number, bedrock/anthropic.claude-v2:1 can take 30+ seconds for the 1st chunk to come in
            "access_log_format": '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s',
        }

        # Optional: recycle workers after N requests to mitigate memory growth
        if max_requests_before_restart is not None:
            gunicorn_options["max_requests"] = max_requests_before_restart
        if max_requests_before_restart_jitter is not None:
            if max_requests_before_restart is None:
                print_startup_warning(
                    "\033[1;33mLiteLLM Proxy: --max_requests_before_restart_jitter "
                    "has no effect without --max_requests_before_restart\033[0m\n",
                )
            else:
                gunicorn_options["max_requests_jitter"] = max_requests_before_restart_jitter

        # Clean up prometheus .db files when a worker exits (prevents ghost gauge values)
        if os.environ.get("PROMETHEUS_MULTIPROC_DIR"):
            from litellm.proxy.prometheus_cleanup import mark_worker_exit

            def child_exit(server, worker):
                mark_worker_exit(worker.pid)

            gunicorn_options["child_exit"] = child_exit

        if ssl_certfile_path is not None and ssl_keyfile_path is not None:
            print_startup_message(
                f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
            )
            gunicorn_options["certfile"] = ssl_certfile_path
            gunicorn_options["keyfile"] = ssl_keyfile_path

        # The master preloads the app and then forks every worker, so native routes are
        # forbidden in it: their runtime threads would not survive the fork.
        from litellm.rust_bridge.fork_guard import reserve_process_for_forking

        reserve_process_for_forking("the gunicorn master")
        start_query_engine_reaper()
        StandaloneApplication(app=app, options=gunicorn_options).run()  # Run gunicorn

    _run_gunicorn_server = run_gunicorn_server

    @staticmethod
    def run_ollama_serve():
        try:
            command: Final = ["ollama", "serve"]

            with open(os.devnull, "w") as devnull:
                subprocess.Popen(command, stdout=devnull, stderr=devnull)
        except Exception as e:
            print_startup_warning(f"""
                LiteLLM Warning: proxy started with `ollama` model\n`ollama serve` failed with Exception{e}. \nEnsure you run `ollama serve`
            """)

    _run_ollama_serve = run_ollama_serve

    @staticmethod
    def is_port_in_use(port):
        import socket

        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            return s.connect_ex(("localhost", port)) == 0

    _is_port_in_use = is_port_in_use

    @staticmethod
    def get_loop_type():
        """Helper function to determine the event loop type based on platform"""
        if sys.platform in ("win32", "cygwin", "cli"):
            return None  # Let uvicorn choose the default loop on Windows
        if os.environ.get("LITELLM_EVENT_LOOP") == "asyncio":
            return "asyncio"
        return "uvloop"

    _get_loop_type = get_loop_type

    @staticmethod
    def _prometheus_callback_configured(litellm_settings: Mapping[str, object] | None) -> bool:
        if litellm_settings is None:
            return False
        configured: Final = tuple(
            litellm_settings.get(key) for key in ("callbacks", "success_callback", "failure_callback")
        )
        return any(
            setting == "prometheus"
            if isinstance(setting, str)
            else isinstance(setting, Sequence) and "prometheus" in setting
            for setting in configured
        )

    @staticmethod
    def maybe_setup_prometheus_multiproc_dir(
        num_workers: int,
        litellm_settings: dict | None,
        prometheus_metrics_port: int | None = None,
    ) -> str | None:
        """
        Auto-create PROMETHEUS_MULTIPROC_DIR when another process needs to read the samples: extra workers
        with prometheus configured as a callback in config.yaml, or the separate metrics server (always, since
        callbacks may also be enabled from the DB after startup).
        """
        import tempfile

        from litellm.proxy.prometheus_cleanup import wipe_directory

        configured_dir: Final = os.environ.get("PROMETHEUS_MULTIPROC_DIR") or os.environ.get("prometheus_multiproc_dir")
        if prometheus_metrics_port is None and (
            num_workers <= 1 or not ProxyInitializationHelpers._prometheus_callback_configured(litellm_settings)
        ):
            if configured_dir:
                wipe_directory(configured_dir)
            return None

        multiproc_dir: Final = configured_dir or os.path.join(tempfile.gettempdir(), "litellm_prometheus_multiproc")
        os.environ["PROMETHEUS_MULTIPROC_DIR"] = multiproc_dir

        os.makedirs(multiproc_dir, exist_ok=True)
        wipe_directory(multiproc_dir)
        action: Final = "Using existing" if configured_dir else "Auto-created"
        print_startup_message(f"LiteLLM: {action} PROMETHEUS_MULTIPROC_DIR={multiproc_dir}")
        return multiproc_dir

    _maybe_setup_prometheus_multiproc_dir = maybe_setup_prometheus_multiproc_dir


@click.command()
@click.argument("cli_args", nargs=-1)
@click.option("--host", default="0.0.0.0", help="Host for the server to listen on.", envvar="HOST")
@click.option("--port", default=4000, help="Port to bind the server to.", envvar="PORT")
@click.option(
    "--num_workers",
    default=DEFAULT_NUM_WORKERS_LITELLM_PROXY,
    help=(
        "Number of worker processes for uvicorn / gunicorn, or Granian worker processes "
        "(--workers). Default is 1 (from DEFAULT_NUM_WORKERS_LITELLM_PROXY). "
        "With --run_granian, use --granian_threads for runtime threads per worker."
    ),
    envvar="NUM_WORKERS",
)
@click.option(
    "--granian_threads",
    default=None,
    type=click.IntRange(min=1),
    help=(
        "Only with --run_granian: runtime threads per worker process "
        "(Granian --runtime-threads / GRANIAN_RUNTIME_THREADS). Omit to use Granian's default (1)."
    ),
    envvar="GRANIAN_RUNTIME_THREADS",
)
@click.option("--api_base", default=None, help="API base URL.")
@click.option(
    "--api_version",
    default=litellm.AZURE_DEFAULT_API_VERSION,
    help="For azure - pass in the api version.",
)
@click.option("--model", "-m", default=None, help="The model name to pass to litellm expects")
@click.option(
    "--alias",
    default=None,
    help='The alias for the model - use this to give a litellm model name (e.g. "huggingface/codellama/CodeLlama-7b-Instruct-hf") a more user-friendly name ("codellama")',
)
@click.option("--add_key", default=None, help="The model name to pass to litellm expects")
@click.option("--headers", default=None, help="headers for the API call")
@click.option("--save", is_flag=True, type=bool, help="Save the model-specific config")
@click.option(
    "--debug",
    default=False,
    is_flag=True,
    type=bool,
    help="To debug the input",
    envvar="DEBUG",
)
@click.option(
    "--detailed_debug",
    default=False,
    is_flag=True,
    type=bool,
    help="To view detailed debug logs",
    envvar="DETAILED_DEBUG",
)
@click.option(
    "--use_queue",
    default=False,
    is_flag=True,
    type=bool,
    help="To use celery workers for async endpoints",
)
@click.option("--temperature", default=None, type=float, help="Set temperature for the model")
@click.option("--max_tokens", default=None, type=int, help="Set max tokens for the model")
@click.option(
    "--request_timeout",
    default=None,
    type=int,
    help="Set timeout in seconds for completion calls",
)
@click.option("--drop_params", is_flag=True, help="Drop any unmapped params")
@click.option(
    "--add_function_to_prompt",
    is_flag=True,
    help="If function passed but unsupported, pass it as prompt",
)
@click.option(
    "--config",
    "-c",
    default=None,
    help="Path to the proxy configuration file (e.g. config.yaml). Usage `litellm --config config.yaml`",
)
@click.option(
    "--max_budget",
    default=None,
    type=float,
    help="Set max budget for API calls - works for hosted models like OpenAI, TogetherAI, Anthropic, etc.`",
)
@click.option(
    "--telemetry",
    default=None,
    type=bool,
    hidden=True,
    expose_value=False,
    help="Deprecated no-op kept so existing start commands still parse",
)
@click.option(
    "--log_config",
    default=None,
    type=str,
    help="Path to the logging configuration file",
)
@click.option(
    "--setup",
    is_flag=True,
    default=False,
    help="Run the interactive setup wizard to configure providers and generate a config file",
)
@click.option(
    "--version",
    "-v",
    default=False,
    is_flag=True,
    type=bool,
    help="Print LiteLLM version",
)
@click.option(
    "--health",
    flag_value=True,
    help="Make a chat/completions request to all llms in config.yaml",
)
@click.option(
    "--test",
    flag_value=True,
    help="proxy chat completions url to make a test request to",
)
@click.option(
    "--test_async",
    default=False,
    is_flag=True,
    help="Calls async endpoints /queue/requests and /queue/response",
)
@click.option(
    "--iam_token_db_auth",
    default=False,
    is_flag=True,
    help="Connects to RDS DB with IAM token",
)
@click.option(
    "--azure_postgresql_auth",
    default=False,
    is_flag=True,
    help="Connects to Azure Database for PostgreSQL with a Microsoft Entra ID token",
)
@click.option(
    "--num_requests",
    default=10,
    type=int,
    help="Number of requests to hit async endpoint with",
)
@click.option(
    "--run_gunicorn",
    default=False,
    is_flag=True,
    help="Starts proxy via gunicorn, instead of uvicorn (better for managing multiple workers)",
)
@click.option(
    "--run_hypercorn",
    default=False,
    is_flag=True,
    help="Starts proxy via hypercorn, instead of uvicorn (supports HTTP/2)",
)
@click.option(
    "--run_granian",
    default=False,
    is_flag=True,
    help=(
        "Starts proxy via Granian (Rust ASGI server) instead of uvicorn. "
        "Requires Python 3.10+ and the `granian` package."
    ),
)
@click.option(
    "--ssl_keyfile_path",
    default=None,
    type=str,
    help="Path to the SSL keyfile. Use this when you want to provide SSL certificate when starting proxy",
    envvar="SSL_KEYFILE_PATH",
)
@click.option(
    "--ssl_certfile_path",
    default=None,
    type=str,
    help="Path to the SSL certfile. Use this when you want to provide SSL certificate when starting proxy",
    envvar="SSL_CERTFILE_PATH",
)
@click.option(
    "--ciphers",
    default=None,
    type=str,
    help="Ciphers to use for the SSL setup.",
)
@click.option(
    "--use_prisma_db_push",
    is_flag=True,
    default=False,
    help="Use prisma db push instead of prisma migrate for database schema updates",
)
@click.option("--local", is_flag=True, default=False, help="no-op, kept for backwards compatibility")
@click.option(
    "--skip_server_startup",
    is_flag=True,
    default=False,
    help="Skip starting the server after setup (useful for migrations only)",
)
@click.option(
    "--validate_config",
    is_flag=True,
    default=False,
    help="Load and validate the config file (including mcp_servers) without starting the server, then exit. Exit code 1 on any config error.",
)
@click.option(
    "--check_migrations_against_existing_data",
    is_flag=True,
    default=False,
    help="Dry-run this version's pending migrations on DATABASE_URL: the real SQL inside one transaction that is "
    "rolled back, so Postgres reports the row or schema that would make the upgrade fail, then exit. Holds the same "
    "locks the upgrade will. Exit code 2 when a migration fails.",
)
@click.option(
    "--keepalive_timeout",
    default=None,
    type=int,
    help="Set the uvicorn keepalive timeout in seconds (uvicorn timeout_keep_alive parameter)",
    envvar="KEEPALIVE_TIMEOUT",
)
@click.option(
    "--timeout_worker_healthcheck",
    default=None,
    type=int,
    help=(
        "Set the uvicorn worker health-check timeout in seconds (uvicorn timeout_worker_healthcheck parameter). "
        "Requires uvicorn>=0.37.0. Only applies when running uvicorn directly with --num_workers>1; "
        "ignored under --run_gunicorn / --run_hypercorn."
    ),
    envvar="TIMEOUT_WORKER_HEALTHCHECK",
)
@click.option(
    "--max_requests_before_restart",
    default=None,
    type=int,
    help=(
        "Recycle each worker after it has served this many HTTP requests, health probes included "
        "(uvicorn: limit_max_requests, gunicorn: max_requests). Only safe with --num_workers 2 or more. "
        "With a single worker the whole proxy stops serving when the limit is hit, uvicorn exits the "
        "process and gunicorn accepts no connections until the replacement worker has booted."
    ),
    envvar="MAX_REQUESTS_BEFORE_RESTART",
)
@click.option(
    "--max_requests_before_restart_jitter",
    default=None,
    type=int,
    help=(
        "Stagger worker restarts by adding a random amount in [0, jitter] to "
        "--max_requests_before_restart so workers do not recycle at the same time "
        "(uvicorn: limit_max_requests_jitter, requires uvicorn>=0.41.0; gunicorn: max_requests_jitter). "
        "Has no effect without --max_requests_before_restart."
    ),
    envvar="MAX_REQUESTS_BEFORE_RESTART_JITTER",
)
@click.option(
    "--limit_concurrency",
    default=None,
    type=click.IntRange(min=1),
    help=(
        "Set uvicorn's concurrency limit. Uvicorn counts both active tasks and "
        "accepted connections and returns HTTP 503 after the limit is reached. "
        "Idle connections can consume capacity, so use upstream connection/header "
        "timeouts and per-client connection limits. Only applies to uvicorn "
        "(ignored under --run_gunicorn / --run_hypercorn / --run_granian)."
    ),
    envvar="LIMIT_CONCURRENCY",
)
@click.option(
    "--enforce_prisma_migration_check",
    is_flag=True,
    default=False,
    hidden=True,
    help=(
        "Deprecated and ignored: the proxy always exits when database setup fails at "
        "startup. It is still accepted so existing commands keep working."
    ),
)
@click.option(
    "--use_v2_migration_resolver",
    is_flag=True,
    default=False,
    help=(
        "Deprecated and ignored: the v2 migration resolver is now the default, "
        "so this flag has no effect. It is still accepted so existing commands "
        "keep working. Pass --use_legacy_migration_resolver, or set "
        "USE_V2_MIGRATION_RESOLVER=false, to opt back into v1."
    ),
    envvar="USE_V2_MIGRATION_RESOLVER",
)
@click.option(
    "--use_legacy_migration_resolver",
    is_flag=True,
    default=False,
    help=(
        "Fall back to the legacy v1 migration resolver. By default the proxy "
        "uses the v2 resolver, which avoids the diff-and-force recovery path "
        "that can cause schema thrashing during rolling deploys where two "
        "LiteLLM versions contend for the same DB."
    ),
)
@click.option(
    "--reload",
    is_flag=True,
    default=False,
    help="Enable uvicorn hot reload (dev only), including every --num_workers worker on uvicorn>=0.50.0 outside Windows. Also reloads when the --config YAML or .env changes. Incompatible with --run_gunicorn, --run_hypercorn, and --run_granian.",
)
@click.option(
    "--prometheus_metrics_port",
    default=None,
    type=click.IntRange(min=1, max=65535),
    help=(
        "Serve Prometheus /metrics from a separate process on this port (bound to --host) so scraping and "
        "multi-worker aggregation never run on an inference worker's event loop. Samples appear once the "
        "`prometheus` callback is enabled (config.yaml or DB). /metrics stays mounted on the main port as well; "
        "the separate port has no virtual-key auth, so keep it off public ingress. Startup fails if the metrics "
        "server cannot bind."
    ),
    envvar="PROMETHEUS_METRICS_PORT",
)
def run_server(
    cli_args,
    host,
    port,
    api_base,
    api_version,
    model,
    alias,
    add_key,
    headers,
    save,
    debug,
    detailed_debug,
    temperature,
    max_tokens,
    request_timeout,
    drop_params,
    add_function_to_prompt,
    config,
    max_budget,
    test,
    local,
    num_workers: int,
    granian_threads,
    test_async,
    iam_token_db_auth,
    azure_postgresql_auth: bool,
    num_requests,
    use_queue,
    health,
    setup,
    version,
    run_gunicorn,
    run_hypercorn,
    run_granian,
    ssl_keyfile_path,
    ssl_certfile_path,
    ciphers,
    log_config,
    use_prisma_db_push: bool,
    skip_server_startup,
    validate_config: bool,
    check_migrations_against_existing_data: bool,
    keepalive_timeout,
    timeout_worker_healthcheck,
    max_requests_before_restart: int | None,
    max_requests_before_restart_jitter: int | None,
    limit_concurrency: int | None,
    enforce_prisma_migration_check: bool,
    use_v2_migration_resolver: bool,
    use_legacy_migration_resolver: bool,
    reload: bool,
    prometheus_metrics_port: int | None,
):
    if cli_args:
        if cli_args == ("xai-oauth", "login"):
            from litellm.llms.xai.oauth import XAIOAuthAuthenticator

            authenticator: Final = XAIOAuthAuthenticator()
            auth_data: Final = authenticator.login()
            click.echo(f"xAI OAuth login successful. Credentials saved to {authenticator.auth_file}.")
            if auth_data.get("expires_at"):
                click.echo(f"Access token expires at {auth_data['expires_at']}.")
            return
        raise click.UsageError(f"Unknown command: {' '.join(cli_args)}")

    if setup:
        from litellm.setup_wizard import run_setup_wizard

        run_setup_wizard()
        return

    args: Final = locals()
    try:
        from litellm.proxy.proxy_server import (
            KeyManagementSettings,
            ProxyConfig,
            app,
            save_worker_config,
        )
    except ModuleNotFoundError as e:
        raise ModuleNotFoundError(f"Missing dependency {e}. Run `pip install 'litellm[proxy]'`") from e
    if version is True:
        ProxyInitializationHelpers.echo_litellm_version()
        return
    if validate_config is True:
        ProxyInitializationHelpers.run_config_validation(config)
        return
    if check_migrations_against_existing_data is True:
        sys.exit(ProxyInitializationHelpers.run_migration_data_check(use_migrate=not use_prisma_db_push))
    held_prisma_notice: Final = (
        _print_or_hold_enforce_prisma_migration_check_notice(
            reads_config_file=config is not None, returns_before_config=health is True or test is True
        )
        if enforce_prisma_migration_check
        else None
    )
    if model and "ollama" in model and api_base is None:
        ProxyInitializationHelpers.run_ollama_serve()
    if health is True:
        ProxyInitializationHelpers.run_health_check(host, port)
        return
    if test is True:
        ProxyInitializationHelpers.run_test_chat_completion(host, port, model, test)
        return
    else:
        if headers:
            headers = json.loads(headers)
        save_worker_config(
            model=model,
            alias=alias,
            api_base=api_base,
            api_version=api_version,
            debug=debug,
            detailed_debug=detailed_debug,
            temperature=temperature,
            max_tokens=max_tokens,
            request_timeout=request_timeout,
            max_budget=max_budget,
            drop_params=drop_params,
            add_function_to_prompt=add_function_to_prompt,
            headers=headers,
            save=save,
            config=config,
            use_queue=use_queue,
        )
        if run_granian:
            try:
                import granian  # noqa: F401
            except ImportError as e:
                raise ImportError(
                    "granian must be installed to use --run_granian. "
                    "Run `pip install granian` or `pip install 'litellm[proxy]'` "
                    "(Granian requires Python 3.10+)."
                ) from e
        else:
            try:
                import uvicorn
            except Exception:
                raise ImportError("uvicorn, gunicorn needs to be imported. Run - `pip install 'litellm[proxy]'`")

        db_connection_pool_limit = 100
        # Starts optional due to config fallback checks; guaranteed non-None before use.
        db_connection_timeout: int | float | None = 60
        db_connect_timeout: int | float | None = None
        db_socket_timeout: int | float | None = None
        db_disable_prepared_statements: bool = False
        db_extra_connection_params: dict | None = None
        db_statement_timeout: float | None = None
        db_lock_timeout: float | None = None
        db_idle_in_transaction_timeout: float | None = None
        general_settings = {}
        ### GET DB TOKEN FOR RDS IAM / AZURE ENTRA AUTH ###

        from litellm.proxy.db.db_url_settings import DatabaseURLSettings
        from litellm.proxy.db.token_auth import (
            AZURE_POSTGRESQL_AUTH_ENV_VAR,
            IAM_TOKEN_DB_AUTH_ENV_VAR,
            resolve_database_token_auth,
            token_auth_flag_enabled,
        )

        wants_rds_iam: Final = iam_token_db_auth or token_auth_flag_enabled(
            os.getenv(IAM_TOKEN_DB_AUTH_ENV_VAR), env_var=IAM_TOKEN_DB_AUTH_ENV_VAR
        )
        wants_azure_entra: Final = azure_postgresql_auth or token_auth_flag_enabled(
            os.getenv(AZURE_POSTGRESQL_AUTH_ENV_VAR), env_var=AZURE_POSTGRESQL_AUTH_ENV_VAR
        )
        if wants_rds_iam:
            os.environ[IAM_TOKEN_DB_AUTH_ENV_VAR] = "True"
        if wants_azure_entra:
            os.environ[AZURE_POSTGRESQL_AUTH_ENV_VAR] = "True"
        if wants_rds_iam or wants_azure_entra:
            DatabaseURLSettings.from_env().apply_writer_url_to_env()

        ### DECRYPT ENV VAR ###

        from litellm.secret_managers.aws_secret_manager import decrypt_env_var

        if os.getenv("USE_AWS_KMS", None) is not None and os.getenv("USE_AWS_KMS") == "True":
            ## V2 IMPLEMENTATION OF AWS KMS - USER WANTS TO DECRYPT MULTIPLE KEYS IN THEIR ENV
            new_env_var: Final = decrypt_env_var()

            for k, v in new_env_var.items():
                os.environ[k] = v

        litellm_settings = None
        env_config_file: Final = os.environ.get("CONFIG_FILE_PATH")
        if (
            env_config_file
            and os.environ.get("LITELLM_CONFIG_BUCKET_NAME") is None
            and ProxyConfig().is_yaml(config_file_path=env_config_file)
        ):
            import asyncio

            _apply_config_suppress_startup_output(
                _CONFIG_STARTUP_OUTPUT_SETTINGS.validate_python(
                    asyncio.run(ProxyConfig().read_config_file(config_file_path=env_config_file))
                ).get("litellm_settings")
            )
        if config is not None:
            """
            Allow user to pass in db url via config

            read from there and save it to os.env['DATABASE_URL']
            """
            try:
                import asyncio

            except Exception:
                raise ImportError("yaml needs to be imported. Run - `pip install 'litellm[proxy]'`")

            proxy_config: Final = ProxyConfig()
            _config: Final = asyncio.run(proxy_config.get_config(config_file_path=config))

            ### LITELLM SETTINGS ###
            litellm_settings = _config.get("litellm_settings", None)
            _apply_config_logging_settings(
                _CONFIG_STARTUP_OUTPUT_SETTINGS.validate_python(_config).get("litellm_settings")
            )
            if held_prisma_notice is not None:
                held_prisma_notice()
            ### GENERAL SETTINGS ###
            general_settings = _config.get("general_settings", {})
            if general_settings is None:
                general_settings = {}
            ### LOAD KEY MANAGEMENT SETTINGS FIRST (needed for custom secret manager) ###
            key_management_settings: Final = general_settings.get("key_management_settings", None)
            if key_management_settings is not None:
                import litellm

                litellm._key_management_settings = KeyManagementSettings(**key_management_settings)

            if general_settings:
                ### LOAD SECRET MANAGER ###
                key_management_system: Final = general_settings.get("key_management_system", None)
                proxy_config.initialize_secret_manager(
                    key_management_system=key_management_system, config_file_path=config
                )
            database_url = general_settings.get("database_url", None)
            if database_url is None and os.getenv("DATABASE_URL") is None:
                # Use helper function to construct DATABASE_URL from individual variables
                from litellm.proxy.utils import construct_database_url_from_env_vars

                database_url = construct_database_url_from_env_vars()
                if database_url:
                    os.environ["DATABASE_URL"] = database_url
            db_connection_pool_limit = general_settings.get(
                "database_connection_pool_limit",
                LiteLLMDatabaseConnectionPool.database_connection_pool_limit.value,
            )
            db_connection_timeout = general_settings.get("database_connection_timeout")
            if db_connection_timeout is None:
                db_connection_timeout = general_settings.get("database_connection_pool_timeout")
            if db_connection_timeout is None:
                db_connection_timeout = LiteLLMDatabaseConnectionPool.database_connection_pool_timeout.value
            db_connect_timeout = general_settings.get("database_connect_timeout")
            db_socket_timeout = general_settings.get("database_socket_timeout")
            _disable_prepared_statements: Final = general_settings.get("database_disable_prepared_statements", False)
            if isinstance(_disable_prepared_statements, str):
                from litellm.secret_managers.main import str_to_bool

                db_disable_prepared_statements = str_to_bool(_disable_prepared_statements) is True
            else:
                db_disable_prepared_statements = bool(_disable_prepared_statements)
            db_extra_connection_params = general_settings.get("database_extra_connection_params")
            db_timeouts: Final = DatabaseTimeoutSettings.model_validate(general_settings)
            db_statement_timeout = db_timeouts.database_statement_timeout
            db_lock_timeout = db_timeouts.database_lock_timeout
            db_idle_in_transaction_timeout = db_timeouts.database_idle_in_transaction_session_timeout
            if database_url and database_url.startswith("os.environ/"):
                original_dir: Final = os.getcwd()
                # set the working directory to where this script is
                sys.path.insert(
                    0, os.path.abspath("../..")
                )  # Adds the parent directory to the system path - for litellm local dev
                import litellm
                from litellm import get_secret_str

                database_url = get_secret_str(database_url, default_value=None)
                os.chdir(original_dir)
            if database_url is not None and isinstance(database_url, str):
                os.environ["DATABASE_URL"] = database_url

        # Handle database URL construction when no config file is used
        if config is None and os.getenv("DATABASE_URL") is None:
            # Use helper function to construct DATABASE_URL from individual variables
            from litellm.proxy.utils import construct_database_url_from_env_vars

            database_url = construct_database_url_from_env_vars()
            if database_url:
                os.environ["DATABASE_URL"] = database_url

        # Set default values for connection pool settings when no config is used
        if config is None:
            db_connection_pool_limit = LiteLLMDatabaseConnectionPool.database_connection_pool_limit.value
            db_connection_timeout = LiteLLMDatabaseConnectionPool.database_connection_pool_timeout.value

        if os.getenv("DATABASE_URL", None) is not None or os.getenv("DIRECT_URL", None) is not None:
            from litellm.proxy.db.db_url_settings import (
                DISABLE_PREPARED_STATEMENTS_ENV_VAR,
                add_missing_query_params,
                idle_lifetime_params,
                postgres_connection_budget_message,
                reader_shareable_params,
                translate_libpq_ssl_params,
                unsupported_db_scheme,
                unsupported_db_scheme_message,
            )

            for _db_env in ("DATABASE_URL", "DIRECT_URL"):
                _candidate_url = os.getenv(_db_env)
                if _candidate_url is None:
                    continue
                _bad_scheme = unsupported_db_scheme(_candidate_url)
                if _bad_scheme is not None:
                    print(
                        terminal_text(
                            f"\033[1;31mLiteLLM Proxy: {unsupported_db_scheme_message(_db_env, _bad_scheme)}\033[0m"
                        ),
                        file=sys.stderr,
                        flush=True,
                    )
                    sys.exit(1)
            from litellm.secret_managers.main import get_secret

            env_disable_prepared_statements: Final = token_auth_flag_enabled(
                os.getenv(DISABLE_PREPARED_STATEMENTS_ENV_VAR), env_var=DISABLE_PREPARED_STATEMENTS_ENV_VAR
            )
            disable_prepared_statements: Final = db_disable_prepared_statements or env_disable_prepared_statements
            connection_url_params: Final = _build_db_connection_url_params(
                connection_limit=db_connection_pool_limit,
                pool_timeout=db_connection_timeout,
                connect_timeout=db_connect_timeout,
                socket_timeout=db_socket_timeout,
                disable_prepared_statements=disable_prepared_statements,
                extra_params=db_extra_connection_params,
            )
            lifetime_params: Final = idle_lifetime_params(general_settings.get("database_max_idle_connection_lifetime"))
            if os.getenv("DATABASE_URL", None) is not None:
                database_url = get_secret("DATABASE_URL", default_value=None)
                resolved_url: Final[str | None] = str(database_url) if database_url else None
                pg_options: Final[str] = _pg_options_with_timeouts(
                    _url_query_value(resolved_url, "options"),
                    db_statement_timeout,
                    db_lock_timeout,
                    db_idle_in_transaction_timeout,
                )
                writer_url: Final = (
                    _with_query_value(resolved_url, "options", pg_options)
                    if resolved_url and pg_options
                    else resolved_url
                )
                modified_url = append_query_params(
                    writer_url,
                    connection_url_params,
                )
                os.environ["DATABASE_URL"] = translate_libpq_ssl_params(
                    add_missing_query_params(modified_url, lifetime_params)
                )
            if os.getenv("DIRECT_URL", None) is not None:
                database_url = os.getenv("DIRECT_URL")
                modified_url = append_query_params(database_url, connection_url_params)
                os.environ["DIRECT_URL"] = translate_libpq_ssl_params(
                    add_missing_query_params(modified_url, lifetime_params)
                )
            # The reader pool is a real pool against the same configured cap, so it
            # gets the allowlisted pool params. Schema-affecting ones, including any
            # the operator smuggled in through database_extra_connection_params, stay
            # on the writer. Anything pinned on the replica URL wins, unlike the
            # writer where the config is applied on top.
            read_replica_url: Final[str | None] = os.getenv("DATABASE_URL_READ_REPLICA")
            if read_replica_url:
                reader_options: Final[str] = _pg_options_with_timeouts(
                    _url_query_value(read_replica_url, "options"),
                    db_statement_timeout,
                    db_lock_timeout,
                    db_idle_in_transaction_timeout,
                )
                os.environ["DATABASE_URL_READ_REPLICA"] = translate_libpq_ssl_params(
                    add_missing_query_params(
                        add_missing_query_params(
                            _with_query_value(read_replica_url, "options", reader_options)
                            if reader_options
                            else read_replica_url,
                            reader_shareable_params(connection_url_params),
                        ),
                        lifetime_params,
                    )
                )
            print_startup_message(
                postgres_connection_budget_message(
                    writer_limit=_url_query_value(os.getenv("DATABASE_URL"), "connection_limit")
                    or f"{db_connection_pool_limit}",
                    reader_limit=(
                        _url_query_value(os.getenv("DATABASE_URL_READ_REPLICA"), "connection_limit")
                        or f"{db_connection_pool_limit}"
                    )
                    if read_replica_url
                    else None,
                    num_workers=f"{1 if run_hypercorn else num_workers}",
                )
            )
            from litellm_proxy_extras.prisma_toolchain import prisma_cli_available

            is_prisma_runnable: Final = prisma_cli_available()

            if is_prisma_runnable:
                from litellm.proxy.db.check_migration import check_prisma_schema_diff
                from litellm.proxy.db.prisma_client import (
                    PrismaManager,
                    should_update_prisma_schema,
                )

                if should_update_prisma_schema(general_settings.get("disable_prisma_schema_update")) is False:
                    check_prisma_schema_diff(db_url=None)
                else:
                    use_v2_resolver: Final = resolve_v2_migration_resolver(
                        use_legacy_flag=use_legacy_migration_resolver,
                        env_value=os.getenv("USE_V2_MIGRATION_RESOLVER"),
                    )
                    if deprecated_v2_flag_passed_on_cli() and use_v2_resolver:
                        print_startup_warning(
                            "\033[1;33mLiteLLM Proxy: --use_v2_migration_resolver is "
                            "deprecated and has no effect, because the v2 migration "
                            "resolver is now the default. You can safely remove it. To "
                            "opt back into the legacy v1 resolver, pass "
                            "--use_legacy_migration_resolver.\033[0m",
                        )
                    if not use_v2_resolver:
                        print_startup_warning(
                            "\033[1;33mLiteLLM Proxy: Using the legacy (v1) migration "
                            "resolver. It performs the diff-and-force recovery that can "
                            "cause schema thrashing during rolling deploys where two "
                            "LiteLLM versions contend for the same DB.\033[0m",
                        )
                    try:
                        migrated: Final = PrismaManager.setup_database(
                            use_migrate=not use_prisma_db_push,
                            use_v2_resolver=use_v2_resolver,
                        )
                        setup_ok: Final = migrated and (
                            not skip_server_startup or PrismaManager.build_request_log_indexes()
                        )
                        if migrated and not skip_server_startup:
                            PrismaManager.start_request_log_index_build()
                    except RuntimeError as e:
                        # Raised on unrecoverable migration errors: the v2
                        # resolver's non-idempotent failures and permission
                        # issues, and any `prisma db push` against a
                        # partitioned LiteLLM_SpendLogs.
                        print(
                            terminal_text(f"\033[1;31mLiteLLM Proxy: Database migration cannot proceed. {e}\033[0m"),
                            file=sys.stderr,
                            flush=True,
                        )
                        sys.exit(2)
                    if not setup_ok:
                        print(
                            terminal_text(
                                "\033[1;31mLiteLLM Proxy: Database setup failed after multiple retries. "
                                "The proxy cannot start safely. Please check your database connection and migration status.\033[0m"
                            )
                        )
                        sys.exit(1)
            else:
                print(
                    terminal_text(
                        "\033[1;31mLiteLLM Proxy: a database URL is set but the prisma CLI is neither on PATH nor importable "
                        "as a package, so the database cannot be set up. Install it with `pip install 'litellm[extra_proxy]'` "
                        "or run a shipped LiteLLM image.\033[0m"
                    )
                )
                sys.exit(1)
        pgbouncer_settings: Final = PgBouncerSettings()
        upstream_database_url: Final = os.getenv("DATABASE_URL")
        if pgbouncer_settings.enabled and upstream_database_url is not None:
            pooled_database_url: Final = start_in_container_pgbouncer(
                pgbouncer_settings, upstream_database_url, token_auth=resolve_database_token_auth()
            )
            if isinstance(pooled_database_url, PgBouncerError):
                print(
                    terminal_text(
                        f"\033[1;31mLiteLLM Proxy: LITELLM_PGBOUNCER_ENABLED is set but the in-container pgbouncer "
                        f"could not start: {pooled_database_url.reason}\033[0m"
                    ),
                    file=sys.stderr,
                    flush=True,
                )
                sys.exit(1)
            export_pooled_database_url(pooled_database_url)
        if port == 4000 and ProxyInitializationHelpers.is_port_in_use(port):
            port = random.randint(1024, 49152)
        if prometheus_metrics_port == port:
            raise click.UsageError("--prometheus_metrics_port must differ from --port")

        import litellm

        if detailed_debug is True:
            litellm.turn_on_debug()

        # DO NOT DELETE - enables global variables to work across files
        from litellm.proxy.proxy_server import app

        os.environ["NUM_WORKERS"] = str(num_workers)

        # Skip server startup if requested (after all setup is done)
        if skip_server_startup:
            print_startup_message("LiteLLM: Setup complete. Skipping server startup as requested.")
            return

        # Auto-create PROMETHEUS_MULTIPROC_DIR for multi-worker setups
        prometheus_multiproc_dir: Final = ProxyInitializationHelpers.maybe_setup_prometheus_multiproc_dir(
            num_workers=num_workers,
            litellm_settings=litellm_settings if config else None,
            prometheus_metrics_port=prometheus_metrics_port,
        )

        if prometheus_metrics_port is not None and prometheus_multiproc_dir is not None:
            from litellm.proxy.prometheus_metrics_server import MetricsServerStartupError, start_metrics_server_process

            try:
                metrics_process: Final = start_metrics_server_process(
                    host=host, port=prometheus_metrics_port, multiproc_dir=prometheus_multiproc_dir
                )
            except MetricsServerStartupError as error:
                raise click.ClickException(str(error)) from error
            print_startup_message(
                f"\033[1;32mLiteLLM: Serving Prometheus metrics on {host}:{prometheus_metrics_port}/metrics "
                f"(pid {metrics_process.pid})\033[0m"
            )

        running_uvicorn: Final = run_gunicorn is False and run_hypercorn is False
        uvicorn_args: Final = ProxyInitializationHelpers.get_default_unvicorn_init_args(
            host=host,
            port=port,
            log_config=log_config,
            keepalive_timeout=keepalive_timeout,
            timeout_worker_healthcheck=(timeout_worker_healthcheck if running_uvicorn else None),
        )
        # Optional: recycle uvicorn workers after N requests
        if max_requests_before_restart is not None:
            uvicorn_args["limit_max_requests"] = max_requests_before_restart
        if run_hypercorn is False and run_granian is False:
            single_worker_warning: Final = ProxyInitializationHelpers.single_worker_max_requests_warning(
                num_workers=num_workers,
                max_requests_before_restart=max_requests_before_restart,
                run_gunicorn=run_gunicorn is True,
            )
            if single_worker_warning is not None:
                print_startup_warning(f"\033[1;33m{single_worker_warning}\033[0m\n")
        if run_gunicorn is False and run_hypercorn is False and run_granian is False:
            if limit_concurrency is not None:
                uvicorn_args["limit_concurrency"] = limit_concurrency
            if max_requests_before_restart_jitter is not None:
                ProxyInitializationHelpers.apply_uvicorn_max_requests_jitter(
                    uvicorn_args=uvicorn_args,
                    max_requests_before_restart=max_requests_before_restart,
                    jitter=max_requests_before_restart_jitter,
                )
            if ssl_certfile_path is not None and ssl_keyfile_path is not None:
                print_startup_message(
                    f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
                )
                uvicorn_args["ssl_keyfile"] = ssl_keyfile_path
                uvicorn_args["ssl_certfile"] = ssl_certfile_path

            loop_type: Final = ProxyInitializationHelpers.get_loop_type()
            if loop_type:
                uvicorn_args["loop"] = loop_type

            if reload:
                ProxyInitializationHelpers.configure_dev_reload(uvicorn_args, config)

            if num_workers > 1:
                start_query_engine_reaper()
            if num_workers > 1 or reload:
                worker_overrides: Final = ProxyInitializationHelpers.child_process_uvicorn_overrides(
                    uvicorn_args.get("log_config")
                )
                if reload and num_workers > 1:
                    from litellm.proxy.dev_reload import (
                        RELOAD_OPTIONS,
                        multi_worker_reload_supported,
                        run_multi_worker_reload,
                    )

                    if multi_worker_reload_supported():
                        run_multi_worker_reload(
                            RELOAD_OPTIONS.validate_python(uvicorn_args),
                            workers=TypeAdapter(int).validate_python(num_workers),
                            worker_overrides=worker_overrides,
                        )
                        return
                    print(
                        "\033[1;33mLiteLLM Proxy: --reload with --num_workers>1 requires uvicorn>=0.50.0 "
                        "and an OS other than Windows, so only a single worker reloads.\033[0m"
                    )
                uvicorn_args.update(worker_overrides)
            uvicorn.run(
                **uvicorn_args,
                workers=num_workers,
            )
        elif run_gunicorn is True:
            ProxyInitializationHelpers.run_gunicorn_server(
                host=host,
                port=port,
                app=app,
                num_workers=num_workers,
                ssl_certfile_path=ssl_certfile_path,
                ssl_keyfile_path=ssl_keyfile_path,
                max_requests_before_restart=max_requests_before_restart,
                max_requests_before_restart_jitter=max_requests_before_restart_jitter,
            )
        elif run_hypercorn is True:
            ProxyInitializationHelpers.init_hypercorn_server(
                app=app,
                host=host,
                port=port,
                ssl_certfile_path=ssl_certfile_path,
                ssl_keyfile_path=ssl_keyfile_path,
                ciphers=ciphers,
            )
        elif run_granian is True:
            ProxyInitializationHelpers.init_granian_server(
                host=host,
                port=port,
                num_workers=num_workers,
                ssl_certfile_path=ssl_certfile_path,
                ssl_keyfile_path=ssl_keyfile_path,
                max_requests_before_restart=max_requests_before_restart,
                ciphers=ciphers,
                granian_runtime_threads=granian_threads,
            )


if __name__ == "__main__":
    run_server()