docs(proxy): clarify _kill_engine_process is on the routine reconnect path

Greptile review on #26225 (P2): the docstring said "Called when disconnect()
fails", and the SIGTERM warning log read "after failed disconnect", but
both were stale — `_kill_engine_process` is now invoked on every routine
reconnect (via the unified `recreate_prisma_client` path), not as a
disconnect-failure recovery branch. The misleading wording would have
produced confusing log lines on every reconnect cycle in production.

Update the docstring to explain the actual reason (avoiding the blocking
`disconnect()` event-loop freeze) and reword the SIGTERM warning to "during
reconnect" so it matches reality.

No behavior change; logs only.

Dockerfile

@@ -1,9 +1,9 @@
 # Base image for building
-ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:f26d42a15d09d9a643b231df929fa3cf609bedc58a728eb445be89a9d8d1da9f
+ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:3258be472764337fd13095bcbb3182da170243b5819fd67ad4c0754590588b31
 
 # Runtime image
-ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:f26d42a15d09d9a643b231df929fa3cf609bedc58a728eb445be89a9d8d1da9f
-ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:733b4042187702f832f7fdecb3aff14a61b288c4ca37af188bb5715c1caebaf8
+ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:3258be472764337fd13095bcbb3182da170243b5819fd67ad4c0754590588b31
+ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
 
 FROM $UV_IMAGE AS uvbin
 

docker/Dockerfile.alpine

@@ -3,7 +3,7 @@ ARG LITELLM_BUILD_IMAGE=python:3.11-alpine@sha256:f07e2ace46f560f09a6eeec7b4913b
 
 # Runtime image
 ARG LITELLM_RUNTIME_IMAGE=python:3.11-alpine@sha256:f07e2ace46f560f09a6eeec7b4913b80ee99546e749ef82342a419a326620856
-ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:733b4042187702f832f7fdecb3aff14a61b288c4ca37af188bb5715c1caebaf8
+ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
 
 FROM $UV_IMAGE AS uvbin
 

docker/Dockerfile.database

@@ -1,9 +1,9 @@
 # Base image for building
-ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:f26d42a15d09d9a643b231df929fa3cf609bedc58a728eb445be89a9d8d1da9f
+ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:3258be472764337fd13095bcbb3182da170243b5819fd67ad4c0754590588b31
 
 # Runtime image
-ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:f26d42a15d09d9a643b231df929fa3cf609bedc58a728eb445be89a9d8d1da9f
-ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:733b4042187702f832f7fdecb3aff14a61b288c4ca37af188bb5715c1caebaf8
+ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:3258be472764337fd13095bcbb3182da170243b5819fd67ad4c0754590588b31
+ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
 
 FROM $UV_IMAGE AS uvbin
 

docker/Dockerfile.dev

@@ -3,7 +3,7 @@ ARG LITELLM_BUILD_IMAGE=python:3.13-slim@sha256:739e7213785e88c0f702dcdc12c0973a
 
 # Runtime image
 ARG LITELLM_RUNTIME_IMAGE=python:3.13-slim@sha256:739e7213785e88c0f702dcdc12c0973afcbd606dbf021a589cab77d6b00b579d
-ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:733b4042187702f832f7fdecb3aff14a61b288c4ca37af188bb5715c1caebaf8
+ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
 
 FROM $UV_IMAGE AS uvbin
 

docker/Dockerfile.health_check

@@ -1,4 +1,4 @@
-ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:733b4042187702f832f7fdecb3aff14a61b288c4ca37af188bb5715c1caebaf8
+ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
 FROM $UV_IMAGE AS uvbin
 
 FROM python:3.13-slim@sha256:739e7213785e88c0f702dcdc12c0973afcbd606dbf021a589cab77d6b00b579d

docker/Dockerfile.non_root

@@ -1,8 +1,8 @@
 # Base images
-ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:f26d42a15d09d9a643b231df929fa3cf609bedc58a728eb445be89a9d8d1da9f
-ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:f26d42a15d09d9a643b231df929fa3cf609bedc58a728eb445be89a9d8d1da9f
+ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:3258be472764337fd13095bcbb3182da170243b5819fd67ad4c0754590588b31
+ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:3258be472764337fd13095bcbb3182da170243b5819fd67ad4c0754590588b31
 ARG PROXY_EXTRAS_SOURCE=published
-ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:733b4042187702f832f7fdecb3aff14a61b288c4ca37af188bb5715c1caebaf8
+ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
 
 FROM $UV_IMAGE AS uvbin
 

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -12,7 +12,8 @@ from litellm.llms.custom_httpx.http_handler import (
 )
 from litellm.proxy._experimental.mcp_server.oauth_utils import (
     TOKEN_NO_CACHE_HEADERS,
-    validate_loopback_redirect_uri,
+    get_request_base_url,
+    validate_trusted_redirect_uri,
 )
 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
 from litellm.proxy.common_utils.encrypt_decrypt_utils import (
@@ -29,55 +30,6 @@ router = APIRouter(
 )
 
 
-def get_request_base_url(request: Request) -> str:
-    """
-    Get the base URL for the request, considering X-Forwarded-* headers.
-
-    When behind a proxy (like nginx), the proxy may set:
-    - X-Forwarded-Proto: The original protocol (http/https)
-    - X-Forwarded-Host: The original host (may include port)
-    - X-Forwarded-Port: The original port (if not in Host header)
-
-    Args:
-        request: FastAPI Request object
-
-    Returns:
-        The reconstructed base URL (e.g., "https://proxy.example.com")
-    """
-    base_url = str(request.base_url).rstrip("/")
-    parsed = urlparse(base_url)
-
-    # Get forwarded headers
-    x_forwarded_proto = request.headers.get("X-Forwarded-Proto")
-    x_forwarded_host = request.headers.get("X-Forwarded-Host")
-    x_forwarded_port = request.headers.get("X-Forwarded-Port")
-
-    # Start with the original scheme
-    scheme = x_forwarded_proto if x_forwarded_proto else parsed.scheme
-
-    # Handle host and port
-    if x_forwarded_host:
-        # X-Forwarded-Host may already include port (e.g., "example.com:8080")
-        if ":" in x_forwarded_host and not x_forwarded_host.startswith("["):
-            # Host includes port
-            netloc = x_forwarded_host
-        elif x_forwarded_port:
-            # Port is separate
-            netloc = f"{x_forwarded_host}:{x_forwarded_port}"
-        else:
-            # Just host, no explicit port
-            netloc = x_forwarded_host
-    else:
-        # No X-Forwarded-Host, use original netloc
-        netloc = parsed.netloc
-        if x_forwarded_port and ":" not in netloc:
-            # Add forwarded port if not already in netloc
-            netloc = f"{netloc}:{x_forwarded_port}"
-
-    # Reconstruct the URL
-    return urlunparse((scheme, netloc, parsed.path, "", "", ""))
-
-
 def encode_state_with_base_url(
     base_url: str,
     original_state: str,
@@ -326,12 +278,12 @@ async def authorize_with_server(
             status_code=400, detail="MCP server authorization url is not set"
         )
 
-    # Loopback-only redirect_uri. The URI is encrypted into the OAuth
-    # state and decoded on /callback to redirect the user back; a non-
-    # loopback URI would be an open-redirect + code-theft primitive
-    # (VERIA-57 root cause B). MCP clients are native apps — loopback is
-    # the spec-compliant callback pattern.
-    validate_loopback_redirect_uri(redirect_uri)
+    # Loopback OR same-origin redirect_uri. The URI is encrypted into the
+    # OAuth state and decoded on /callback to redirect the user back;
+    # restricting to trusted origins blocks the open-redirect +
+    # code-theft primitive (VERIA-57 root cause B). Loopback supports
+    # native MCP clients; same-origin supports the proxy's own UI callback.
+    validate_trusted_redirect_uri(request, redirect_uri)
     parsed = urlparse(redirect_uri)
     base_url = urlunparse(parsed._replace(query=""))
     request_base_url = get_request_base_url(request)
@@ -648,18 +600,19 @@ async def token_endpoint(
 
 
 @router.get("/callback")
-async def callback(code: str, state: str):
+async def callback(request: Request, code: str, state: str):
     try:
         state_data = decode_state_hash(state)
         base_url = state_data["base_url"]
         original_state = state_data["original_state"]
 
-        # Re-validate loopback at the sink. /authorize rejects non-loopback
+        # Re-validate at the sink. /authorize rejects untrusted
         # redirect_uri before encoding into state, but encrypted states
         # minted before that check was added have no expiry and remain
-        # valid indefinitely. Validating here blocks the open-redirect +
-        # code-theft primitive even for pre-fix states.
-        validate_loopback_redirect_uri(base_url)
+        # valid indefinitely. Validating here (same-origin OR loopback)
+        # blocks the open-redirect + code-theft primitive even for pre-fix
+        # states while allowing the UI's same-origin callback to work.
+        validate_trusted_redirect_uri(request, base_url)
 
         params = {"code": code, "state": original_state}
         complete_returned_url = f"{base_url}?{urlencode(params)}"

litellm/proxy/_experimental/mcp_server/oauth_utils.py

@@ -2,15 +2,56 @@
 (BYOK + discoverable / pass-through OAuth proxy)."""
 
 from ipaddress import ip_address
-from urllib.parse import urlparse
+from urllib.parse import urlparse, urlunparse
 
-from fastapi import HTTPException
+from fastapi import HTTPException, Request
+
+from litellm._logging import verbose_logger
 
 # RFC 6749 §5.1 / OAuth 2.1 draft-15 §4.1.3: token-endpoint responses
 # must not be cached — both success and error bodies may reveal secrets.
 TOKEN_NO_CACHE_HEADERS = {"Cache-Control": "no-store", "Pragma": "no-cache"}
 
 
+def get_request_base_url(request: Request) -> str:
+    """
+    Get the base URL for the request, considering X-Forwarded-* headers.
+
+    When behind a proxy (like nginx), the proxy may set:
+    - X-Forwarded-Proto: The original protocol (http/https)
+    - X-Forwarded-Host: The original host (may include port)
+    - X-Forwarded-Port: The original port (if not in Host header)
+
+    Args:
+        request: FastAPI Request object
+
+    Returns:
+        The reconstructed base URL (e.g., "https://proxy.example.com")
+    """
+    base_url = str(request.base_url).rstrip("/")
+    parsed = urlparse(base_url)
+
+    x_forwarded_proto = request.headers.get("X-Forwarded-Proto")
+    x_forwarded_host = request.headers.get("X-Forwarded-Host")
+    x_forwarded_port = request.headers.get("X-Forwarded-Port")
+
+    scheme = x_forwarded_proto if x_forwarded_proto else parsed.scheme
+
+    if x_forwarded_host:
+        if ":" in x_forwarded_host and not x_forwarded_host.startswith("["):
+            netloc = x_forwarded_host
+        elif x_forwarded_port:
+            netloc = f"{x_forwarded_host}:{x_forwarded_port}"
+        else:
+            netloc = x_forwarded_host
+    else:
+        netloc = parsed.netloc
+        if x_forwarded_port and ":" not in netloc:
+            netloc = f"{netloc}:{x_forwarded_port}"
+
+    return urlunparse((scheme, netloc, parsed.path, "", "", ""))
+
+
 def validate_loopback_redirect_uri(redirect_uri: str) -> None:
     """Require a loopback ``redirect_uri`` (OAuth 2.1 §4.1.2.1 + RFC 8252
     §7.3 native-app pattern). MCP clients are native apps that listen on
@@ -46,3 +87,60 @@ def validate_loopback_redirect_uri(redirect_uri: str) -> None:
         # don't let it bubble up as a 500.
         pass
     raise HTTPException(status_code=400, detail="invalid_request")
+
+
+def validate_trusted_redirect_uri(request: Request, redirect_uri: str) -> None:
+    """Accept same-origin (proxy's own origin) OR loopback ``redirect_uri``.
+
+    Same-origin is required for the LiteLLM UI's OAuth flow: the UI
+    redirects to ``<proxy>/ui/mcp/oauth/callback`` which is not loopback
+    but is on the proxy's own trusted HTTPS origin. An attacker cannot
+    host content on the proxy's own origin without already owning the
+    proxy, so the open-redirect / code-theft primitive that motivated
+    :func:`validate_loopback_redirect_uri` does not apply here.
+
+    Loopback continues to be accepted for native MCP clients (per
+    OAuth 2.1 §4.1.2.1 + RFC 8252 §7.3).
+
+    Use this in the discoverable OAuth proxy endpoints that serve both
+    native clients and the proxy's own UI. BYOK endpoints that only
+    support native clients should keep
+    :func:`validate_loopback_redirect_uri`.
+    """
+    try:
+        parsed = urlparse(redirect_uri)
+    except ValueError:
+        raise HTTPException(status_code=400, detail="invalid_request")
+    if parsed.scheme not in ("http", "https"):
+        raise HTTPException(status_code=400, detail="invalid_request")
+    if parsed.fragment:
+        raise HTTPException(status_code=400, detail="invalid_request")
+
+    # Same-origin: scheme + netloc (host[:port]) must match the proxy's
+    # own base URL at this request.
+    try:
+        proxy_base = urlparse(get_request_base_url(request))
+        if (
+            parsed.netloc
+            and parsed.scheme == proxy_base.scheme
+            and parsed.netloc.lower() == proxy_base.netloc.lower()
+        ):
+            return
+    except Exception as exc:
+        # If we can't determine the proxy's origin, fall through to
+        # loopback. Log so the failure is diagnosable in production.
+        verbose_logger.warning(
+            "validate_trusted_redirect_uri: could not determine proxy origin, "
+            "falling back to loopback-only check. error=%s",
+            exc,
+        )
+
+    host = (parsed.hostname or "").lower()
+    if host == "localhost":
+        return
+    try:
+        if ip_address(host).is_loopback:
+            return
+    except ValueError:
+        pass
+    raise HTTPException(status_code=400, detail="invalid_request")

litellm/proxy/db/prisma_client.py

@@ -52,18 +52,25 @@ class PrismaWrapper:
             engine = self._original_prisma._engine
             process = getattr(engine, "process", None) if engine is not None else None
             if process is not None:
-                return process.pid
+                pid = process.pid
+                if isinstance(pid, int):
+                    return pid
         except (AttributeError, TypeError):
             pass
         return 0
 
     @staticmethod
     async def _kill_engine_process(pid: int) -> None:
-        """Force-kill an orphaned engine subprocess to prevent DB connection pool leaks.
+        """Force-kill the engine subprocess to prevent DB connection pool leaks.
 
-        Called when disconnect() fails and the old engine process may still be
-        holding open connections. Sends SIGTERM for graceful shutdown, waits
-        briefly, then SIGKILL as a backstop.
+        Called on every reconnect (in `recreate_prisma_client`) to retire the
+        old query-engine subprocess without invoking prisma-client-py's
+        synchronous `disconnect()` — which blocks the asyncio event loop on
+        `subprocess.Popen.wait()` for 30-120+ seconds when the engine is
+        stuck on TCP close.
+
+        Sends SIGTERM for graceful shutdown, waits briefly, then SIGKILL as
+        a backstop.
         """
         if pid <= 0:
             return
@@ -72,7 +79,7 @@ class PrismaWrapper:
         except (ProcessLookupError, PermissionError, OSError):
             return  # Already dead or inaccessible
         verbose_proxy_logger.warning(
-            "Sent SIGTERM to orphaned prisma-query-engine PID %s after failed disconnect.",
+            "Sent SIGTERM to prisma-query-engine PID %s during reconnect.",
             pid,
         )
         # Brief wait for graceful shutdown, then force-kill
@@ -217,15 +224,18 @@ class PrismaWrapper:
     async def recreate_prisma_client(
         self, new_db_url: str, http_client: Optional[Any] = None
     ):
-        """Disconnect and reconnect the Prisma client with a new database URL."""
+        """Disconnect and reconnect the Prisma client with a new database URL.
+
+        Kills the old engine subprocess directly (SIGTERM → SIGKILL) rather than
+        calling `disconnect()`. prisma-client-py's `disconnect()` calls a
+        synchronous `subprocess.Popen.wait()` that can freeze the asyncio event
+        loop for 30-120+ seconds when the engine is stuck on TCP close,
+        breaking `/health/liveliness` and causing Kubernetes pod restarts.
+        """
         from prisma import Prisma  # type: ignore
 
         old_engine_pid = self._get_engine_pid()
-
-        try:
-            await self._original_prisma.disconnect()
-        except Exception as e:
-            verbose_proxy_logger.warning(f"Failed to disconnect Prisma client: {e}")
+        if old_engine_pid > 0:
             await self._kill_engine_process(old_engine_pid)
 
         if http_client is not None:

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -151,8 +151,14 @@ if MCP_AVAILABLE:
         UserAPIKeyAuth,
         UserMCPManagementMode,
     )
-    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
-    from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
+    from litellm.proxy.auth.user_api_key_auth import (
+        _user_api_key_auth_builder,
+        user_api_key_auth,
+    )
+    from litellm.proxy.common_utils.http_parsing_utils import (
+        _read_request_body,
+        populate_request_with_path_params,
+    )
     from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
     from litellm.proxy.management_helpers.utils import management_endpoint_wrapper
     from litellm.types.mcp import MCPCredentials
@@ -1447,6 +1453,55 @@ if MCP_AVAILABLE:
 
         return _redact_mcp_credentials(temp_record)
 
+    async def _mcp_oauth_user_api_key_auth(request: Request) -> UserAPIKeyAuth:
+        """
+        Auth dependency for MCP OAuth browser-navigation endpoints (/authorize, /token).
+
+        Tries the Authorization header first. Falls back to decoding the UI
+        'token' session cookie (set by SSO login) to extract the API key, which
+        allows browser-based OAuth redirects to work without an explicit
+        Authorization header.
+        """
+        import jwt as _jwt
+
+        from litellm.proxy.proxy_server import master_key
+
+        auth_header = request.headers.get("Authorization", "")
+        api_key = auth_header  # _get_bearer_token will strip "Bearer " prefix
+
+        if not api_key:
+            token_cookie = request.cookies.get("token")
+            if token_cookie and master_key:
+                try:
+                    decoded = _jwt.decode(
+                        token_cookie,
+                        master_key,
+                        algorithms=["HS256"],
+                        # UI session cookies may omit exp; don't require it.
+                        options={"verify_exp": False},
+                    )
+                    if decoded.get("login_method") in ("sso", "username_password"):
+                        cookie_key = decoded.get("key", "")
+                        if cookie_key:
+                            api_key = f"Bearer {cookie_key}"
+                except _jwt.InvalidTokenError:
+                    pass
+
+        request_data = await _read_request_body(request=request)
+        request_data = populate_request_with_path_params(
+            request_data=request_data, request=request
+        )
+
+        return await _user_api_key_auth_builder(
+            request=request,
+            api_key=api_key,
+            azure_api_key_header="",
+            anthropic_api_key_header=None,
+            google_ai_studio_api_key_header=None,
+            azure_apim_header=None,
+            request_data=request_data,
+        )
+
     async def _get_cached_temporary_mcp_server_or_404(
         server_id: str,
         user_api_key_dict: UserAPIKeyAuth,
@@ -1497,12 +1552,12 @@ if MCP_AVAILABLE:
     @router.get(
         "/server/oauth/{server_id}/authorize",
         include_in_schema=False,
-        dependencies=[Depends(user_api_key_auth)],
+        dependencies=[Depends(_mcp_oauth_user_api_key_auth)],
     )
     async def mcp_authorize(
         request: Request,
         server_id: str,
-        user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+        user_api_key_dict: UserAPIKeyAuth = Depends(_mcp_oauth_user_api_key_auth),
         client_id: Optional[str] = None,
         redirect_uri: str = Query(...),
         state: str = "",
@@ -1542,12 +1597,12 @@ if MCP_AVAILABLE:
     @router.post(
         "/server/oauth/{server_id}/token",
         include_in_schema=False,
-        dependencies=[Depends(user_api_key_auth)],
+        dependencies=[Depends(_mcp_oauth_user_api_key_auth)],
     )
     async def mcp_token(
         request: Request,
         server_id: str,
-        user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+        user_api_key_dict: UserAPIKeyAuth = Depends(_mcp_oauth_user_api_key_auth),
         grant_type: str = Form(...),
         code: Optional[str] = Form(None),
         redirect_uri: Optional[str] = Form(None),

litellm/proxy/utils.py

@@ -4134,17 +4134,20 @@ class PrismaClient:
             )
 
             async def _do_direct_reconnect() -> None:
-                old_pid = self._get_engine_pid()
-                try:
-                    await self.db.disconnect()
-                except Exception as disconnect_err:
-                    verbose_proxy_logger.warning(
-                        "Prisma DB disconnect before reconnect failed: %s",
-                        disconnect_err,
+                db_url = os.getenv("DATABASE_URL", "")
+                if not db_url:
+                    verbose_proxy_logger.error(
+                        "DATABASE_URL not set; cannot reconnect Prisma client."
                     )
-                    await PrismaWrapper._kill_engine_process(old_pid)
-
-                await self.db.connect()
+                    raise RuntimeError("DATABASE_URL not set")
+                # Fresh Prisma client + new engine subprocess. The previous
+                # "lightweight" path called `disconnect()` which blocks the
+                # event loop on `subprocess.Popen.wait()`; since that call
+                # ends up killing the engine anyway, we do it non-blockingly
+                # via `_kill_engine_process` inside `recreate_prisma_client`.
+                self._cleanup_engine_watcher()
+                await self.db.recreate_prisma_client(db_url)
+                await self._start_engine_watcher()
                 await self.db.query_raw("SELECT 1")
 
             await asyncio.wait_for(_do_direct_reconnect(), timeout=effective_timeout)

tests/litellm/proxy/test_prisma_engine_watchdog.py

@@ -254,32 +254,48 @@ async def test_run_reconnect_cycle_uses_heavy_path_when_confirmed_dead(
 
 
 @pytest.mark.asyncio
-async def test_run_reconnect_cycle_uses_lightweight_path_when_engine_alive(
+async def test_run_reconnect_cycle_uses_direct_path_when_engine_alive(
     engine_client,
 ) -> None:
-    """_run_reconnect_cycle uses disconnect/connect when engine is alive."""
-    engine_client._engine_pid = 1234
+    """Direct reconnect (engine alive) calls recreate_prisma_client + SELECT 1.
 
-    with patch.object(engine_client, "_is_engine_alive", return_value=True):
+    The old "lightweight" path called `disconnect()` + `connect()`, which
+    blocks the event loop on the sync `process.wait()` inside aclose().
+    The fix routes both engine-alive and engine-dead paths through
+    `recreate_prisma_client`, which non-blockingly kills the old engine.
+    """
+    engine_client._engine_pid = 1234
+    engine_client._start_engine_watcher = AsyncMock()
+
+    with (
+        patch.object(engine_client, "_is_engine_alive", return_value=True),
+        patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}),
+    ):
         await engine_client._run_reconnect_cycle(timeout_seconds=5.0)
 
-    engine_client.db.connect.assert_awaited_once()
+    engine_client.db.recreate_prisma_client.assert_awaited_once_with(
+        "postgresql://test"
+    )
     engine_client.db.query_raw.assert_awaited_once_with("SELECT 1")
-    engine_client.db.recreate_prisma_client.assert_not_awaited()
+    engine_client.db.disconnect.assert_not_awaited()
 
 
 @pytest.mark.asyncio
-async def test_run_reconnect_cycle_uses_lightweight_path_when_pid_unknown(
+async def test_run_reconnect_cycle_uses_direct_path_when_pid_unknown(
     engine_client,
 ) -> None:
-    """_run_reconnect_cycle uses lightweight path when engine PID is not tracked."""
+    """When the engine PID is not tracked, direct reconnect still runs."""
     engine_client._engine_pid = 0
+    engine_client._start_engine_watcher = AsyncMock()
 
-    await engine_client._run_reconnect_cycle(timeout_seconds=5.0)
+    with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}):
+        await engine_client._run_reconnect_cycle(timeout_seconds=5.0)
 
-    engine_client.db.connect.assert_awaited_once()
+    engine_client.db.recreate_prisma_client.assert_awaited_once_with(
+        "postgresql://test"
+    )
     engine_client.db.query_raw.assert_awaited_once_with("SELECT 1")
-    engine_client.db.recreate_prisma_client.assert_not_awaited()
+    engine_client.db.disconnect.assert_not_awaited()
 
 
 @pytest.mark.asyncio
@@ -473,36 +489,38 @@ def test_on_engine_death_from_thread_ignores_stale_pid(engine_client):
 
 
 @pytest.mark.asyncio
-async def test_escalation_after_consecutive_lightweight_failures(engine_client):
-    """After N consecutive lightweight reconnect failures, _engine_confirmed_dead
+async def test_escalation_after_consecutive_direct_reconnect_failures(engine_client):
+    """After N consecutive direct reconnect failures, _engine_confirmed_dead
     is set to True so _run_reconnect_cycle takes the heavy reconnect path."""
     engine_client._reconnect_escalation_threshold = 3
     engine_client._consecutive_reconnect_failures = 0
     engine_client._db_reconnect_cooldown_seconds = 0  # disable cooldown for test
+    engine_client._start_engine_watcher = AsyncMock(return_value=None)
 
-    # Make lightweight reconnect fail every time
-    engine_client.db.disconnect = AsyncMock(return_value=None)
-    engine_client.db.connect = AsyncMock(side_effect=Exception("connect failed"))
+    # Make direct reconnect fail every time
+    engine_client.db.recreate_prisma_client = AsyncMock(
+        side_effect=Exception("recreate failed")
+    )
 
     # Run 3 failed reconnect attempts
-    for i in range(3):
-        result = await engine_client._attempt_reconnect_inside_lock(
-            force=True, reason="test", timeout_seconds=5.0
-        )
-        assert result is False
+    with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}):
+        for _ in range(3):
+            result = await engine_client._attempt_reconnect_inside_lock(
+                force=True, reason="test", timeout_seconds=5.0
+            )
+            assert result is False
 
     assert engine_client._consecutive_reconnect_failures == 3
 
-    # Next attempt should escalate: _engine_confirmed_dead set to True before _run_reconnect_cycle
+    # Next attempt should escalate to the heavy path (recreate_prisma_client still
+    # the call, but via the _engine_confirmed_dead branch that also re-arms the watcher).
     engine_client.db.recreate_prisma_client = AsyncMock(return_value=None)
-    engine_client._start_engine_watcher = AsyncMock(return_value=None)
 
     with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}):
         result = await engine_client._attempt_reconnect_inside_lock(
             force=True, reason="test_escalation", timeout_seconds=5.0
         )
 
