From 165f061b81f928e4414d77af4dea31c64ff093b2 Mon Sep 17 00:00:00 2001
From: yassin <yassin@berri.ai>
Date: Tue, 13 Oct 2026 21:30:40 +0000
Subject: [PATCH] feat(control_plane): --control-plane flag serves the control
 plane API from the existing image

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 litellm/_logging.py                         |   2 +-
 litellm/proxy/control_plane/main.py         | 142 ++++
 litellm/proxy/proxy_cli.py                  | 104 ++-
 tests/unit/proxy/control_plane/test_main.py | 237 ++++++
 tests/unit/proxy/test_proxy_cli.py          | 804 +++++++++-----------
 5 files changed, 827 insertions(+), 462 deletions(-)
 create mode 100644 litellm/proxy/control_plane/main.py
 create mode 100644 tests/unit/proxy/control_plane/test_main.py

diff --git a/litellm/_logging.py b/litellm/_logging.py
index 6a38eea2b12..70c5f477d28 100644
--- a/litellm/_logging.py
+++ b/litellm/_logging.py
@@ -1122,7 +1122,7 @@ def _initialize_loggers_with_handler(handler: logging.Handler):
         logging.getLogger().setLevel(resolve_log_level(configured_level))
 
 
-def _get_uvicorn_json_log_config():
+def _get_uvicorn_json_log_config() -> Mapping[str, object]:
     """
     Generate a uvicorn log_config dictionary that applies JSON formatting to all loggers.
 
diff --git a/litellm/proxy/control_plane/main.py b/litellm/proxy/control_plane/main.py
new file mode 100644
index 00000000000..db8ec39c7de
--- /dev/null
+++ b/litellm/proxy/control_plane/main.py
@@ -0,0 +1,142 @@
+import os
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Final, Protocol
+
+from fastapi import FastAPI
+from typing_extensions import ReadOnly, TypedDict
+
+from litellm.proxy.auth.litellm_license import LicenseCheck
+from litellm.proxy.control_plane.app import License, create_app
+from litellm.proxy.control_plane.container import build_container
+from litellm.proxy.control_plane.settings import ControlPlaneSettings, load_settings
+
+
+@dataclass(frozen=True, slots=True)
+class Tls:
+    keyfile: str
+    certfile: str
+
+
+class _LogConfigKwargs(TypedDict, total=False):
+    log_config: ReadOnly[str | dict[str, object]]  # mutable-ok: Uvicorn accepts dict log configs
+    log_level: ReadOnly[int]
+
+
+class _UvicornRunner(Protocol):
+    def __call__(
+        self,
+        *,
+        app: FastAPI,
+        host: str = ...,
+        port: int = ...,
+        server_header: bool = ...,
+        timeout_graceful_shutdown: int | None = ...,
+        timeout_keep_alive: int = ...,
+        h11_max_incomplete_event_size: int | None = ...,
+        ssl_keyfile: str | None = ...,
+        ssl_certfile: str | None = ...,
+        limit_concurrency: int | None = ...,
+        log_config: str | dict[str, object] | None = ...,  # mutable-ok: Uvicorn accepts dict log configs
+        log_level: int | None = ...,
+    ) -> None: ...
+
+
+class Runner(Protocol):
+    def __call__(
+        self,
+        app: FastAPI,
+        *,
+        host: str,
+        port: int,
+        tls: Tls | None,
+        limit_concurrency: int | None,
+        log_config: str | None,
+        keepalive_timeout: int | None,
+        max_request_header_size: int | None,
+    ) -> None: ...
+
+
+def _uvicorn(
+    app: FastAPI,
+    *,
+    host: str,
+    port: int,
+    tls: Tls | None,
+    limit_concurrency: int | None,
+    log_config: str | None,
+    keepalive_timeout: int | None,
+    max_request_header_size: int | None,
+    uvicorn_runner: _UvicornRunner | None = None,
+    json_logs: bool | None = None,
+    environ: Mapping[str, str] | None = None,
+) -> None:
+    import uvicorn
+
+    import litellm
+    from litellm._logging import (
+        _get_uvicorn_json_log_config,  # pyright: ignore[reportPrivateUsage]  # shared proxy logging config
+        resolve_log_level,
+    )
+    from litellm.constants import DEFAULT_MAX_REQUEST_HEADER_SIZE
+    from litellm.proxy.shutdown.graceful_shutdown_manager import GracefulShutdownManager
+
+    keyfile: Final = None if tls is None else tls.keyfile
+    certfile: Final = None if tls is None else tls.certfile
+    json_logs_enabled: Final = litellm.json_logs if json_logs is None else json_logs
+    environment: Final = os.environ if environ is None else environ
+    litellm_log: Final = environment.get("LITELLM_LOG")
+    log_config_kwargs: Final[_LogConfigKwargs] = (
+        {"log_config": log_config}
+        if log_config is not None
+        else _LogConfigKwargs(log_config=dict(_get_uvicorn_json_log_config()))
+        if json_logs_enabled
+        else {"log_level": resolve_log_level(litellm_log)}
+        if litellm_log
+        else {}
+    )
+    runner: Final[_UvicornRunner] = uvicorn.run if uvicorn_runner is None else uvicorn_runner
+    runner(
+        app=app,
+        host=host,
+        port=port,
+        server_header=False,
+        timeout_graceful_shutdown=round(GracefulShutdownManager.get_timeout()),
+        **log_config_kwargs,
+        timeout_keep_alive=5 if keepalive_timeout is None else keepalive_timeout,
+        h11_max_incomplete_event_size=(
+            DEFAULT_MAX_REQUEST_HEADER_SIZE if max_request_header_size is None else max_request_header_size
+        ),
+        ssl_keyfile=keyfile,
+        ssl_certfile=certfile,
+        limit_concurrency=limit_concurrency,
+    )
+
+
+def build_app(settings: ControlPlaneSettings, license_check: License) -> FastAPI:
+    return create_app(build_container(settings), license_check)
+
+
+def serve(
+    host: str,
+    port: int,
+    tls: Tls | None = None,
+    settings: Callable[[], ControlPlaneSettings] = load_settings,
+    license_check: Callable[[], License] = LicenseCheck,
+    run: Runner = _uvicorn,
+    limit_concurrency: int | None = None,
+    log_config: str | None = None,
+    keepalive_timeout: int | None = None,
+    max_request_header_size: int | None = None,
+) -> None:
+    app: Final = build_app(settings(), license_check())
+    run(
+        app,
+        host=host,
+        port=port,
+        tls=tls,
+        limit_concurrency=limit_concurrency,
+        log_config=log_config,
+        keepalive_timeout=keepalive_timeout,
+        max_request_header_size=max_request_header_size,
+    )
diff --git a/litellm/proxy/proxy_cli.py b/litellm/proxy/proxy_cli.py
index 0c89627e74d..7e93693d90e 100644
--- a/litellm/proxy/proxy_cli.py
+++ b/litellm/proxy/proxy_cli.py
@@ -10,9 +10,10 @@ import subprocess
 import sys
 import urllib.parse as urlparse
 from collections.abc import Callable, Iterable, Mapping, Sequence
+from dataclasses import dataclass
 from pathlib import Path
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Annotated, Any, Final
+from typing import TYPE_CHECKING, Annotated, Any, Final, Protocol
 
 import click
 import httpx
@@ -35,6 +36,8 @@ from litellm.types.llms.base import LiteLLMBaseModel
 
 if TYPE_CHECKING:
     from fastapi import FastAPI
+
+    from litellm.proxy.control_plane.main import Tls
 else:
     FastAPI = Any
 
@@ -888,6 +891,50 @@ class ProxyInitializationHelpers:
     _maybe_setup_prometheus_multiproc_dir = maybe_setup_prometheus_multiproc_dir
 
 
+def control_plane_requested(control_plane: bool, skip_server_startup: bool, utility_command: bool = False) -> bool:
+    return control_plane and not skip_server_startup and not utility_command
+
+
+class _ControlPlaneLicense(Protocol):
+    def is_premium(self) -> bool: ...
+
+
+def _decrypt_env_var() -> Mapping[str, str]:
+    from litellm.secret_managers.aws_secret_manager import decrypt_env_var
+
+    return decrypt_env_var()
+
+
+def _control_plane_license_check() -> _ControlPlaneLicense:
+    from litellm.proxy.auth.litellm_license import LicenseCheck
+
+    return LicenseCheck()
+
+
+@dataclass(frozen=True, slots=True)
+class CliDependencies:
+    decrypt_env: Callable[[], Mapping[str, str]] = _decrypt_env_var
+    license_check: Callable[[], _ControlPlaneLicense] = _control_plane_license_check
+
+
+def _decrypt_kms_env_vars(decrypt_env: Callable[[], Mapping[str, str]]) -> None:
+    if os.getenv("USE_AWS_KMS") != "True":
+        return
+
+    for key, value in decrypt_env().items():
+        os.environ[key] = value
+
+
+def control_plane_tls(ssl_keyfile_path: str | None, ssl_certfile_path: str | None) -> "Tls | None":
+    from litellm.proxy.control_plane.main import Tls
+
+    if ssl_keyfile_path is None and ssl_certfile_path is None:
+        return None
+    if ssl_keyfile_path is None or ssl_certfile_path is None:
+        raise click.UsageError("--control-plane needs both --ssl_keyfile_path and --ssl_certfile_path for TLS")
+    return Tls(keyfile=ssl_keyfile_path, certfile=ssl_certfile_path)
+
+
 @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")
@@ -1195,6 +1242,17 @@ class ProxyInitializationHelpers:
     ),
     envvar="USE_V2_MIGRATION_RESOLVER",
 )
+@click.option(
+    "--control-plane",
+    "control_plane",
+    is_flag=True,
+    default=False,
+    envvar="LITELLM_CONTROL_PLANE",
+    help=(
+        "Serve the global control plane API instead of the proxy, on the same image. Enterprise only, "
+        "refuses to start without LITELLM_LICENSE. Configure with CONTROL_PLANE_* env vars."
+    ),
+)
 @click.option(
     "--use_legacy_migration_resolver",
     is_flag=True,
@@ -1225,10 +1283,12 @@ class ProxyInitializationHelpers:
     ),
     envvar="PROMETHEUS_METRICS_PORT",
 )
+@click.pass_context
 def run_server(
+    ctx: click.Context,
     cli_args,
-    host,
-    port,
+    host: str,
+    port: int,
     api_base,
     api_version,
     model,
@@ -1279,7 +1339,9 @@ def run_server(
     use_legacy_migration_resolver: bool,
     reload: bool,
     prometheus_metrics_port: int | None,
+    control_plane: bool,
 ):
+    dependencies: Final = ctx.obj if isinstance(ctx.obj, CliDependencies) else CliDependencies()
     if cli_args:
         if cli_args == ("xai-oauth", "login"):
             from litellm.llms.xai.oauth import XAIOAuthAuthenticator
@@ -1297,6 +1359,30 @@ def run_server(
 
         run_setup_wizard()
         return
+    utility_command: Final = bool(
+        version or validate_config or health or test or check_migrations_against_existing_data
+    )
+    if control_plane_requested(control_plane, skip_server_startup, utility_command):
+        if num_workers > 1:
+            raise click.UsageError("scale the control plane with replicas.")
+        from litellm.proxy.control_plane.main import serve
+
+        if run_gunicorn or run_hypercorn or run_granian or reload:
+            raise click.UsageError(
+                "--control-plane serves with plain uvicorn, drop --run_gunicorn, --run_hypercorn, --run_granian and --reload"
+            )
+        _decrypt_kms_env_vars(dependencies.decrypt_env)
+        serve(
+            host=host,
+            port=port,
+            tls=control_plane_tls(ssl_keyfile_path, ssl_certfile_path),
+            license_check=dependencies.license_check,
+            limit_concurrency=limit_concurrency,
+            log_config=log_config,
+            keepalive_timeout=keepalive_timeout,
+            max_request_header_size=max_request_header_size,
+        )
+        return
 
     args: Final = locals()
     try:
@@ -1401,17 +1487,7 @@ def run_server(
         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
-
+        _decrypt_kms_env_vars(dependencies.decrypt_env)
         litellm_settings = None
         env_config_file: Final = os.environ.get("CONFIG_FILE_PATH")
         if (
diff --git a/tests/unit/proxy/control_plane/test_main.py b/tests/unit/proxy/control_plane/test_main.py
new file mode 100644
index 00000000000..519d803557e
--- /dev/null
+++ b/tests/unit/proxy/control_plane/test_main.py
@@ -0,0 +1,237 @@
+import logging
+from dataclasses import dataclass, field
+from typing import Final
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from litellm._logging import _get_uvicorn_json_log_config
+from litellm.proxy.control_plane.app import LicenseRequired
+from litellm.proxy.control_plane.main import Tls, _uvicorn, serve
+from litellm.proxy.control_plane.settings import ControlPlaneSettings
+from tests.unit.proxy.control_plane.support import MASTER_KEY
+
+
+@dataclass(frozen=True)
+class _License:
+    premium: bool
+
+    def is_premium(self) -> bool:
+        return self.premium
+
+
+@dataclass
+class _Runner:  # mutable-ok: records the app it was asked to serve
+    served: list[tuple[FastAPI, str, int, Tls | None, int | None, str | None, int | None, int | None]] = field(
+        default_factory=list
+    )
+
+    def __call__(
+        self,
+        app: FastAPI,
+        *,
+        host: str,
+        port: int,
+        tls: Tls | None,
+        limit_concurrency: int | None,
+        log_config: str | None,
+        keepalive_timeout: int | None,
+        max_request_header_size: int | None,
+    ) -> None:
+        self.served.append(
+            (app, host, port, tls, limit_concurrency, log_config, keepalive_timeout, max_request_header_size)
+        )
+
+
+@dataclass(frozen=True)
+class _LogConfigUnset:
+    pass
+
+
+_LOG_CONFIG_UNSET: Final = _LogConfigUnset()
+
+
+@dataclass(frozen=True)
+class _LogLevelUnset:
+    pass
+
+
+_LOG_LEVEL_UNSET: Final = _LogLevelUnset()
+
+
+@dataclass(frozen=True)
+class _UvicornCall:
+    app: FastAPI
+    host: str
+    port: int
+    server_header: bool
+    timeout_graceful_shutdown: int | None
+    timeout_keep_alive: int
+    h11_max_incomplete_event_size: int | None
+    ssl_keyfile: str | None
+    ssl_certfile: str | None
+    limit_concurrency: int | None
+    log_config: str | dict[str, object] | None | _LogConfigUnset
+    log_level: int | None | _LogLevelUnset
+
+
+@dataclass
+class _UvicornCallRecorder:  # mutable-ok: records calls made to the injected Uvicorn runner
+    calls: list[_UvicornCall] = field(default_factory=list)
+
+    def __call__(
+        self,
+        *,
+        app: FastAPI,
+        host: str,
+        port: int,
+        server_header: bool,
+        timeout_graceful_shutdown: int | None,
+        timeout_keep_alive: int,
+        h11_max_incomplete_event_size: int | None,
+        ssl_keyfile: str | None,
+        ssl_certfile: str | None,
+        limit_concurrency: int | None,
+        log_config: str | dict[str, object] | None | _LogConfigUnset = _LOG_CONFIG_UNSET,
+        log_level: int | None | _LogLevelUnset = _LOG_LEVEL_UNSET,
+    ) -> None:
+        self.calls.append(
+            _UvicornCall(
+                app,
+                host,
+                port,
+                server_header,
+                timeout_graceful_shutdown,
+                timeout_keep_alive,
+                h11_max_incomplete_event_size,
+                ssl_keyfile,
+                ssl_certfile,
+                limit_concurrency,
+                log_config,
+                log_level,
+            )
+        )
+
+
+def _settings() -> ControlPlaneSettings:
+    return ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": MASTER_KEY})
+
+
+def test_serve_builds_the_control_plane_app_and_hands_it_to_the_runner() -> None:
+    runner: Final = _Runner()
+    serve("127.0.0.1", 8100, settings=_settings, license_check=lambda: _License(True), run=runner)
+    [(app, host, port, _tls, _limit_concurrency, _log_config, _keepalive, _header_size)] = runner.served
+    assert (host, port) == ("127.0.0.1", 8100)
+    client: Final = TestClient(app)
+    assert client.get("/healthz").json() == {"status": "ok"}
+    assert client.get("/api/v1/dataplanes", headers={"Authorization": f"Bearer {MASTER_KEY}"}).json() == {"items": []}
+    assert client.get("/v1/chat/completions").status_code == 404, "the proxy's routes must not be on this app"
+
+
+def test_serve_refuses_without_a_license_and_never_binds() -> None:
+    runner: Final = _Runner()
+    with pytest.raises(LicenseRequired):
+        serve("0.0.0.0", 4000, settings=_settings, license_check=lambda: _License(False), run=runner)
+    assert runner.served == []
+
+
+def test_serve_hands_configured_tls_to_the_runner() -> None:
+    runner: Final = _Runner()
+    tls: Final = Tls(keyfile="/certs/tls.key", certfile="/certs/tls.crt")
+    serve("0.0.0.0", 8443, tls=tls, settings=_settings, license_check=lambda: _License(True), run=runner)
+    assert [entry[3] for entry in runner.served] == [tls]
+
+
+def test_serve_hands_limit_concurrency_to_the_runner() -> None:
+    runner: Final = _Runner()
+    serve(
+        "0.0.0.0",
+        8100,
+        settings=_settings,
+        license_check=lambda: _License(True),
+        run=runner,
+        limit_concurrency=250,
+    )
+    assert [entry[4] for entry in runner.served] == [250]
+
+
+def test_uvicorn_keeps_its_default_logging_config_when_unset() -> None:
+    runner: Final = _UvicornCallRecorder()
+    _uvicorn(
+        FastAPI(),
+        host="127.0.0.1",
+        port=8100,
+        tls=None,
+        limit_concurrency=None,
+        log_config=None,
+        keepalive_timeout=None,
+        max_request_header_size=None,
+        uvicorn_runner=runner,
+        json_logs=False,
+        environ={},
+    )
+    assert len(runner.calls) == 1
+    assert runner.calls[0].log_config is _LOG_CONFIG_UNSET
+    assert runner.calls[0].log_level is _LOG_LEVEL_UNSET
+
+
+def test_uvicorn_uses_json_logging_before_the_environment_log_level() -> None:
+    runner: Final = _UvicornCallRecorder()
+    _uvicorn(
+        FastAPI(),
+        host="127.0.0.1",
+        port=8100,
+        tls=None,
+        limit_concurrency=None,
+        log_config=None,
+        keepalive_timeout=None,
+        max_request_header_size=None,
+        uvicorn_runner=runner,
+        json_logs=True,
+        environ={"LITELLM_LOG": "DEBUG"},
+    )
+
+    call: Final = runner.calls[0]
+    assert call.log_config == _get_uvicorn_json_log_config()
+    assert call.log_level is _LOG_LEVEL_UNSET
+
+
+def test_uvicorn_uses_the_environment_log_level_when_json_logging_is_disabled() -> None:
+    runner: Final = _UvicornCallRecorder()
+    _uvicorn(
+        FastAPI(),
+        host="127.0.0.1",
+        port=8100,
+        tls=None,
+        limit_concurrency=None,
+        log_config=None,
+        keepalive_timeout=None,
+        max_request_header_size=None,
+        uvicorn_runner=runner,
+        json_logs=False,
+        environ={"LITELLM_LOG": "WARNING"},
+    )
+
+    assert runner.calls[0].log_config is _LOG_CONFIG_UNSET
+    assert runner.calls[0].log_level == logging.WARNING
+
+
+def test_uvicorn_forwards_an_explicit_logging_config() -> None:
+    runner: Final = _UvicornCallRecorder()
+    _uvicorn(
+        FastAPI(),
+        host="127.0.0.1",
+        port=8100,
+        tls=None,
+        limit_concurrency=None,
+        log_config="logging.yaml",
+        keepalive_timeout=None,
+        max_request_header_size=None,
+        uvicorn_runner=runner,
+        json_logs=True,
+        environ={"LITELLM_LOG": "DEBUG"},
+    )
+    assert len(runner.calls) == 1
+    assert runner.calls[0].log_config == "logging.yaml"
+    assert runner.calls[0].log_level is _LOG_LEVEL_UNSET
diff --git a/tests/unit/proxy/test_proxy_cli.py b/tests/unit/proxy/test_proxy_cli.py
index 6347c165865..0311609b5bc 100644
--- a/tests/unit/proxy/test_proxy_cli.py
+++ b/tests/unit/proxy/test_proxy_cli.py
@@ -5,6 +5,10 @@ import inspect
 import json
 import logging
 import os
+import runpy
+import sys
+import types
+import urllib.parse as urlparse
 from collections.abc import Iterator
 from contextlib import nullcontext
 from pathlib import Path
@@ -13,19 +17,13 @@ from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import click
-import fastapi
 import httpx
 import pytest
-
-
-import builtins
-import runpy
-import sys
-import types
-import urllib.parse as urlparse
-
 import uvicorn
 import yaml
+from click.testing import CliRunner
+from fastapi.testclient import TestClient
+from pydantic import TypeAdapter
 from uvicorn.config import LOOP_FACTORIES
 from uvicorn.importer import import_from_string
 
@@ -33,8 +31,11 @@ import litellm
 from litellm._logging import turn_on_json
 from litellm.constants import DEFAULT_MAX_REQUEST_HEADER_SIZE, GRACEFUL_SHUTDOWN_TEARDOWN_HEADROOM_SECONDS
 from litellm.proxy import proxy_cli
-from litellm.proxy.proxy_cli import ProxyInitializationHelpers, run_server
+from litellm.proxy.control_plane.main import serve as serve_control_plane
+from litellm.proxy.control_plane.settings import ControlPlaneSettings
+from litellm.proxy.proxy_cli import CliDependencies, ProxyInitializationHelpers, run_server
 from litellm.proxy.shutdown.graceful_shutdown_manager import DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT
+from tests.unit.proxy.control_plane.support import MASTER_KEY, Licensed
 
 
 @pytest.fixture(autouse=True)
@@ -201,42 +202,30 @@ class TestProxyInitializationHelpers:
 
         # Execute
         with pytest.raises(ValueError, match="Invalid test value"):
-            ProxyInitializationHelpers._run_test_chat_completion(
-                "localhost", 8000, "gpt-3.5-turbo", True
-            )
+            ProxyInitializationHelpers._run_test_chat_completion("localhost", 8000, "gpt-3.5-turbo", True)
 
         # Test with valid string test value
-        ProxyInitializationHelpers._run_test_chat_completion(
-            "localhost", 8000, "gpt-3.5-turbo", "http://test-url"
-        )
+        ProxyInitializationHelpers._run_test_chat_completion("localhost", 8000, "gpt-3.5-turbo", "http://test-url")
 
         # Assert
-        mock_openai.assert_called_once_with(
-            api_key="My API Key", base_url="http://test-url"
-        )
+        mock_openai.assert_called_once_with(api_key="My API Key", base_url="http://test-url")
         mock_client.chat.completions.create.assert_called()
 
     def test_get_default_unvicorn_init_args(self):
         # Test without log_config
-        args = ProxyInitializationHelpers._get_default_unvicorn_init_args(
-            "localhost", 8000
-        )
+        args = ProxyInitializationHelpers._get_default_unvicorn_init_args("localhost", 8000)
         assert args["app"] == "litellm.proxy.proxy_server:app"
         assert args["host"] == "localhost"
         assert args["port"] == 8000
         assert args["server_header"] is False
 
         # Test with log_config
-        args = ProxyInitializationHelpers._get_default_unvicorn_init_args(
-            "localhost", 8000, "log_config.json"
-        )
+        args = ProxyInitializationHelpers._get_default_unvicorn_init_args("localhost", 8000, "log_config.json")
         assert args["log_config"] == "log_config.json"
 
         # Test with json_logs=True
         with patch("litellm.json_logs", True):
-            args = ProxyInitializationHelpers._get_default_unvicorn_init_args(
-                "localhost", 8000
-            )
+            args = ProxyInitializationHelpers._get_default_unvicorn_init_args("localhost", 8000)
             # When json_logs is True, log_config should be set to the JSON log config dict
             assert args["log_config"] is not None
             assert isinstance(args["log_config"], dict)
@@ -244,15 +233,11 @@ class TestProxyInitializationHelpers:
             assert "formatters" in args["log_config"]
 
         # Test with keepalive_timeout
-        args = ProxyInitializationHelpers._get_default_unvicorn_init_args(
-            "localhost", 8000, None, 60
-        )
+        args = ProxyInitializationHelpers._get_default_unvicorn_init_args("localhost", 8000, None, 60)
         assert args["timeout_keep_alive"] == 60
 
         # Test with both log_config and keepalive_timeout
-        args = ProxyInitializationHelpers._get_default_unvicorn_init_args(
-            "localhost", 8000, "log_config.json", 120
-        )
+        args = ProxyInitializationHelpers._get_default_unvicorn_init_args("localhost", 8000, "log_config.json", 120)
         assert args["log_config"] == "log_config.json"
         assert args["timeout_keep_alive"] == 120
 
@@ -383,14 +368,10 @@ class TestProxyInitializationHelpers:
         py_file = tmp_path / "module.py"
         py_file.write_text("x = 1\n")
 
-        applied = ProxyInitializationHelpers._patch_statreload_extra_paths(
-            [str(config_file)]
-        )
+        applied: Final = ProxyInitializationHelpers._patch_statreload_extra_paths([str(config_file)])
         assert applied is True
 
-        fake_self = types.SimpleNamespace(
-            config=types.SimpleNamespace(reload_dirs=[tmp_path])
-        )
+        fake_self: Final = types.SimpleNamespace(config=types.SimpleNamespace(reload_dirs=[tmp_path]))
         yielded_paths = {Path(p).resolve() for p in StatReload.iter_py_files(fake_self)}
 
         assert config_file.resolve() in yielded_paths
@@ -407,14 +388,10 @@ class TestProxyInitializationHelpers:
         env_file = tmp_path / ".env"
         env_file.write_text("FOO=bar\n")
 
-        applied = ProxyInitializationHelpers._patch_statreload_extra_paths(
-            [str(env_file)]
-        )
+        applied: Final = ProxyInitializationHelpers._patch_statreload_extra_paths([str(env_file)])
         assert applied is True
 
-        fake_self = types.SimpleNamespace(
-            config=types.SimpleNamespace(reload_dirs=[tmp_path])
-        )
+        fake_self: Final = types.SimpleNamespace(config=types.SimpleNamespace(reload_dirs=[tmp_path]))
         yielded_paths = {Path(p).resolve() for p in StatReload.iter_py_files(fake_self)}
 
         assert env_file.resolve() in yielded_paths
@@ -426,10 +403,7 @@ class TestProxyInitializationHelpers:
             StatReload._litellm_patched_config_paths.clear()
 
         assert ProxyInitializationHelpers._patch_statreload_extra_paths([]) is False
-        assert (
-            ProxyInitializationHelpers._patch_statreload_extra_paths([None, ""])
-            is False
-        )
+        assert ProxyInitializationHelpers._patch_statreload_extra_paths([None, ""]) is False
 
     def test_patch_statreload_extra_paths_is_idempotent(self, tmp_path):
         from pathlib import Path
@@ -447,18 +421,14 @@ class TestProxyInitializationHelpers:
         for _ in range(3):
             ProxyInitializationHelpers._patch_statreload_extra_paths([str(config_file)])
 
-        fake_self = types.SimpleNamespace(
-            config=types.SimpleNamespace(reload_dirs=[tmp_path])
-        )
+        fake_self: Final = types.SimpleNamespace(config=types.SimpleNamespace(reload_dirs=[tmp_path]))
         yielded = list(StatReload.iter_py_files(fake_self))
         assert len(yielded) == len(set(map(str, yielded)))
         yielded_paths = {Path(p).resolve() for p in yielded}
         assert config_file.resolve() in yielded_paths
         assert py_file.resolve() in yielded_paths
 
-    def test_configure_dev_reload_watches_env_and_sets_override_flag(
-        self, tmp_path, monkeypatch
-    ):
+    def test_configure_dev_reload_watches_env_and_sets_override_flag(self, tmp_path, monkeypatch):
         from pathlib import Path
 
         from uvicorn.supervisors.statreload import StatReload
@@ -475,9 +445,7 @@ class TestProxyInitializationHelpers:
 
         uvicorn_args: dict = {}
         with patch("litellm._logging.verbose_proxy_logger.warning") as mock_warning:
-            ProxyInitializationHelpers._configure_dev_reload(
-                uvicorn_args, str(config_file)
-            )
+            ProxyInitializationHelpers._configure_dev_reload(uvicorn_args, str(config_file))
 
         assert os.environ["LITELLM_DEV_ENV_HOT_RELOAD"] == "True"
         assert uvicorn_args["reload"] is True
@@ -488,9 +456,7 @@ class TestProxyInitializationHelpers:
         assert "override" in warning_text
         assert ".env" in warning_text
 
-        fake_self = types.SimpleNamespace(
-            config=types.SimpleNamespace(reload_dirs=[tmp_path])
-        )
+        fake_self: Final = types.SimpleNamespace(config=types.SimpleNamespace(reload_dirs=[tmp_path]))
         yielded_paths = {Path(p).resolve() for p in StatReload.iter_py_files(fake_self)}
         assert env_file.resolve() in yielded_paths
         assert config_file.resolve() in yielded_paths
@@ -514,17 +480,13 @@ class TestProxyInitializationHelpers:
         mock_app = MagicMock()
 
         # Execute
-        ProxyInitializationHelpers._init_hypercorn_server(
-            mock_app, "localhost", 8000, None, None, None
-        )
+        ProxyInitializationHelpers._init_hypercorn_server(mock_app, "localhost", 8000, None, None, None)
 
         # Assert
         mock_asyncio_run.assert_called_once()
 
         # Test with SSL
-        ProxyInitializationHelpers._init_hypercorn_server(
-            mock_app, "localhost", 8000, "cert.pem", "key.pem", "ECDHE"
-        )
+        ProxyInitializationHelpers._init_hypercorn_server(mock_app, "localhost", 8000, "cert.pem", "key.pem", "ECDHE")
 
     @patch("granian.Granian")
     @patch("builtins.print")
@@ -693,10 +655,14 @@ class TestProxyInitializationHelpers:
             database_name_enc = urllib.parse.quote_plus(database_name)
 
             # Construct DATABASE_URL from the provided variables
-            database_url = f"postgresql://{database_username_enc}:{database_password_enc}@{database_host}/{database_name_enc}"
+            database_url: Final = (
+                f"postgresql://{database_username_enc}:{database_password_enc}@{database_host}/{database_name_enc}"
+            )
 
             # Assert the correct URL was constructed with properly escaped characters
-            expected_url = "postgresql://user%40with%2Bspecial:test-password-special-chars@localhost:5432/db_name%2Ftest"
+            expected_url: Final = (
+                "postgresql://user%40with%2Bspecial:test-password-special-chars@localhost:5432/db_name%2Ftest"
+            )
             assert database_url == expected_url
 
             # Test appending query parameters
@@ -714,9 +680,7 @@ class TestProxyInitializationHelpers:
     @patch("uvicorn.run")
     @patch("atexit.register")  # critical
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_skip_server_startup(
         self, mock_should_update, mock_setup_db, mock_atexit_register, mock_uvicorn_run, tmp_path: Path
     ):
@@ -734,11 +698,7 @@ class TestProxyInitializationHelpers:
         )
         # Remove DATABASE_URL/DIRECT_URL so the CLI doesn't attempt
         # real prisma operations when these are set in CI.
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env["PROMETHEUS_MULTIPROC_DIR"] = str(tmp_path)
         live_proxy_samples = tmp_path / "counter_123.db"
         live_proxy_samples.write_bytes(b"samples of a proxy that is still running")
@@ -758,9 +718,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -771,18 +729,12 @@ class TestProxyInitializationHelpers:
             # --- skip startup ---
             result = runner.invoke(run_server, ["--local", "--skip_server_startup"])
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             assert "Skipping server startup" in result.output
             mock_uvicorn_run.assert_not_called()
 
-            result = runner.invoke(
-                run_server, ["--local", "--skip_server_startup", "--telemetry", "False"]
-            )
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            result = runner.invoke(run_server, ["--local", "--skip_server_startup", "--telemetry", "False"])
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             assert "Skipping server startup" in result.output
             assert "telemetry" not in runner.invoke(run_server, ["--help"]).output
             assert live_proxy_samples.exists()
@@ -792,9 +744,7 @@ class TestProxyInitializationHelpers:
 
             result = runner.invoke(run_server, ["--local"])
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_uvicorn_run.assert_called_once()
             assert not live_proxy_samples.exists()
 
@@ -831,9 +781,7 @@ class TestProxyInitializationHelpers:
     @patch("uvicorn.run")
     @patch("atexit.register")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_limit_concurrency_passed_to_uvicorn(
         self, mock_should_update, mock_setup_db, mock_atexit_register, mock_uvicorn_run
     ):
@@ -850,11 +798,7 @@ class TestProxyInitializationHelpers:
             KeyManagementSettings=MagicMock(),
             save_worker_config=MagicMock(),
         )
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -864,9 +808,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.side_effect = lambda *a, **k: {
                 "app": "litellm.proxy.proxy_server:app",
@@ -874,20 +816,14 @@ class TestProxyInitializationHelpers:
                 "port": 8000,
             }
 
-            result = runner.invoke(
-                run_server, ["--local", "--limit_concurrency", "250"]
-            )
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            result = runner.invoke(run_server, ["--local", "--limit_concurrency", "250"])
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_uvicorn_run.assert_called_once()
             assert mock_uvicorn_run.call_args.kwargs.get("limit_concurrency") == 250
 
             mock_uvicorn_run.reset_mock()
             result = runner.invoke(run_server, ["--local"])
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_uvicorn_run.assert_called_once()
             assert "limit_concurrency" not in mock_uvicorn_run.call_args.kwargs
 
@@ -905,7 +841,9 @@ class TestProxyInitializationHelpers:
     @patch("httpx.HTTPTransport.handle_request")
     @patch("atexit.register")
     @patch("subprocess.Popen")
-    @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
+    @patch(
+        "litellm.proxy.db.prisma_client.PrismaManager.setup_database"
+    )  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
     @patch(  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
         "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
     )
@@ -943,9 +881,7 @@ class TestProxyInitializationHelpers:
             save_worker_config=MagicMock(),
         )
         clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL", "PROMETHEUS_METRICS_PORT")
+            k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL", "PROMETHEUS_METRICS_PORT")
         }
         clean_env["PROMETHEUS_MULTIPROC_DIR"] = str(tmp_path)
         with (
@@ -971,9 +907,7 @@ class TestProxyInitializationHelpers:
                 run_server,
                 ["--local", "--host", "127.0.0.1", "--port", "4000", "--prometheus_metrics_port", "4001"],
             )
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_popen.assert_called_once()
             spawned = list(mock_popen.call_args.args[0])
             assert spawned[1:3] == ["-m", "litellm.proxy.prometheus_metrics_server"]
@@ -996,9 +930,7 @@ class TestProxyInitializationHelpers:
             mock_popen.reset_mock()
             mock_uvicorn_run.reset_mock()
             result = runner.invoke(run_server, ["--local"])
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_popen.assert_not_called()
             mock_uvicorn_run.assert_called_once()
 
@@ -1012,9 +944,7 @@ class TestProxyInitializationHelpers:
             mock_popen.assert_not_called()
             mock_uvicorn_run.assert_not_called()
 
-            result = runner.invoke(
-                run_server, ["--local", "--prometheus_metrics_port", "0"]
-            )
+            result = runner.invoke(run_server, ["--local", "--prometheus_metrics_port", "0"])
             assert result.exit_code == 2
             assert "Invalid value for '--prometheus_metrics_port'" in result.output
             mock_popen.assert_not_called()
@@ -1036,9 +966,7 @@ class TestProxyInitializationHelpers:
     @patch("subprocess.run")
     @patch("atexit.register")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_db_timeout_settings_are_forwarded_to_pool_timeout(
         self,
         mock_should_update,
@@ -1071,11 +999,7 @@ class TestProxyInitializationHelpers:
             }
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
 
         with (
             patch.dict(os.environ, clean_env, clear=True),
@@ -1086,9 +1010,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             patch(
                 "litellm.proxy.proxy_cli.append_query_params",
                 side_effect=lambda url, params: (
@@ -1107,9 +1029,7 @@ class TestProxyInitializationHelpers:
                 ["--local", "--config", "test-config.yaml", "--skip_server_startup"],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_append_query_params.assert_called()
             appended_params = mock_append_query_params.call_args.args[1]
             assert appended_params["connection_limit"] == 5
@@ -1164,9 +1084,7 @@ class TestProxyInitializationHelpers:
     @patch("subprocess.run")
     @patch("atexit.register")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_db_connection_extra_params_forwarded_to_url(
         self,
         mock_should_update,
@@ -1201,11 +1119,7 @@ class TestProxyInitializationHelpers:
             }
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
 
         with (
             patch.dict(os.environ, clean_env, clear=True),
@@ -1216,9 +1130,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             patch(
                 "litellm.proxy.proxy_cli.append_query_params",
                 side_effect=lambda url, params: str(url),
@@ -1235,9 +1147,7 @@ class TestProxyInitializationHelpers:
                 ["--local", "--config", "test-config.yaml", "--skip_server_startup"],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_append_query_params.assert_called()
             appended_params = mock_append_query_params.call_args.args[1]
             assert appended_params["connect_timeout"] == 15
@@ -1288,9 +1198,7 @@ class TestProxyInitializationHelpers:
     @patch("subprocess.run")
     @patch("atexit.register")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_disable_prepared_statements_forwarded_to_url(
         self,
         mock_should_update,
@@ -1322,11 +1230,7 @@ class TestProxyInitializationHelpers:
             }
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
 
         with (
             patch.dict(os.environ, clean_env, clear=True),
@@ -1337,9 +1241,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             patch(
                 "litellm.proxy.proxy_cli.append_query_params",
                 side_effect=lambda url, params: str(url),
@@ -1356,9 +1258,7 @@ class TestProxyInitializationHelpers:
                 ["--local", "--config", "test-config.yaml", "--skip_server_startup"],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_append_query_params.assert_called()
             appended_params = mock_append_query_params.call_args.args[1]
             if expect_pgbouncer:
@@ -1380,9 +1280,7 @@ class TestProxyInitializationHelpers:
     @patch("subprocess.run")
     @patch("atexit.register")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_disable_prepared_statements_env_var_forwarded_to_url(
         self,
         mock_should_update,
@@ -1430,9 +1328,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             patch(
                 "litellm.proxy.proxy_cli.append_query_params",
                 side_effect=lambda url, params: str(url),
@@ -1449,9 +1345,7 @@ class TestProxyInitializationHelpers:
                 ["--local", "--config", "test-config.yaml", "--skip_server_startup"],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             appended_params = mock_append_query_params.call_args.args[1]
             if expect_pgbouncer:
                 assert appended_params["pgbouncer"] == "true", appended_params
@@ -1461,9 +1355,7 @@ class TestProxyInitializationHelpers:
     @patch("subprocess.run")
     @patch("atexit.register")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_malformed_disable_prepared_statements_env_var_is_rejected_even_when_config_enables_it(
         self,
         mock_should_update,
@@ -1509,9 +1401,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             patch(
                 "litellm.proxy.proxy_cli.append_query_params",
                 side_effect=lambda url, params: str(url),
@@ -1535,9 +1425,7 @@ class TestProxyInitializationHelpers:
     @patch("uvicorn.run")
     @patch("atexit.register")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
     def test_proxy_default_api_version_uses_azure_default(
         self, mock_should_update, mock_setup_db, mock_atexit_register, mock_uvicorn_run
     ):
@@ -1554,11 +1442,7 @@ class TestProxyInitializationHelpers:
             KeyManagementSettings=MagicMock(),
             save_worker_config=MagicMock(),
         )
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -1568,9 +1452,7 @@ class TestProxyInitializationHelpers:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -1578,9 +1460,7 @@ class TestProxyInitializationHelpers:
                 "port": 8000,
             }
             result = runner.invoke(run_server, ["--local", "--skip_server_startup"])
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_proxy_module.save_worker_config.assert_called_once()
             call_kwargs = mock_proxy_module.save_worker_config.call_args[1]
             assert call_kwargs["api_version"] == litellm.AZURE_DEFAULT_API_VERSION
@@ -1588,12 +1468,8 @@ class TestProxyInitializationHelpers:
     @patch("uvicorn.run")
     @patch("builtins.print")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
-    def test_keepalive_timeout_flag(
-        self, mock_should_update, mock_setup_db, mock_print, mock_uvicorn_run
-    ):
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
+    def test_keepalive_timeout_flag(self, mock_should_update, mock_setup_db, mock_print, mock_uvicorn_run):
         """Test that the keepalive_timeout flag is properly passed to uvicorn"""
         from click.testing import CliRunner
 
@@ -1610,11 +1486,7 @@ class TestProxyInitializationHelpers:
         # DB-setup block (un-timeout'd `subprocess.run(["prisma"])` +
         # migrate-deploy retry loop) — same isolation every other run_server
         # test in this file uses.
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
 
         with (
             patch.dict(os.environ, clean_env, clear=True),
@@ -1629,9 +1501,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             patch(
                 "litellm.proxy.proxy_cli.ProxyInitializationHelpers.is_port_in_use",
                 return_value=False,
@@ -1664,12 +1534,8 @@ class TestProxyInitializationHelpers:
     @patch("uvicorn.run")
     @patch("builtins.print")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    @patch(
-        "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
-    )
-    def test_timeout_worker_healthcheck_flag(
-        self, mock_should_update, mock_setup_db, mock_print, mock_uvicorn_run
-    ):
+    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False)
+    def test_timeout_worker_healthcheck_flag(self, mock_should_update, mock_setup_db, mock_print, mock_uvicorn_run):
         """Test that the --timeout_worker_healthcheck flag is threaded through to the uvicorn init helper."""
         from click.testing import CliRunner
 
@@ -1686,11 +1552,7 @@ class TestProxyInitializationHelpers:
         # DB-setup block (un-timeout'd `subprocess.run(["prisma"])` +
         # migrate-deploy retry loop) — same isolation every other run_server
         # test in this file uses.
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
 
         with (
             patch.dict(os.environ, clean_env, clear=True),
@@ -1705,9 +1567,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             patch(
                 "litellm.proxy.proxy_cli.ProxyInitializationHelpers.is_port_in_use",
                 return_value=False,
@@ -1719,9 +1579,7 @@ class TestProxyInitializationHelpers:
                 "port": 8000,
             }
 
-            result = runner.invoke(
-                run_server, ["--local", "--timeout_worker_healthcheck", "15"]
-            )
+            result: Final = runner.invoke(run_server, ["--local", "--timeout_worker_healthcheck", "15"])
 
             assert result.exit_code == 0
             mock_get_args.assert_called_once_with(
@@ -1736,9 +1594,7 @@ class TestProxyInitializationHelpers:
     @patch("uvicorn.run")
     @patch("builtins.print")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    def test_max_requests_before_restart_flag(
-        self, mock_setup_db, mock_print, mock_uvicorn_run
-    ):
+    def test_max_requests_before_restart_flag(self, mock_setup_db, mock_print, mock_uvicorn_run):
         """Test that the max_requests_before_restart flag is passed to uvicorn as limit_max_requests"""
         from click.testing import CliRunner
 
@@ -1751,11 +1607,7 @@ class TestProxyInitializationHelpers:
         mock_key_mgmt = MagicMock()
         mock_save_worker_config = MagicMock()
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(
                 os.environ,
@@ -1773,9 +1625,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -1783,13 +1633,9 @@ class TestProxyInitializationHelpers:
                 "port": 8000,
             }
 
-            result = runner.invoke(
-                run_server, ["--local", "--max_requests_before_restart", "123"]
-            )
+            result: Final = runner.invoke(run_server, ["--local", "--max_requests_before_restart", "123"])
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_uvicorn_run.assert_called_once()
 
             # Check that uvicorn.run was called with limit_max_requests parameter
@@ -1799,9 +1645,7 @@ class TestProxyInitializationHelpers:
     @patch("uvicorn.run")
     @patch("builtins.print")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    def test_max_requests_before_restart_jitter_flag(
-        self, mock_setup_db, mock_print, mock_uvicorn_run
-    ):
+    def test_max_requests_before_restart_jitter_flag(self, mock_setup_db, mock_print, mock_uvicorn_run):
         """--max_requests_before_restart_jitter maps to uvicorn limit_max_requests_jitter"""
         from click.testing import CliRunner
 
@@ -1812,11 +1656,7 @@ class TestProxyInitializationHelpers:
                 pass
 
         runner = CliRunner()
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch("uvicorn.Config", _NewUvicornConfig),
@@ -1831,9 +1671,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -1852,9 +1690,7 @@ class TestProxyInitializationHelpers:
                 ],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_uvicorn_run.assert_called_once()
             call_args = mock_uvicorn_run.call_args
             assert call_args[1]["limit_max_requests"] == 1000
@@ -1873,11 +1709,7 @@ class TestProxyInitializationHelpers:
         from litellm.proxy.proxy_cli import run_server
 
         runner = CliRunner()
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -1891,9 +1723,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -1913,9 +1743,7 @@ class TestProxyInitializationHelpers:
                 ],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_uvicorn_run.assert_not_called()
             mock_run_gunicorn.assert_called_once()
             g_kwargs = mock_run_gunicorn.call_args[1]
@@ -1961,20 +1789,14 @@ class TestProxyInitializationHelpers:
 
     @patch("uvicorn.run")
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
-    def test_single_worker_max_requests_warning_printed_on_uvicorn(
-        self, mock_setup_db, mock_uvicorn_run
-    ):
+    def test_single_worker_max_requests_warning_printed_on_uvicorn(self, mock_setup_db, mock_uvicorn_run):
         """uvicorn single-worker startup prints the does-not-recycle warning"""
         from click.testing import CliRunner
 
         from litellm.proxy.proxy_cli import run_server
 
         runner = CliRunner()
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -1988,9 +1810,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -1998,13 +1818,9 @@ class TestProxyInitializationHelpers:
                 "port": 8000,
             }
 
-            result = runner.invoke(
-                run_server, ["--local", "--max_requests_before_restart", "123"]
-            )
+            result: Final = runner.invoke(run_server, ["--local", "--max_requests_before_restart", "123"])
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             assert "--max_requests_before_restart 123 with --num_workers 1" in result.output
             assert "uvicorn exits the process" in result.output
 
@@ -2021,11 +1837,7 @@ class TestProxyInitializationHelpers:
         from litellm.proxy.proxy_cli import run_server
 
         runner = CliRunner()
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -2039,9 +1851,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -2054,12 +1864,8 @@ class TestProxyInitializationHelpers:
                 ["--local", "--num_workers", "2", "--max_requests_before_restart", "123"],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
-            assert not any(
-                "does not recycle a worker" in str(c) for c in mock_print.call_args_list
-            )
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
+            assert not any("does not recycle a worker" in str(c) for c in mock_print.call_args_list)
 
     @patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.run_gunicorn_server")
     @patch("uvicorn.run")
@@ -2073,11 +1879,7 @@ class TestProxyInitializationHelpers:
         from litellm.proxy.proxy_cli import run_server
 
         runner = CliRunner()
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -2091,9 +1893,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -2106,9 +1906,7 @@ class TestProxyInitializationHelpers:
                 ["--local", "--run_gunicorn", "--max_requests_before_restart", "123"],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             assert "--max_requests_before_restart 123 with --num_workers 1" in result.output
             assert "gunicorn accepts no connections" in result.output
 
@@ -2125,11 +1923,7 @@ class TestProxyInitializationHelpers:
         from litellm.proxy.proxy_cli import run_server
 
         runner = CliRunner()
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -2143,9 +1937,7 @@ class TestProxyInitializationHelpers:
                     )
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -2158,13 +1950,9 @@ class TestProxyInitializationHelpers:
                 ["--local", "--run_hypercorn", "--max_requests_before_restart", "5"],
             )
 
-            assert (
-                result.exit_code == 0
-            ), f"exit_code={result.exit_code}, output={result.output}"
+            assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
             mock_init_hypercorn.assert_called_once()
-            assert not any(
-                "does not recycle a worker" in str(c) for c in mock_print.call_args_list
-            )
+            assert not any("does not recycle a worker" in str(c) for c in mock_print.call_args_list)
 
     @pytest.mark.skipif(os.name == "nt", reason="gunicorn server path skips Windows")
     def test_gunicorn_options_include_max_requests_jitter(self):
@@ -2370,7 +2158,9 @@ class TestProxyInitializationHelpers:
 
         with patch.dict(os.environ, test_env_special):
             result = construct_database_url_from_env_vars()
-            expected_url = "postgresql://user%40with%2Bspecial:test-password-special-chars@localhost:5432/db_name%2Ftest"
+            expected_url = (
+                "postgresql://user%40with%2Bspecial:test-password-special-chars@localhost:5432/db_name%2Ftest"
+            )
             assert result == expected_url
 
         # Test without password (should still work)
@@ -2405,7 +2195,6 @@ class TestProxyInitializationHelpers:
     @patch("builtins.print")
     def test_run_server_no_config_passed(self, mock_print, mock_uvicorn_run):
         """Test that run_server properly handles the case when no config is passed"""
-        import asyncio
 
         from click.testing import CliRunner
 
@@ -2498,11 +2287,7 @@ class TestQueryEngineReaperWiring:
         from litellm.proxy.proxy_cli import run_server
 
         runner = CliRunner()
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         with (
             patch.dict(os.environ, clean_env, clear=True),
             patch.dict(
@@ -2517,12 +2302,8 @@ class TestQueryEngineReaperWiring:
                 },
             ),
             patch("uvicorn.run") as mock_uvicorn_run,
-            patch(
-                "litellm.proxy.proxy_cli.start_query_engine_reaper"
-            ) as mock_start_reaper,
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.start_query_engine_reaper") as mock_start_reaper,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -2533,17 +2314,13 @@ class TestQueryEngineReaperWiring:
         return result, mock_uvicorn_run, mock_start_reaper
 
     def test_multi_worker_uvicorn_starts_reaper(self):
-        result, mock_uvicorn_run, mock_start_reaper = self._invoke_run_server(
-            ["--local", "--num_workers", "2"]
-        )
+        result, mock_uvicorn_run, mock_start_reaper = self._invoke_run_server(["--local", "--num_workers", "2"])
         assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
         mock_uvicorn_run.assert_called_once()
         mock_start_reaper.assert_called_once()
 
     def test_single_worker_uvicorn_does_not_start_reaper(self):
-        result, mock_uvicorn_run, mock_start_reaper = self._invoke_run_server(
-            ["--local", "--num_workers", "1"]
-        )
+        result, mock_uvicorn_run, mock_start_reaper = self._invoke_run_server(["--local", "--num_workers", "1"])
         assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
         mock_uvicorn_run.assert_called_once()
         mock_start_reaper.assert_not_called()
@@ -2554,9 +2331,7 @@ class TestQueryEngineReaperWiring:
 
         with (
             patch("gunicorn.app.base.BaseApplication.run"),
-            patch(
-                "litellm.proxy.proxy_cli.start_query_engine_reaper"
-            ) as mock_start_reaper,
+            patch("litellm.proxy.proxy_cli.start_query_engine_reaper") as mock_start_reaper,
         ):
             ProxyInitializationHelpers._run_gunicorn_server(
                 host="127.0.0.1",
@@ -2604,11 +2379,7 @@ class TestRunServerDbSetup:
             save_worker_config=MagicMock(),
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env["DATABASE_URL"] = "postgresql://test:test@localhost:5432/test"
 
         with (
@@ -2620,9 +2391,7 @@ class TestRunServerDbSetup:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -2653,10 +2422,18 @@ class TestRunServerDbSetup:
             mock_setup_database.assert_called_with(use_migrate=False, use_v2_resolver=True)
 
     @patch("atexit.register")
-    @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
-    @patch("litellm.proxy.db.prisma_client.PrismaManager.build_request_log_indexes")  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
-    @patch("litellm.proxy.db.check_migration.check_prisma_schema_diff")  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
-    @patch("litellm.proxy.db.prisma_client.should_update_prisma_schema")  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
+    @patch(
+        "litellm.proxy.db.prisma_client.PrismaManager.setup_database"
+    )  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
+    @patch(
+        "litellm.proxy.db.prisma_client.PrismaManager.build_request_log_indexes"
+    )  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
+    @patch(
+        "litellm.proxy.db.check_migration.check_prisma_schema_diff"
+    )  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
+    @patch(
+        "litellm.proxy.db.prisma_client.should_update_prisma_schema"
+    )  # test-quality-ok: run_server always wires the DB; same isolation as the sibling CLI tests above
     def test_migrations_run_when_the_prisma_cli_is_not_on_path(
         self,
         mock_should_update_schema,
@@ -2680,11 +2457,7 @@ class TestRunServerDbSetup:
             save_worker_config=MagicMock(),
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env["DATABASE_URL"] = "postgresql://test:test@localhost:5432/test"
         clean_env["PATH"] = str(empty_bin)
 
@@ -2844,11 +2617,7 @@ class TestRunServerDbSetup:
             save_worker_config=MagicMock(),
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env["DATABASE_URL"] = "postgresql://test:test@localhost:5432/test"
 
         with (
@@ -2860,9 +2629,7 @@ class TestRunServerDbSetup:
                     "litellm.proxy.proxy_server": mock_proxy_module,
                 },
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -2905,11 +2672,7 @@ class TestRunServerDbSetup:
             save_worker_config=MagicMock(),
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env["DATABASE_URL"] = "sqlite:///data/litellm.db"
 
         with (
@@ -2923,9 +2686,7 @@ class TestRunServerDbSetup:
             ),
         ):
             with pytest.raises(SystemExit) as exc_info:
-                run_server.main(
-                    ["--local", "--skip_server_startup"], standalone_mode=False
-                )
+                run_server.main(["--local", "--skip_server_startup"], standalone_mode=False)
             assert exc_info.value.code == 1
             mock_setup_database.assert_not_called()
         err: Final = capsys.readouterr().err
@@ -2967,11 +2728,7 @@ class TestRunServerDbSetup:
             save_worker_config=MagicMock(),
         )
 
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env["DATABASE_URL"] = "postgresql://test:test@localhost:5432/test"
         clean_env["USE_V2_MIGRATION_RESOLVER"] = "true"
 
@@ -2985,9 +2742,7 @@ class TestRunServerDbSetup:
                 },
             ),
         ):
-            run_server.main(
-                ["--local", "--skip_server_startup"], standalone_mode=False
-            )
+            run_server.main(["--local", "--skip_server_startup"], standalone_mode=False)
 
         mock_setup_database.assert_called_once_with(use_migrate=True, use_v2_resolver=True)
         assert "--use_v2_migration_resolver is deprecated" not in capsys.readouterr().out
@@ -3030,9 +2785,7 @@ class TestRunServerDbSetup:
                 {"proxy_server": mock_proxy_module, "litellm.proxy.proxy_server": mock_proxy_module},
             ),
         ):
-            run_server.main(
-                ["--local", "--skip_server_startup", "--use_v2_migration_resolver"], standalone_mode=False
-            )
+            run_server.main(["--local", "--skip_server_startup", "--use_v2_migration_resolver"], standalone_mode=False)
 
         mock_setup_database.assert_called_once_with(use_migrate=True, use_v2_resolver=True)
         assert "--use_v2_migration_resolver is deprecated and has no effect" in capsys.readouterr().out
@@ -3169,12 +2922,7 @@ class TestRunServerDbSetup:
     def test_resolve_v2_migration_resolver(self, use_legacy_flag, env_value, expected):
         from litellm.proxy.proxy_cli import resolve_v2_migration_resolver
 
-        assert (
-            resolve_v2_migration_resolver(
-                use_legacy_flag=use_legacy_flag, env_value=env_value
-            )
-            is expected
-        )
+        assert resolve_v2_migration_resolver(use_legacy_flag=use_legacy_flag, env_value=env_value) is expected
 
     def test_deprecated_v2_flag_not_reported_outside_a_cli_invocation(self):
         from litellm.proxy.proxy_cli import deprecated_v2_flag_passed_on_cli
@@ -3215,9 +2963,7 @@ class TestRunServerDbSetup:
         )
 
         clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL", "USE_V2_MIGRATION_RESOLVER")
+            k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL", "USE_V2_MIGRATION_RESOLVER")
         }
         clean_env["DATABASE_URL"] = "postgresql://test:test@localhost:5432/test"
 
@@ -3298,9 +3044,7 @@ class TestRunServerDbSetup:
                 "sys.modules",
                 {"proxy_server": mock_proxy_module, "litellm.proxy.proxy_server": mock_proxy_module},
             ),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
             outcome as exc_info,
         ):
             mock_get_args.return_value = {"app": "litellm.proxy.proxy_server:app", "host": "localhost", "port": 8000}
@@ -3352,11 +3096,7 @@ class TestWorkerStartupHooks:
             "LITELLM_WORKER_STARTUP_HOOKS": "tests.unit.proxy.test_proxy_cli:_dummy_hook",
         }
         # Remove DATABASE_URL to avoid real DB setup
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env.update(env_overrides)
 
         with patch.dict(os.environ, clean_env, clear=True):
@@ -3379,11 +3119,7 @@ class TestWorkerStartupHooks:
         env_overrides = {
             "LITELLM_WORKER_STARTUP_HOOKS": "tests.unit.proxy.test_proxy_cli:_dummy_async_hook",
         }
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env.update(env_overrides)
 
         with patch.dict(os.environ, clean_env, clear=True):
@@ -3403,11 +3139,7 @@ class TestWorkerStartupHooks:
         env_overrides = {
             "LITELLM_WORKER_STARTUP_HOOKS": "tests.unit.proxy.test_proxy_cli:_failing_hook",
         }
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env.update(env_overrides)
 
         with patch.dict(os.environ, clean_env, clear=True):
@@ -3436,18 +3168,11 @@ class TestWorkerStartupHooks:
 
         from litellm.proxy.proxy_server import proxy_startup_event
 
-        hooks = (
-            "tests.unit.proxy.test_proxy_cli:_dummy_hook,"
-            "tests.unit.proxy.test_proxy_cli:_dummy_async_hook"
-        )
+        hooks: Final = "tests.unit.proxy.test_proxy_cli:_dummy_hook,tests.unit.proxy.test_proxy_cli:_dummy_async_hook"
         env_overrides = {
             "LITELLM_WORKER_STARTUP_HOOKS": hooks,
         }
-        clean_env = {
-            k: v
-            for k, v in os.environ.items()
-            if k not in ("DATABASE_URL", "DIRECT_URL")
-        }
+        clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")}
         clean_env.update(env_overrides)
 
         with patch.dict(os.environ, clean_env, clear=True):
@@ -3593,13 +3318,9 @@ class TestPostgresStatementTimeoutOptions:
         import yaml
 
         config_path = tmp_path / "config.yaml"
-        config_path.write_text(
-            yaml.dump({"model_list": [], "general_settings": {"database_statement_timeout": 60}})
-        )
+        config_path.write_text(yaml.dump({"model_list": [], "general_settings": {"database_statement_timeout": 60}}))
 
-        captured = _run_server_and_capture_urls(
-            str(config_path), direct_url="postgresql://t:t@localhost:5432/t"
-        )
+        captured: Final = _run_server_and_capture_urls(str(config_path), direct_url="postgresql://t:t@localhost:5432/t")
 
         assert "options" in urlparse.parse_qs(urlparse.urlparse(captured["DATABASE_URL"]).query)
         assert "options" not in urlparse.parse_qs(urlparse.urlparse(captured["DIRECT_URL"]).query)
@@ -3771,9 +3492,7 @@ class TestReadReplicaConnectionParams:
         import yaml
 
         config_path: Final = tmp_path / "config.yaml"
-        config_path.write_text(
-            yaml.dump({"model_list": [], "general_settings": {"database_connection_pool_limit": 3}})
-        )
+        config_path.write_text(yaml.dump({"model_list": [], "general_settings": {"database_connection_pool_limit": 3}}))
 
         _run_server_and_capture_urls(
             str(config_path),
@@ -3793,9 +3512,7 @@ class TestReadReplicaConnectionParams:
         import yaml
 
         config_path: Final = tmp_path / "config.yaml"
-        config_path.write_text(
-            yaml.dump({"model_list": [], "general_settings": {"database_connection_pool_limit": 3}})
-        )
+        config_path.write_text(yaml.dump({"model_list": [], "general_settings": {"database_connection_pool_limit": 3}}))
 
         _run_server_and_capture_urls(
             str(config_path),
@@ -3990,9 +3707,7 @@ class TestTokenAuthCliFlags:
             patch("litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False),
             patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database"),
             patch("uvicorn.run"),
-            patch(
-                "litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args"
-            ) as mock_get_args,
+            patch("litellm.proxy.proxy_cli.ProxyInitializationHelpers.get_default_unvicorn_init_args") as mock_get_args,
         ):
             mock_get_args.return_value = {
                 "app": "litellm.proxy.proxy_server:app",
@@ -4006,9 +3721,7 @@ class TestTokenAuthCliFlags:
         return result, database_url, toggle
 
     def test_azure_flag_assembles_a_token_bearing_database_url(self):
-        result, database_url, toggle = self._invoke_with_azure_host(
-            ["--local", "--azure_postgresql_auth"]
-        )
+        result, database_url, toggle = self._invoke_with_azure_host(["--local", "--azure_postgresql_auth"])
 
         assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
         assert database_url is not None
@@ -4236,7 +3949,11 @@ def probe_logger() -> Iterator[logging.Logger]:
 
 
 def _log_config_source(kind: str, tmp_path: Path) -> str | dict[str, object]:
-    config: Final[dict[str, object]] = {"version": 1, "incremental": True, "loggers": {_PROBE_LOGGER: {"level": "ERROR"}}}
+    config: Final[dict[str, object]] = {
+        "version": 1,
+        "incremental": True,
+        "loggers": {_PROBE_LOGGER: {"level": "ERROR"}},
+    }
     if kind == "dict":
         return config
     written: Final = tmp_path / f"log_config.{kind}"
@@ -4565,3 +4282,196 @@ def test_migration_data_check_requires_database_url(monkeypatch):
 
     with pytest.raises(click.UsageError, match="requires DATABASE_URL"):
         ProxyInitializationHelpers.run_migration_data_check(use_migrate=True)
+
+
+def test_control_plane_rejects_multiple_workers() -> None:
+    from litellm.proxy.proxy_cli import run_server
+
+    result: Final = CliRunner().invoke(run_server, ["--control-plane", "--port", "8100", "--num_workers", "3"])
+    assert result.exit_code == 2
+    assert "scale the control plane with replicas." in result.output
+
+
+@patch("uvicorn.run")
+def test_migration_data_check_precedes_control_plane_startup(
+    mock_uvicorn_run: MagicMock, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.delenv("DATABASE_URL", raising=False)
+    monkeypatch.delenv("USE_AWS_KMS", raising=False)
+    from litellm.proxy.proxy_cli import run_server
+
+    result: Final = CliRunner().invoke(run_server, ["--control-plane", "--check_migrations_against_existing_data"])
+    assert result.exit_code == 2
+    assert "requires DATABASE_URL" in result.output
+    mock_uvicorn_run.assert_not_called()
+
+
+def test_control_plane_cli_uses_decrypted_environment(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("USE_AWS_KMS", "True")
+    monkeypatch.delenv("LITELLM_MASTER_KEY", raising=False)
+    monkeypatch.delenv("KMS_TEST_SETTING", raising=False)
+
+    def decrypt_env() -> dict[str, str]:
+        return {"LITELLM_MASTER_KEY": MASTER_KEY, "KMS_TEST_SETTING": "decrypted"}
+
+    with patch("uvicorn.run") as uvicorn_run:
+        result: Final = CliRunner().invoke(
+            run_server,
+            ["--control-plane", "--host", "127.0.0.1", "--port", "8100"],
+            obj=CliDependencies(decrypt_env=decrypt_env, license_check=Licensed),
+        )
+
+    assert result.exit_code == 0, result.output
+    assert os.environ["KMS_TEST_SETTING"] == "decrypted"
+    assert os.environ["LITELLM_MASTER_KEY"] == MASTER_KEY
+    app: Final = uvicorn_run.call_args.kwargs["app"]
+    assert TestClient(app).get("/healthz").status_code == 200
+
+
+def test_control_plane_cli_serves_the_real_app_with_uvicorn_options(monkeypatch: pytest.MonkeyPatch) -> None:
+    context: Final = run_server.make_context(
+        "litellm",
+        [
+            "--control-plane",
+            "--host",
+            "127.0.0.1",
+            "--port",
+            "8100",
+            "--log_config",
+            "logging.yaml",
+            "--keepalive_timeout",
+            "30",
+            "--max_request_header_size",
+            "8192",
+            "--limit_concurrency",
+            "250",
+        ],
+    )
+    params: Final = TypeAdapter(dict[str, object]).validate_python(context.params)
+    host: Final = params["host"]
+    port: Final = params["port"]
+    limit_concurrency: Final = params["limit_concurrency"]
+    log_config: Final = params["log_config"]
+    keepalive_timeout: Final = params["keepalive_timeout"]
+    max_request_header_size: Final = params["max_request_header_size"]
+    assert params["control_plane"] is True
+    assert isinstance(host, str) and isinstance(port, int)
+    assert limit_concurrency == 250
+    assert log_config == "logging.yaml" and isinstance(log_config, str)
+    assert keepalive_timeout == 30 and isinstance(keepalive_timeout, int)
+    assert max_request_header_size == 8192 and isinstance(max_request_header_size, int)
+    settings: Final = ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": MASTER_KEY})
+    uvicorn_run: Final = MagicMock()
+    with patch("uvicorn.run", uvicorn_run):
+        serve_control_plane(
+            host=host,
+            port=port,
+            settings=lambda: settings,
+            license_check=Licensed,
+            limit_concurrency=limit_concurrency,
+            log_config=log_config,
+            keepalive_timeout=keepalive_timeout,
+            max_request_header_size=max_request_header_size,
+        )
+
+    call: Final = uvicorn_run.call_args.kwargs
+    assert (call["host"], call["port"]) == (host, port)
+    assert (call["log_config"], call["timeout_keep_alive"], call["h11_max_incomplete_event_size"]) == (
+        "logging.yaml",
+        30,
+        8192,
+    )
+    assert call["limit_concurrency"] == 250
+    response: Final = TestClient(call["app"]).get("/healthz")
+    assert (response.status_code, response.json()) == (200, {"status": "ok"})
+
+
+def test_limit_concurrency_environment_reaches_control_plane_serve(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("LIMIT_CONCURRENCY", "321")
+    context: Final = run_server.make_context("litellm", ["--control-plane"])
+    params: Final = TypeAdapter(dict[str, object]).validate_python(context.params)
+    host: Final = params["host"]
+    port: Final = params["port"]
+    limit_concurrency: Final = params["limit_concurrency"]
+    assert isinstance(host, str) and isinstance(port, int)
+    assert limit_concurrency == 321 and isinstance(limit_concurrency, int)
+    settings: Final = ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": MASTER_KEY})
+    uvicorn_run: Final = MagicMock()
+    with patch("uvicorn.run", uvicorn_run):
+        serve_control_plane(
+            host=host,
+            port=port,
+            settings=lambda: settings,
+            license_check=Licensed,
+            limit_concurrency=limit_concurrency,
+        )
+    assert uvicorn_run.call_args.kwargs["limit_concurrency"] == 321
+    assert TestClient(uvicorn_run.call_args.kwargs["app"]).get("/healthz").status_code == 200
+
+
+@pytest.mark.parametrize("args", [("--control-plane", "--version"), ("--control-plane", "--health")])
+def test_utility_commands_do_not_decrypt_kms_environment(
+    args: tuple[str, ...], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("USE_AWS_KMS", "True")
+
+    def fail_if_called() -> dict[str, str]:
+        raise AssertionError("utility command called the decryptor")
+
+    with patch(
+        "httpx.get",
+        return_value=httpx.Response(
+            200,
+            json={"status": "ok"},
+            request=httpx.Request("GET", "http://localhost/health"),
+        ),
+    ):
+        result: Final = CliRunner().invoke(
+            run_server,
+            args,
+            obj=CliDependencies(decrypt_env=fail_if_called),
+        )
+
+    assert result.exit_code == 0, result.output
+
+
+@pytest.mark.parametrize(
+    ("control_plane", "skip_server_startup", "utility_command", "expected"),
+    [
+        (True, False, False, True),
+        (True, True, False, False),
+        (True, False, True, False),
+        (False, False, False, False),
+        (False, True, True, False),
+    ],
+)
+def test_migrations_and_utility_commands_win_over_an_inherited_control_plane_flag(
+    control_plane: bool, skip_server_startup: bool, utility_command: bool, expected: bool
+) -> None:
+    from litellm.proxy.proxy_cli import control_plane_requested
+
+    assert control_plane_requested(control_plane, skip_server_startup, utility_command) is expected
+
+
+@patch("uvicorn.run")
+def test_version_prints_the_version_even_with_the_control_plane_flag(mock_uvicorn_run: MagicMock) -> None:
+    result: Final = CliRunner().invoke(run_server, ["--control-plane", "--version"])
+    assert result.exit_code == 0, result.output
+    assert "version" in result.output.lower() and mock_uvicorn_run.call_count == 0
+
+
+@pytest.mark.parametrize("flag", ["--run_hypercorn", "--run_gunicorn", "--run_granian", "--reload"])
+def test_other_servers_and_reload_are_a_usage_error_with_the_control_plane_flag(flag: str) -> None:
+    from litellm.proxy.proxy_cli import run_server
+
+    result: Final = CliRunner().invoke(run_server, ["--control-plane", flag])
+    assert result.exit_code == 2 and "plain uvicorn" in result.output
+
+
+def test_control_plane_rejects_a_lone_tls_file() -> None:
+    result: Final = CliRunner().invoke(run_server, ["--control-plane", "--ssl_keyfile_path", "/c/tls.key"])
+    assert result.exit_code == 2 and "both" in result.output