fix(mcp): delegate PKCE bypass for internal MCP servers

CLAUDE.md

@@ -117,6 +117,7 @@ LiteLLM is a unified interface for 100+ LLM providers with two main components:
 - **Always use `antd` for new UI components** — we are migrating off of `@tremor/react`. Do not introduce new `Badge`, `Text`, `Card`, `Grid`, `Title`, or other imports from `@tremor/react` in any new or modified file. Use `antd` equivalents: `Tag` for labels, `Typography.Text` / `Typography.Title` / `Typography.Paragraph` for textual content (avoid plain text-only `<span>`, `<p>`, `<h*>` when Typography fits), and `Card` from `antd`. Note that `antd` has no `"yellow"` Tag color — use `"gold"` for amber/yellow.
 
 ### MCP OAuth / OpenAPI Transport Mapping
+- **`available_on_public_internet: false` with `delegate_auth_to_upstream: true` (oauth2, interactive — not `client_credentials`)** — LiteLLM still allows the anonymous upstream PKCE path (no proxy API key for `/authorize` and matching MCP routes). The internal-only flag mainly affects other surfaces (e.g. IP-based discovery). Rely on the upstream IdP and network policy; the dashboard shows a warning when both are set, and the proxy logs a warning when the server is loaded from config or the database.
 - `TRANSPORT.OPENAPI` is a UI-only concept. The backend only accepts `"http"`, `"sse"`, or `"stdio"`. Always map it to `"http"` before any API call (including pre-OAuth temp-session calls).
 - FastAPI validation errors return `detail` as an array of `{loc, msg, type}` objects. Error extractors must handle: array (map `.msg`), string, nested `{error: string}`, and fallback.
 - When an MCP server already has `authorization_url` stored, skip OAuth discovery (`_discovery_metadata`) — the server URL for OpenAPI MCPs is the spec file, not the API base, and fetching it causes timeouts.

litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py