-    # Heavy reconnect should have been attempted (recreate_prisma_client called)
     engine_client.db.recreate_prisma_client.assert_awaited_once()
 
 
@@ -511,15 +529,16 @@ async def test_successful_reconnect_resets_failure_counter(engine_client):
     """A successful reconnect resets _consecutive_reconnect_failures to 0."""
     engine_client._consecutive_reconnect_failures = 2
     engine_client._db_reconnect_cooldown_seconds = 0
+    engine_client._start_engine_watcher = AsyncMock()
 
     # Make reconnect succeed
-    engine_client.db.disconnect = AsyncMock(return_value=None)
-    engine_client.db.connect = AsyncMock(return_value=None)
+    engine_client.db.recreate_prisma_client = AsyncMock(return_value=None)
     engine_client.db.query_raw = AsyncMock(return_value=[{"result": 1}])
 
-    result = await engine_client._attempt_reconnect_inside_lock(
-        force=True, reason="test", timeout_seconds=5.0
-    )
+    with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}):
+        result = await engine_client._attempt_reconnect_inside_lock(
+            force=True, reason="test", timeout_seconds=5.0
+        )
 
     assert result is True
     assert engine_client._consecutive_reconnect_failures == 0

tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py

@@ -1512,6 +1512,12 @@ async def test_oauth_callback_redirects_with_state():
         "client_redirect_uri": "http://localhost:3000/ui/mcp/oauth/callback",
     }
 
