fix(mcp): scope user-field-values endpoints to caller's allowed servers

The per-server GET/POST/DELETE /v1/mcp/server/{server_id}/user-field-values
handlers previously called get_mcp_server directly with no access check.
A non-admin user who knew (or guessed) another team's server_id could
fetch the admin-declared user_fields metadata (display names, header
names, env var names) or write/delete user-field values for that server.

Gate the three endpoints behind get_all_mcp_servers_for_user (same scoping
already used by the aggregated /v1/mcp/user-field-values endpoint and by
fetch_mcp_server). Admins bypass the check; everyone else gets a 404
(not 403) so the status code does not leak server existence.

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -2100,6 +2100,28 @@ if MCP_AVAILABLE:
     # their own values via these endpoints; values are injected at request
     # time as either HTTP headers (http/sse) or env vars (stdio).
 
+    async def _assert_user_can_access_mcp_server(
+        prisma_client: Any,
+        user_api_key_dict: UserAPIKeyAuth,
+        server_id: str,
+    ) -> None:
+        """Refuse access to MCP servers the caller is not permitted to see.
+
+        Returns ``404`` (rather than ``403``) on failure so callers can't
+        probe for the existence of servers outside their allowed set by
+        comparing status codes. Admin viewers bypass the check.
+        """
+        if _user_has_admin_view(user_api_key_dict):
+            return
+        allowed_servers = await get_all_mcp_servers_for_user(
+            prisma_client, user_api_key_dict
+        )
+        if not does_mcp_server_exist(allowed_servers, server_id):
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail={"error": f"MCP Server {server_id} not found"},
+            )
+
     def _build_user_fields_status(
         server: "LiteLLM_MCPServerTable",
         stored_values: Optional[Dict[str, str]],
@@ -2161,6 +2183,9 @@ if MCP_AVAILABLE:
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail={"error": "User ID not found in token"},
             )
+        await _assert_user_can_access_mcp_server(
+            prisma_client, user_api_key_dict, server_id
+        )
         server = await get_mcp_server(prisma_client, server_id)
         if server is None:
             raise HTTPException(
@@ -2194,6 +2219,9 @@ if MCP_AVAILABLE:
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail={"error": "User ID not found in token"},
             )
+        await _assert_user_can_access_mcp_server(
+            prisma_client, user_api_key_dict, server_id
+        )
         server = await get_mcp_server(prisma_client, server_id)
         if server is None:
             raise HTTPException(
@@ -2277,6 +2305,9 @@ if MCP_AVAILABLE:
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail={"error": "User ID not found in token"},
             )
+        await _assert_user_can_access_mcp_server(
+            prisma_client, user_api_key_dict, server_id
+        )
         server = await get_mcp_server(prisma_client, server_id)
         if server is None:
             raise HTTPException(

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

@@ -474,6 +474,10 @@ async def test_get_user_field_values_endpoint_reports_missing():
             "litellm.proxy._experimental.mcp_server.db.get_user_field_values",
             AsyncMock(return_value=None),
         ),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_all_mcp_servers_for_user",
+            AsyncMock(return_value=[server_row]),
+        ),
     ):
         # Call the endpoint function directly to skip FastAPI auth wiring.
         from litellm.proxy.management_endpoints import mcp_management_endpoints as mod
@@ -546,6 +550,10 @@ async def test_post_user_field_values_rejects_undeclared_keys():
         patch(
             "litellm.proxy._experimental.mcp_server.server._invalidate_user_fields_cache"
         ),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_all_mcp_servers_for_user",
+            AsyncMock(return_value=[server_row]),
+        ),
     ):
         from fastapi import FastAPI
         from fastapi.testclient import TestClient
@@ -587,9 +595,15 @@ async def test_post_user_field_values_rejects_server_with_no_declared_fields():
         return_value=server_row
     )
 
-    with patch(
-        "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
-        return_value=prisma_client,
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
+            return_value=prisma_client,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_all_mcp_servers_for_user",
+            AsyncMock(return_value=[server_row]),
+        ),
     ):
         from fastapi import FastAPI
         from fastapi.testclient import TestClient
@@ -606,3 +620,58 @@ async def test_post_user_field_values_rejects_server_with_no_declared_fields():
             "/v1/mcp/server/srv-1/user-field-values", json={"values": {"X": "y"}}
         )
         assert res.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_user_field_value_endpoints_404_when_server_not_in_allowed_set():
+    """Non-admin callers must not be able to probe servers outside their allowed set.
+
+    Without this gate, an authenticated user who knows another team's
+    ``server_id`` could call GET/POST/DELETE on the user-field-values
+    endpoints and leak the admin-declared field descriptors (header names,
+    env var names) for that server.
+    """
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.proxy.management_endpoints import mcp_management_endpoints as mod
+
+    server_row = _server_row_with_user_fields()
+    prisma_client = MagicMock()
+    prisma_client.db.litellm_mcpservertable.find_unique = AsyncMock(
+        return_value=server_row
+    )
+
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
+            return_value=prisma_client,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_all_mcp_servers_for_user",
+            AsyncMock(return_value=[]),
+        ),
+    ):
+        from fastapi import FastAPI
+        from fastapi.testclient import TestClient
+
+        from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+
+        app = FastAPI()
+        app.include_router(mod.router)
+        app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
+            api_key="hashed", user_id="outsider"
+        )
+        client = TestClient(app)
+
+        get_res = client.get("/v1/mcp/server/srv-1/user-field-values")
+        assert get_res.status_code == 404
+        post_res = client.post(
+            "/v1/mcp/server/srv-1/user-field-values",
+            json={"values": {"GMAIL_TOKEN": "tok"}},
+        )
+        assert post_res.status_code == 404
+        del_res = client.delete("/v1/mcp/server/srv-1/user-field-values")
+        assert del_res.status_code == 404
+
+        # find_unique must never be reached — the access gate fails first,
+        # before any server metadata is read or returned.
+        prisma_client.db.litellm_mcpservertable.find_unique.assert_not_called()