@@ -332,8 +332,6 @@ def _target_servers_delegate_auth_to_upstream(
             # non-bool must not silently enable the bypass.
             if getattr(server, "delegate_auth_to_upstream", False) is not True:
                 return False
-            if not getattr(server, "available_on_public_internet", True):
-                return False
             # Never delegate for M2M (client_credentials) servers: LiteLLM
             # fetches the upstream token automatically using stored credentials,
             # so allowing anonymous bypass would let any external caller invoke

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -145,6 +145,30 @@ def _warn(field_name: str, value: Optional[str]) -> None:
     _warn("server_name", server_name)
 
 
+def _warn_internal_delegate_pkce_if_applicable(
+    server: MCPServer, *, source: str
+) -> None:
+    """Surface internal + upstream PKCE delegate in logs for operators."""
+    if server.auth_type != MCPAuth.oauth2:
+        return
+    if getattr(server, "delegate_auth_to_upstream", False) is not True:
+        return
+    if getattr(server, "available_on_public_internet", True):
+        return
+    if server.has_client_credentials:
+        return
+    label = get_server_prefix(server)
+    verbose_logger.warning(
+        "MCP server %r (id=%s, source=%s): internal-only (available_on_public_internet=false) "
+        "with delegate_auth_to_upstream=true. Anonymous callers can reach the upstream OAuth2 "
+        "/authorize flow and complete PKCE without a LiteLLM API key session; ensure the "
+        "upstream IdP and network enforce your access policy.",
+        label,
+        server.server_id,
+        source,
+    )
+
+
 def _deserialize_json_dict(data: Any) -> Optional[Dict[str, str]]:
     """
     Deserialize optional JSON mappings stored in the database.
@@ -297,32 +321,6 @@ async def load_servers_from_config(
             )()
             name_for_prefix = get_server_prefix(temp_server)
 
-            # Use alias for name if present, else server_name
-            alias = server_config.get("alias", None)
-
-            # Apply mcp_aliases mapping if provided
-            if mcp_aliases and alias is None:
-                # Check if this server_name has an alias in mcp_aliases
-                for alias_name, target_server_name in mcp_aliases.items():
-                    if (
-                        target_server_name == server_name
-                        and alias_name not in used_aliases
-                    ):
-                        alias = alias_name
-                        used_aliases.add(alias_name)
-                        verbose_logger.debug(
-                            f"Mapped alias '{alias_name}' to server '{server_name}'"
-                        )
-                        break
-
-            # Create a temporary server object to use with get_server_prefix utility
-            temp_server = type(
-                "TempServer",
-                (),
-                {"alias": alias, "server_name": server_name, "server_id": None},
-            )()
-            name_for_prefix = get_server_prefix(temp_server)
-
             server_url = server_config.get("url", None) or ""
             # Generate stable server ID based on parameters
             server_id = self._generate_stable_server_id(
@@ -425,6 +423,7 @@ async def load_servers_from_config(
                 ),
             )
             self._assign_unique_short_prefix(new_server)
+            _warn_internal_delegate_pkce_if_applicable(new_server, source="config")
             self.config_mcp_servers[server_id] = new_server
 
             # Check if this is an OpenAPI-based server
@@ -834,6 +833,7 @@ async def build_mcp_server_from_table(
             )
             or "urn:ietf:params:oauth:token-type:access_token",
         )
+        _warn_internal_delegate_pkce_if_applicable(new_server, source="database")
         return new_server
 
     async def _maybe_register_openapi_tools(
@@ -995,9 +995,6 @@ async def get_allowed_mcp_servers(
                     # unauthenticated caller would get LiteLLM to proxy tool
                     # calls using its stored client_credentials.
                     and not server.has_client_credentials
-                    # Internal-only servers must not be reachable from public
-                    # internet callers who happen to carry an upstream token.
-                    and getattr(server, "available_on_public_internet", True)
                 ]
                 combined_servers.update(delegate_server_ids)
 

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -1578,7 +1578,6 @@ async def _mcp_oauth_user_api_key_auth(request: Request) -> UserAPIKeyAuth:
                     _s
                     and getattr(_s, "auth_type", None) == MCPAuth.oauth2
                     and getattr(_s, "delegate_auth_to_upstream", False) is True
-                    and getattr(_s, "available_on_public_internet", True)
                     # M2M servers fetch tokens with stored credentials; never
                     # expose their /authorize or /token endpoints anonymously.
                     and not _s.has_client_credentials

tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py

@@ -1462,13 +1462,11 @@ async def mock_auth_raises(*_args, **_kwargs):
             assert exc_info.value.status_code == 401
             mock_auth.assert_called_once()
 
-    async def test_delegate_ignored_for_non_public_server(self):
+    async def test_delegate_bypass_for_internal_server(self):
         """
-        Internal-only delegate servers must not bypass LiteLLM auth for
-        anonymous public callers.
+        Delegate + oauth2 interactive servers bypass LiteLLM auth even when
+        ``available_on_public_internet`` is False (internal MCPs).
         """
-        from fastapi import HTTPException
-
         from litellm.types.mcp import MCPAuth
         from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
@@ -1489,6 +1487,8 @@ async def test_delegate_ignored_for_non_public_server(self):
         )
 
         async def mock_auth_raises(*_args, **_kwargs):
+            from fastapi import HTTPException
+
             raise HTTPException(status_code=401, detail="No key provided")
 
         with (
@@ -1501,10 +1501,9 @@ async def mock_auth_raises(*_args, **_kwargs):
             ) as mock_mgr,
         ):
             mock_mgr.get_mcp_server_by_name.return_value = internal_server
-            with pytest.raises(HTTPException) as exc_info:
-                await MCPRequestHandler.process_mcp_request(scope)
-            assert exc_info.value.status_code == 401
-            mock_auth.assert_called_once()
+            auth, *_rest = await MCPRequestHandler.process_mcp_request(scope)
+            mock_auth.assert_not_called()
+            assert auth.api_key is None
 
     async def test_get_allowed_servers_excludes_client_credentials_delegate(self):
         """
@@ -1551,10 +1550,10 @@ async def test_get_allowed_servers_excludes_client_credentials_delegate(self):
         assert "pkce-server" in result
         assert "m2m-server" not in result
 
-    async def test_get_allowed_servers_excludes_non_public_delegate(self):
+    async def test_get_allowed_servers_includes_internal_delegate(self):
         """
         Internal-only (available_on_public_internet=False) delegate servers
-        must not appear in the anonymous allow-list.
+        appear in the anonymous allow-list like public delegate servers.
         """
         from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
             MCPServerManager,
@@ -1593,7 +1592,7 @@ async def test_get_allowed_servers_excludes_non_public_delegate(self):
             result = await manager.get_allowed_mcp_servers(None)
 
         assert "public-server" in result
-        assert "internal-server" not in result
+        assert "internal-server" in result
 
     def test_extract_target_server_names_matches_routing_parser(self):
         """

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

@@ -2633,6 +2633,67 @@ async def test_round_trip_timestamps_preserved(self):
         assert rebuilt_table.updated_at == updated
 
 
+class TestInternalDelegatePkceWarningLog:
+    @pytest.mark.asyncio
+    async def test_build_mcp_server_logs_on_internal_delegate_interactive(self, caplog):
+        caplog.set_level(logging.WARNING, logger="LiteLLM")
+        manager = MCPServerManager()
+        table_record = LiteLLM_MCPServerTable(
+            server_id="warn-del-1",
+            server_name="warn_server",
+            url="https://example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://idp.example.com/authorize",
+            token_url="https://idp.example.com/token",
+            available_on_public_internet=False,
+            delegate_auth_to_upstream=True,
+        )
+        await manager.build_mcp_server_from_table(table_record)
+        combined = " ".join(r.getMessage() for r in caplog.records)
+        assert "internal-only" in combined
+        assert "delegate_auth_to_upstream=true" in combined
+
+    @pytest.mark.asyncio
+    async def test_build_mcp_server_no_internal_delegate_log_when_public(self, caplog):
+        caplog.set_level(logging.WARNING, logger="LiteLLM")
+        manager = MCPServerManager()
+        table_record = LiteLLM_MCPServerTable(
+            server_id="warn-del-2",
+            server_name="warn_server_pub",
+            url="https://example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://idp.example.com/authorize",
+            token_url="https://idp.example.com/token",
+            available_on_public_internet=True,
+            delegate_auth_to_upstream=True,
+        )
+        await manager.build_mcp_server_from_table(table_record)
+        combined = " ".join(r.getMessage() for r in caplog.records)
+        assert "internal-only" not in combined
+
+    def test_warn_skipped_for_client_credentials(self, caplog):
+        caplog.set_level(logging.WARNING, logger="LiteLLM")
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            _warn_internal_delegate_pkce_if_applicable,
+        )
+
+        server = MCPServer(
+            server_id="m2m-1",
+            name="x",
+            url="https://example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            oauth2_flow="client_credentials",
+            available_on_public_internet=False,
+            delegate_auth_to_upstream=True,
+        )
+        _warn_internal_delegate_pkce_if_applicable(server, source="test")
+        combined = " ".join(r.getMessage() for r in caplog.records)
+        assert "internal-only" not in combined
+
+
 class TestHasClientCredentialsOAuth2Flow:
     """
     Regression tests for the M2M auto-detection bug.

tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py

@@ -1696,10 +1696,10 @@ async def test_mcp_oauth_user_api_key_auth_requires_oauth2_for_delegate_bypass(
         assert call_kwargs["api_key"] == ""
 
     @pytest.mark.asyncio
-    async def test_mcp_oauth_user_api_key_auth_requires_public_server_for_delegate_bypass(
+    async def test_mcp_oauth_user_api_key_auth_internal_delegate_bypasses(
         self,
     ):
-        """Internal-only delegate servers must still require LiteLLM auth."""
+        """Internal-only delegate servers still get anonymous PKCE /authorize bypass."""
         from litellm.proxy.management_endpoints.mcp_management_endpoints import (
             _mcp_oauth_user_api_key_auth,
         )
@@ -1711,6 +1711,8 @@ async def test_mcp_oauth_user_api_key_auth_requires_public_server_for_delegate_b
         mock_request.headers = {}
         mock_request.cookies = {}
         mock_request.path_params = {"server_id": "server-1"}
+        # Real path so ``endswith("/token")`` is not fooled by MagicMock truthiness.
+        mock_request.url = types.SimpleNamespace(path="/server-1/authorize")
         internal_server = MagicMock()
         internal_server.auth_type = MCPAuth.oauth2
         internal_server.delegate_auth_to_upstream = True
@@ -1742,10 +1744,8 @@ async def test_mcp_oauth_user_api_key_auth_requires_public_server_for_delegate_b
         ):
             result = await _mcp_oauth_user_api_key_auth(mock_request)
 
-        assert result is expected_auth
-        auth_builder_mock.assert_awaited_once()
-        _, call_kwargs = auth_builder_mock.call_args
-        assert call_kwargs["api_key"] == ""
+        assert isinstance(result, UserAPIKeyAuth)
+        auth_builder_mock.assert_not_called()
 
     @pytest.mark.asyncio
     async def test_mcp_authorize_proxies_to_discoverable_endpoint(self):

ui/litellm-dashboard/src/components/mcp_tools/MCPPermissionManagement.tsx

@@ -1,5 +1,5 @@
 import React, { useEffect } from "react";
-import { Form, Select, Tooltip, Collapse, Input, Space, Button, Switch } from "antd";
+import { Alert, Form, Select, Tooltip, Collapse, Input, Space, Button, Switch } from "antd";
 import { InfoCircleOutlined, MinusCircleOutlined, PlusOutlined } from "@ant-design/icons";
 import { MCPServer, AUTH_TYPE } from "./types";
 const { Panel } = Collapse;
@@ -25,6 +25,12 @@ const MCPPermissionManagement: React.FC<MCPPermissionManagementProps> = ({
   const form = Form.useFormInstance();
   const watchedAuthType = Form.useWatch("auth_type", form);
   const isOAuth2 = watchedAuthType === AUTH_TYPE.OAUTH2;
+  const watchedDelegateAuth = Form.useWatch("delegate_auth_to_upstream", form);
+  const watchedPublicInternet = Form.useWatch("available_on_public_internet", form);
+  const showInternalDelegatePkceWarning =
+    isOAuth2 &&
+    watchedDelegateAuth === true &&
+    watchedPublicInternet === false;
 
   // Set initial values when mcpServer changes
   useEffect(() => {
@@ -144,6 +150,16 @@ const MCPPermissionManagement: React.FC<MCPPermissionManagementProps> = ({
             </div>
           )}
 
+          {showInternalDelegatePkceWarning && (
+            <Alert
+              type="warning"
+              showIcon
+              className="mb-2"
+              message="Internal server with upstream OAuth delegation"
+              description="This MCP server is configured as internal-only but delegates auth to upstream. Anonymous users will be able to reach the upstream OAuth2 /authorize flow without a LiteLLM session. Ensure your upstream provider and network enforce access controls."
+            />
+          )}
+
           <Form.Item
             label={
               <span className="text-sm font-medium text-gray-700 flex items-center">