+    from fastapi import Request
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "http://localhost:3000/"
+    mock_request.headers = {}
+
     with patch(
         "litellm.proxy._experimental.mcp_server.discoverable_endpoints.decode_state_hash"
     ) as mock_decode:
@@ -1519,6 +1525,7 @@ async def test_oauth_callback_redirects_with_state():
 
         # Call callback endpoint with code and state
         response = await callback(
+            request=mock_request,
             code="test_authorization_code_12345",
             state="encrypted_state_value",
         )
@@ -1546,6 +1553,12 @@ async def test_oauth_callback_handles_invalid_state():
     except ImportError:
         pytest.skip("MCP discoverable endpoints not available")
 
+    from fastapi import Request
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "http://localhost:3000/"
+    mock_request.headers = {}
+
     # Mock state decoding to raise an exception
     with patch(
         "litellm.proxy._experimental.mcp_server.discoverable_endpoints.decode_state_hash"
@@ -1554,6 +1567,7 @@ async def test_oauth_callback_handles_invalid_state():
 
         # Call callback endpoint with invalid state
         response = await callback(
+            request=mock_request,
             code="test_code",
             state="invalid_encrypted_state",
         )
@@ -1929,10 +1943,16 @@ async def test_callback_revalidates_loopback_on_decoded_base_url():
     valid. /callback must re-validate the decoded base_url so those
     stale states can't be used as an open-redirect + code-theft
     primitive."""
+    from fastapi import Request
+
     from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
         callback,
     )
 
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://litellm.example.com/"
+    mock_request.headers = {}
+
     with patch(
         "litellm.proxy._experimental.mcp_server.discoverable_endpoints.decode_state_hash"
     ) as mock_decode:
@@ -1944,10 +1964,100 @@ async def test_callback_revalidates_loopback_on_decoded_base_url():
             "client_redirect_uri": "https://attacker.example.com/cb",
         }
         with pytest.raises(HTTPException) as exc_info:
-            await callback(code="stolen_code", state="encrypted_stale_state")
+            await callback(
+                request=mock_request,
+                code="stolen_code",
+                state="encrypted_stale_state",
+            )
         assert exc_info.value.status_code == 400
 
 
+@pytest.mark.asyncio
+async def test_authorize_endpoint_accepts_same_origin_redirect_uri():
+    """PR #27296: the LiteLLM UI's OAuth flow uses ``<proxy>/ui/mcp/oauth/callback``
+    as redirect_uri — that's not loopback but is on the proxy's own
+    trusted origin, so it must be accepted."""
+    from fastapi import Request
+
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+        authorize,
+    )
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
+    from litellm.proxy._types import MCPTransport
+    from litellm.types.mcp import MCPAuth
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+    global_mcp_server_manager.registry.clear()
+    oauth2_server = MCPServer(
+        server_id="test_oauth_server",
+        name="test_oauth",
+        server_name="test_oauth",
+        alias="test_oauth",
+        transport=MCPTransport.http,
+        auth_type=MCPAuth.oauth2,
+        client_id="cid",
+        client_secret="cs",
+        authorization_url="https://provider.com/oauth/authorize",
+        token_url="https://provider.com/oauth/token",
+    )
+    global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://litellm.example.com/"
+    mock_request.headers = {}
+
+    with patch(
+        "litellm.proxy._experimental.mcp_server.discoverable_endpoints.encrypt_value_helper"
+    ) as mock_encrypt:
+        mock_encrypt.return_value = "mocked_encrypted_state"
+        response = await authorize(
+            request=mock_request,
+            client_id="cid",
+            mcp_server_name="test_oauth",
+            redirect_uri="https://litellm.example.com/ui/mcp/oauth/callback",
+            state="s",
+        )
+    assert response.status_code == 307
+
+
+@pytest.mark.asyncio
+async def test_callback_accepts_same_origin_on_decoded_base_url():
+    """PR #27296: /callback must accept a same-origin decoded base_url
+    so the UI's own callback URI (``<proxy>/ui/mcp/oauth/callback``)
+    completes the flow successfully."""
+    from fastapi import Request
+
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+        callback,
+    )
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://litellm.example.com/"
+    mock_request.headers = {}
+
+    with patch(
+        "litellm.proxy._experimental.mcp_server.discoverable_endpoints.decode_state_hash"
+    ) as mock_decode:
+        mock_decode.return_value = {
+            "base_url": "https://litellm.example.com/ui/mcp/oauth/callback",
+            "original_state": "ui-state",
+            "code_challenge": None,
+            "code_challenge_method": None,
+            "client_redirect_uri": "https://litellm.example.com/ui/mcp/oauth/callback",
+        }
+        response = await callback(
+            request=mock_request,
+            code="auth_code_123",
+            state="encrypted_state",
+        )
+    assert response.status_code == 302
+    assert "litellm.example.com/ui/mcp/oauth/callback" in response.headers["location"]
+    assert "code=auth_code_123" in response.headers["location"]
+    assert "state=ui-state" in response.headers["location"]
+
+
 @pytest.mark.asyncio
 async def test_token_endpoint_sets_no_store_cache_control():
     """RFC 6749 §5.1 / OAuth 2.1 draft-15 §4.1.3: the token response

tests/test_litellm/proxy/db/test_prisma_self_heal.py

@@ -34,18 +34,18 @@ async def test_attempt_db_reconnect_should_succeed(mock_proxy_logging):
     client = PrismaClient(
         database_url="mock://test", proxy_logging_obj=mock_proxy_logging
     )
-    client.db.disconnect = AsyncMock(return_value=None)
-    client.db.connect = AsyncMock(return_value=None)
+    client.db.recreate_prisma_client = AsyncMock(return_value=None)
     client.db.query_raw = AsyncMock(return_value=[{"result": 1}])
+    client._start_engine_watcher = AsyncMock()
 
-    result = await client.attempt_db_reconnect(
-        reason="unit_test_reconnect_success",
-        force=True,
-    )
+    with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}):
+        result = await client.attempt_db_reconnect(
+            reason="unit_test_reconnect_success",
+            force=True,
+        )
 
     assert result is True
-    client.db.disconnect.assert_awaited_once()
-    client.db.connect.assert_awaited_once()
+    client.db.recreate_prisma_client.assert_awaited_once_with("postgresql://test")
     client.db.query_raw.assert_awaited_once_with("SELECT 1")
 
 
@@ -140,15 +140,19 @@ async def test_attempt_db_reconnect_should_set_cooldown_after_attempt(
     )
     client._db_last_reconnect_attempt_ts = 0.0
     client._db_reconnect_cooldown_seconds = 10
-    client.db.disconnect = AsyncMock(return_value=None)
-    client.db.connect = AsyncMock(return_value=None)
+    client.db.recreate_prisma_client = AsyncMock(return_value=None)
     client.db.query_raw = AsyncMock(return_value=[{"result": 1}])
+    client._start_engine_watcher = AsyncMock()
 
     # Use a counter-based mock to avoid StopIteration when time.time() is called
     # more times than expected (varies by Python version / internal code paths).
     fake_clock = iter(range(100, 10000))
-    with patch(
-        "litellm.proxy.utils.time.time", side_effect=lambda: float(next(fake_clock))
+    with (
+        patch(
+            "litellm.proxy.utils.time.time",
+            side_effect=lambda: float(next(fake_clock)),
+        ),
+        patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}),
     ):
         result = await client.attempt_db_reconnect(
             reason="unit_test_cooldown_timestamp_after_attempt",
@@ -162,23 +166,28 @@ async def test_attempt_db_reconnect_should_set_cooldown_after_attempt(
 
 
 @pytest.mark.asyncio
-async def test_run_reconnect_cycle_watchdog_should_use_direct_db_ops(
+async def test_run_reconnect_cycle_watchdog_should_use_recreate_prisma_client(
     mock_proxy_logging,
 ):
+    """Direct reconnect goes through recreate_prisma_client (which non-blockingly
+    kills the old engine) instead of calling disconnect() — see issue #26191.
+    """
     client = PrismaClient(
         database_url="mock://test", proxy_logging_obj=mock_proxy_logging
     )
-    client.disconnect = AsyncMock(side_effect=AssertionError("wrapper disconnect used"))
-    client.connect = AsyncMock(side_effect=AssertionError("wrapper connect used"))
-    client.db.disconnect = AsyncMock(return_value=None)
-    client.db.connect = AsyncMock(return_value=None)
+    client.db.disconnect = AsyncMock(
+        side_effect=AssertionError("disconnect must not be called")
+    )
+    client.db.recreate_prisma_client = AsyncMock(return_value=None)
     client.db.query_raw = AsyncMock(return_value=[{"result": 1}])
+    client._start_engine_watcher = AsyncMock()
 
-    await client._run_reconnect_cycle(timeout_seconds=None)
+    with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}):
+        await client._run_reconnect_cycle(timeout_seconds=None)
 
-    client.db.disconnect.assert_awaited_once()
-    client.db.connect.assert_awaited_once()
+    client.db.recreate_prisma_client.assert_awaited_once_with("postgresql://test")
     client.db.query_raw.assert_awaited_once_with("SELECT 1")
+    client.db.disconnect.assert_not_awaited()
 
 
 @pytest.mark.asyncio
@@ -189,19 +198,22 @@ async def test_run_reconnect_cycle_watchdog_should_use_default_timeout_budget(
         database_url="mock://test", proxy_logging_obj=mock_proxy_logging
     )
     client._db_watchdog_reconnect_timeout_seconds = 0.1
-    client.db.disconnect = AsyncMock(return_value=None)
+    client._start_engine_watcher = AsyncMock()
 
-    async def _slow_connect():
+    async def _slow_recreate(_db_url):
         await asyncio.sleep(0.08)
 
     async def _slow_query(_query: str):
         await asyncio.sleep(0.08)
         return [{"result": 1}]
 
-    client.db.connect = AsyncMock(side_effect=_slow_connect)
+    client.db.recreate_prisma_client = AsyncMock(side_effect=_slow_recreate)
     client.db.query_raw = AsyncMock(side_effect=_slow_query)
 
-    with pytest.raises(asyncio.TimeoutError):
+    with (
+        pytest.raises(asyncio.TimeoutError),
+        patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}),
+    ):
         await client._run_reconnect_cycle(timeout_seconds=None)
 
 
@@ -212,19 +224,22 @@ async def test_run_reconnect_cycle_timeout_should_use_single_overall_budget(
     client = PrismaClient(
         database_url="mock://test", proxy_logging_obj=mock_proxy_logging
     )
-    client.db.disconnect = AsyncMock(return_value=None)
+    client._start_engine_watcher = AsyncMock()
 
-    async def _slow_connect():
+    async def _slow_recreate(_db_url):
         await asyncio.sleep(0.08)
 
     async def _slow_query(_query: str):
         await asyncio.sleep(0.08)
         return [{"result": 1}]
 
-    client.db.connect = AsyncMock(side_effect=_slow_connect)
+    client.db.recreate_prisma_client = AsyncMock(side_effect=_slow_recreate)
     client.db.query_raw = AsyncMock(side_effect=_slow_query)
 
-    with pytest.raises(asyncio.TimeoutError):
+    with (
+        pytest.raises(asyncio.TimeoutError),
+        patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}),
+    ):
         await client._run_reconnect_cycle(timeout_seconds=0.1)
 
 
@@ -319,42 +334,32 @@ async def test_db_health_watchdog_start_stop_lifecycle(mock_proxy_logging):
 
 
 @pytest.mark.asyncio
-async def test_lightweight_reconnect_kills_engine_on_disconnect_failure(
+async def test_recreate_prisma_client_kills_old_engine_without_disconnect(
     mock_proxy_logging,
 ):
-    """Lightweight reconnect must kill the old engine PID when disconnect() fails."""
+    """recreate_prisma_client SIGTERMs the old engine PID directly rather than
+    calling `disconnect()`, which blocks the asyncio event loop on the sync
+    `subprocess.Popen.wait()` inside prisma-client-py — see issue #26191.
+    """
     client = PrismaClient(
         database_url="mock://test", proxy_logging_obj=mock_proxy_logging
     )
-    client.db.disconnect = AsyncMock(side_effect=Exception("disconnect failed"))
-    client.db.connect = AsyncMock(return_value=None)
-    client.db.query_raw = AsyncMock(return_value=[{"result": 1}])
+    disconnect_mock = AsyncMock(
+        side_effect=AssertionError("disconnect must not be called on reconnect path")
+    )
+    client.db._original_prisma.disconnect = disconnect_mock
 
     with (
-        patch.object(client, "_get_engine_pid", return_value=9999),
-        patch("os.kill") as mock_kill,
-        patch("asyncio.sleep", new_callable=AsyncMock),
+        patch.object(client.db, "_get_engine_pid", return_value=9999),
+        patch("litellm.proxy.db.prisma_client.os.kill") as mock_kill,
+        patch("litellm.proxy.db.prisma_client.asyncio.sleep", new_callable=AsyncMock),
     ):
-        await client._run_reconnect_cycle(timeout_seconds=5.0)
+        # Return a Prisma instance whose connect() is awaitable.
+        fake_new_prisma = MagicMock()
+        fake_new_prisma.connect = AsyncMock(return_value=None)
+        with patch("prisma.Prisma", return_value=fake_new_prisma):
+            await client.db.recreate_prisma_client("postgresql://test")
 
     mock_kill.assert_any_call(9999, signal.SIGTERM)
-    client.db.connect.assert_awaited_once()
-    client.db.query_raw.assert_awaited_once_with("SELECT 1")
-
-
-@pytest.mark.asyncio
-async def test_lightweight_reconnect_skips_kill_on_successful_disconnect(
-    mock_proxy_logging,
-):
-    """Lightweight reconnect must NOT kill when disconnect() succeeds."""
-    client = PrismaClient(
-        database_url="mock://test", proxy_logging_obj=mock_proxy_logging
-    )
-    client.db.disconnect = AsyncMock(return_value=None)
-    client.db.connect = AsyncMock(return_value=None)
-    client.db.query_raw = AsyncMock(return_value=[{"result": 1}])
-
-    with patch("os.kill") as mock_kill:
-        await client._run_reconnect_cycle(timeout_seconds=5.0)
-
-    mock_kill.assert_not_called()
+    disconnect_mock.assert_not_awaited()
+    fake_new_prisma.connect.assert_awaited_once()

tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py

@@ -1551,6 +1551,98 @@ class TestTemporaryMCPSessionEndpoints:
 
         assert "permission" in str(exc_info.value)
 
+    @pytest.mark.asyncio
+    async def test_mcp_oauth_user_api_key_auth_falls_back_to_token_cookie(self):
+        """
+        When the Authorization header is absent but a valid 'token' cookie is
+        present (browser navigation), _mcp_oauth_user_api_key_auth should
+        decode the cookie JWT and authenticate via the API key stored in it.
+        """
+        import jwt
+
+        from litellm.proxy.management_endpoints.mcp_management_endpoints import (
+            _mcp_oauth_user_api_key_auth,
+        )
+
+        master_key = "test-master-key"
+        api_key_in_cookie = "sk-test-cookie-key"
+        token_cookie = jwt.encode(
+            {
+                "user_id": "user@example.com",
+                "key": api_key_in_cookie,
+                "user_role": "proxy_admin",
+                "login_method": "sso",
+            },
+            master_key,
+            algorithm="HS256",
+        )
+
+        mock_request = MagicMock()
+        mock_request.headers = {}
+        mock_request.cookies = {"token": token_cookie}
+
+        expected_auth = generate_mock_user_api_key_auth(
+            user_role=LitellmUserRoles.PROXY_ADMIN, api_key=api_key_in_cookie
+        )
+        fake_proxy_server = types.SimpleNamespace(master_key=master_key)
+
+        with (
+            patch.dict(sys.modules, {"litellm.proxy.proxy_server": fake_proxy_server}),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints._user_api_key_auth_builder",
+                AsyncMock(return_value=expected_auth),
+            ) as auth_builder_mock,
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints._read_request_body",
+                AsyncMock(return_value={}),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.populate_request_with_path_params",
+                side_effect=lambda request_data, request: request_data,
+            ),
+        ):
+            result = await _mcp_oauth_user_api_key_auth(mock_request)
+
+        assert result is expected_auth
+        _, call_kwargs = auth_builder_mock.call_args
+        assert call_kwargs["api_key"] == f"Bearer {api_key_in_cookie}"
+
+    @pytest.mark.asyncio
+    async def test_mcp_oauth_user_api_key_auth_uses_authorization_header_when_present(
+        self,
+    ):
+        """When Authorization header is present it takes priority over the cookie."""
+        from litellm.proxy.management_endpoints.mcp_management_endpoints import (
+            _mcp_oauth_user_api_key_auth,
+        )
+
+        expected_auth = generate_mock_user_api_key_auth(
+            user_role=LitellmUserRoles.PROXY_ADMIN
+        )
+        mock_request = MagicMock()
+        mock_request.headers = {"Authorization": "Bearer sk-header-key"}
+        mock_request.cookies = {}
+
+        with (
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints._user_api_key_auth_builder",
+                AsyncMock(return_value=expected_auth),
+            ) as auth_builder_mock,
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints._read_request_body",
+                AsyncMock(return_value={}),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.populate_request_with_path_params",
+                side_effect=lambda request_data, request: request_data,
+            ),
+        ):
+            result = await _mcp_oauth_user_api_key_auth(mock_request)
+
+        assert result is expected_auth
+        _, call_kwargs = auth_builder_mock.call_args
+        assert call_kwargs["api_key"] == "Bearer sk-header-key"
+
     @pytest.mark.asyncio
     async def test_mcp_authorize_proxies_to_discoverable_endpoint(self):
         from litellm.proxy.management_endpoints.mcp_management_endpoints import (