feat(mcp): warn about the ID-JAG capture gap for config-declared servers and on the SSO debug page (#39350)

* feat(sso): surface the ID-JAG capture gap on the SSO debug page

/sso/debug/callback is where an operator lands when they are already trying to work out
why ID-JAG is failing, so the reason belongs on it. The annotation appears only when the
active SSO provider captures no identity assertion AND an oauth2_id_jag server is
registered for that gap to break; a deployment without both renders the page it rendered
before, byte for byte. Only the provider name and the remedy are rendered, never a
configured value, and an unreachable MCP table costs the page its annotation rather than
the page itself.

The payload carries the one mutable-ok in this work. Conditionally including a member of a
JSON document has to construct a mapping, and the rejected alternatives are recorded on the
helper so the next reader does not rediscover them.

Held out of the diagnosability PR deliberately: that PR is already reviewed and green, and
this surface ships with the remaining config-load warning as one follow-up.

* feat(mcp): warn at config load when an oauth2_id_jag server outruns the SSO provider's assertion capture

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(sso): trim comments on the ID-JAG debug page diagnostic

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(sso): clean up merged imports

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(sso): satisfy type discipline for diagnostic payload

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* style(sso): keep the optional ID-JAG payload member on one line for ruff format

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(sso): use Python 3.10-compatible assert_never

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Yassin Kortam <yassin@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -149,6 +149,9 @@ from litellm.proxy._types import (
 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
 from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value_helper
 from litellm.proxy.common_utils.user_api_key_cache import get_management_object_ttl
+from litellm.proxy.management_endpoints.sso.id_jag_assertion_capture import (
+    id_jag_assertion_capture_gap_at_startup,
+)
 from litellm.proxy.utils import PrismaClient, ProxyLogging, get_server_root_path
 from litellm.repositories.table_repositories import MCPServerRepository
 from litellm.types.llms.custom_http import httpxSpecialProvider
@@ -1212,6 +1215,20 @@ def _warn_internal_delegate_pkce_if_applicable(server: MCPServer, *, source: str
     )
 
 
+def _warn_config_id_jag_server_outruns_sso(server: MCPServer) -> None:
+    if server.auth_type != MCPAuth.oauth2_id_jag:
+        return
+    gap: Final = id_jag_assertion_capture_gap_at_startup()
+    if gap is None:
+        return
+    verbose_logger.warning(
+        "MCP server %r (id=%s, source=config) is declared with auth_type=oauth2_id_jag, but %s.",
+        get_server_prefix(server),
+        server.server_id,
+        gap,
+    )
+
+
 def _deserialize_json_dict(data: str | _StringMap | None) -> dict[str, str] | None:
     """
     Deserialize optional JSON mappings stored in the database.
@@ -2204,6 +2221,7 @@ class MCPServerManager:
             )
             self._assign_unique_short_prefix(new_server)
             _warn_internal_delegate_pkce_if_applicable(new_server, source="config")
+            _warn_config_id_jag_server_outruns_sso(new_server)
             self.config_mcp_servers[server_id] = new_server
             self._set_oauth_discovery_deferred(
                 server_id,

litellm/proxy/management_endpoints/sso/id_jag_assertion_capture.py

@@ -12,7 +12,8 @@ from __future__ import annotations
 
 import os
 from enum import Enum
-from typing import assert_never
+
+from typing_extensions import assert_never
 
 from litellm.proxy.management_endpoints.sso.saml_sso import SAMLAuthHandler
 
@@ -69,3 +70,12 @@ def id_jag_assertion_capture_gap() -> str | None:
             )
         case _:
             assert_never(provider)
+
+
+def id_jag_assertion_capture_gap_at_startup() -> str | None:
+    """Config load runs before SSO settings stored in the database are reconciled into the process
+    environment, so an unresolved provider at that point is not yet a gap; the SSO callback reports it
+    once a login happens."""
+    if active_sso_provider() is ActiveSSOProvider.none:
+        return None
+    return id_jag_assertion_capture_gap()

litellm/proxy/management_endpoints/ui_sso.py

@@ -1694,13 +1694,26 @@ async def warn_if_id_jag_assertion_uncaptured(assertion: SSOIdentityAssertion |
     except Exception as exc:  # noqa: BLE001  # diagnostics must never break the login
         verbose_proxy_logger.debug("Could not check for oauth2_id_jag MCP servers after SSO login: %s", exc)
         return
-    gap = id_jag_assertion_capture_gap()
+    gap: Final = id_jag_assertion_capture_gap()
     verbose_proxy_logger.warning(
         "SSO login captured no IdP identity assertion while an oauth2_id_jag MCP server is registered: %s",
         gap if gap is not None else "the identity provider's token response carried no usable id_token",
     )
 
 
+async def id_jag_capture_gap_to_surface() -> str | None:
+    """The gap worth showing an operator: a real capture gap and an ``oauth2_id_jag`` server
+    registered for it to break."""
+    gap: Final = id_jag_assertion_capture_gap()
+    if gap is None:
+        return None
+    try:
+        return gap if await ema_assertion_retention_enabled() else None
+    except Exception as exc:  # noqa: BLE001  # diagnostics must never break the page they annotate
+        verbose_proxy_logger.debug("Could not check for oauth2_id_jag MCP servers: %s", exc)
+        return None
+
+
 async def create_team_member_add_task(team_id, user_info):
     """Create a task for adding a member to a team."""
     try:
@@ -4759,10 +4772,13 @@ async def debug_sso_callback(request: Request):
     safe_raw_claims: Final = {k: v for k, v in (received_response or {}).items() if k not in _OAUTH_TOKEN_FIELDS}
     safe_access_token_claims = {k: v for k, v in (access_token_payload or {}).items() if k not in _OAUTH_TOKEN_FIELDS}
 
+    gap: Final = await id_jag_capture_gap_to_surface()
+    id_jag: Final = {"id_jag_assertion_capture": gap} if gap is not None else {}  # mutable-ok: optional JSON member
     sso_payload: Final = {
         "parsed_by_proxy": filtered_result,
         "raw_claims": safe_raw_claims,
         "access_token_claims": safe_access_token_claims,
+        **id_jag,
     }
 
     # Replace the placeholder in the template with the actual data

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

@@ -461,6 +461,30 @@ class TestMCPServerManager:
         base.update(overrides)
         return {"m2mserver": base}
 
+    def _id_jag_config(self):
+        return {
+            "idjag_server": {
+                "url": "https://example.com/mcp",
+                "transport": MCPTransport.http,
+                "auth_type": MCPAuth.oauth2_id_jag,
+                "client_id": "cid",
+                "client_secret": "csec",
+                "token_exchange_endpoint": "https://idp.example.com/token",
+                "id_jag_resource_token_endpoint": "https://resource.example.com/token",
+                "id_jag_resource": "https://resource.example.com",
+            }
+        }
+
+    def _clear_sso_env(self, monkeypatch):
+        for env_var in (
+            "GOOGLE_CLIENT_ID",
+            "MICROSOFT_CLIENT_ID",
+            "GENERIC_CLIENT_ID",
+            "SAML_IDP_METADATA_URL",
+            "SAML_IDP_METADATA_XML",
+        ):
+            monkeypatch.delenv(env_var, raising=False)
+
     @pytest.mark.parametrize("value", ["1", "true", "TRUE", "yes", "on"])
     def test_mcp_oauth_discovery_on_startup_true_values(self, value):
         with patch.dict(os.environ, {"LITELLM_MCP_OAUTH_DISCOVERY_ON_STARTUP": value}):
@@ -1130,6 +1154,72 @@ class TestMCPServerManager:
         server = next(iter(manager.config_mcp_servers.values()))
         assert server.oauth2_flow is None
 
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_warns_for_id_jag_with_google_sso(self, monkeypatch, caplog):
+        self._clear_sso_env(monkeypatch)
+        monkeypatch.setenv("GOOGLE_CLIENT_ID", "google-cid")
+        manager = MCPServerManager()
+        with (
+            patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)),
+            patch.object(manager, "_hydrate_config_servers_dcr_clients", new=AsyncMock()),
+            caplog.at_level(logging.WARNING, logger="LiteLLM"),
+        ):
+            await manager.load_servers_from_config(self._id_jag_config())
+
+        warnings = [message for message in caplog.messages if "oauth2_id_jag" in message]
+        assert len(warnings) == 1
+        assert "idjag_server" in warnings[0]
+        assert "GENERIC_CLIENT_ID" in warnings[0]
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_does_not_warn_for_id_jag_without_sso(self, monkeypatch, caplog):
+        self._clear_sso_env(monkeypatch)
+        manager = MCPServerManager()
+        with (
+            patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)),
+            patch.object(manager, "_hydrate_config_servers_dcr_clients", new=AsyncMock()),
+            caplog.at_level(logging.WARNING, logger="LiteLLM"),
+        ):
+            await manager.load_servers_from_config(self._id_jag_config())
+
+        assert not any("oauth2_id_jag" in message for message in caplog.messages)
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_does_not_warn_for_api_key_with_google_sso(self, monkeypatch, caplog):
+        self._clear_sso_env(monkeypatch)
+        monkeypatch.setenv("GOOGLE_CLIENT_ID", "google-cid")
+        manager = MCPServerManager()
+        config = {
+            "api_key_server": {
+                "url": "https://example.com/mcp",
+                "transport": MCPTransport.http,
+                "auth_type": MCPAuth.api_key,
+                "auth_value": "upstream-secret",
+            }
+        }
+        with (
+            patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)),
+            patch.object(manager, "_hydrate_config_servers_dcr_clients", new=AsyncMock()),
+            caplog.at_level(logging.WARNING, logger="LiteLLM"),
+        ):
+            await manager.load_servers_from_config(config)
+
+        assert not any("oauth2_id_jag" in message for message in caplog.messages)
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_does_not_warn_for_id_jag_with_generic_sso(self, monkeypatch, caplog):
+        self._clear_sso_env(monkeypatch)
+        monkeypatch.setenv("GENERIC_CLIENT_ID", "generic-cid")
+        manager = MCPServerManager()
+        with (
+            patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)),
+            patch.object(manager, "_hydrate_config_servers_dcr_clients", new=AsyncMock()),
+            caplog.at_level(logging.WARNING, logger="LiteLLM"),
+        ):
+            await manager.load_servers_from_config(self._id_jag_config())
+
+        assert not any("oauth2_id_jag" in message for message in caplog.messages)
+
     def _client_forwarded_config(self, auth_type, **overrides):
         base = {
             "url": "https://example.com/mcp",

tests/test_litellm/proxy/management_endpoints/test_id_jag_assertion_capture.py

@@ -4,6 +4,7 @@ from litellm.proxy.management_endpoints.sso.id_jag_assertion_capture import (
     ActiveSSOProvider,
     active_sso_provider,
     id_jag_assertion_capture_gap,
+    id_jag_assertion_capture_gap_at_startup,
 )
 
 _SSO_ENV_VARS = (
@@ -98,3 +99,19 @@ class TestIdJagAssertionCaptureGap:
         gap = id_jag_assertion_capture_gap()
         assert gap is not None
         assert "google" in gap
+
+
+class TestIdJagAssertionCaptureGapAtStartup:
+    def test_no_provider_at_startup_is_not_yet_a_gap(self):
+        assert id_jag_assertion_capture_gap_at_startup() is None
+
+    def test_google_provider_at_startup_reports_the_capture_gap(self, monkeypatch):
+        monkeypatch.setenv("GOOGLE_CLIENT_ID", "google-cid")
+        startup_gap = id_jag_assertion_capture_gap_at_startup()
+        callback_gap = id_jag_assertion_capture_gap()
+        assert startup_gap is not None
+        assert startup_gap == callback_gap
+
+    def test_generic_provider_at_startup_has_no_gap(self, monkeypatch):
+        monkeypatch.setenv("GENERIC_CLIENT_ID", "generic-cid")
+        assert id_jag_assertion_capture_gap_at_startup() is None

tests/test_litellm/proxy/management_endpoints/test_ui_sso.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import os
+from contextlib import ExitStack
 from contextlib import asynccontextmanager
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -8167,6 +8168,131 @@ async def test_debug_sso_callback_handles_missing_raw_response():
     assert "user@example.com" in body
 
 
+# ── The debug page is where an operator lands when ID-JAG is failing ──────────
+
+_GOOGLE_DEBUG_CLIENT_ID = "debug-google-client-id"
+_GENERIC_DEBUG_CLIENT_ID = "debug-generic-client-id"
+
+
+async def _render_debug_page(provider_env, id_jag_registered, gap_override=None):
+    """Drive /sso/debug/callback and return the raw response body."""
+    from litellm.proxy.management_endpoints.ui_sso import GoogleSSOHandler, debug_sso_callback
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "http://proxy.example.com/"
+    mock_request.cookies = {}
+    mock_request.query_params = {}
+
+    parsed = {"sub": "user_123", "email": "u@example.com"}
+
+    async def fake_generic(**kwargs):
+        return parsed, {"sub": "user_123"}, {"scope": "openid"}, None
+
+    async def fake_google(**kwargs):
+        return parsed
+
+    stack = [
+        patch.dict(os.environ, provider_env, clear=False),
+        patch("litellm.proxy.management_endpoints.ui_sso.get_generic_sso_response", side_effect=fake_generic),
+        patch.object(GoogleSSOHandler, "get_google_callback_response", side_effect=fake_google),
+        patch(
+            "litellm.proxy.management_endpoints.ui_sso.ema_assertion_retention_enabled",
+            AsyncMock(return_value=id_jag_registered),
+        ),
+        patch("litellm.proxy.proxy_server.general_settings", {}),
+        patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
+        patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
+        patch("litellm.proxy.proxy_server.jwt_handler", MagicMock(spec=JWTHandler)),
+    ]
+    if gap_override is not None:
+        stack.append(
+            patch(
+                "litellm.proxy.management_endpoints.ui_sso.id_jag_capture_gap_to_surface",
+                AsyncMock(return_value=gap_override["value"]),
+            )
+        )
+
+    with ExitStack() as es:
+        for ctx in stack:
+            es.enter_context(ctx)
+        for var in ("MICROSOFT_CLIENT_ID", "GOOGLE_CLIENT_ID", "GENERIC_CLIENT_ID", "SAML_IDP_METADATA_URL"):
+            if var not in provider_env:
+                os.environ.pop(var, None)
+        response = await debug_sso_callback(mock_request)
+
+    return response.body.decode()
+
+
+@pytest.mark.asyncio
+async def test_debug_page_surfaces_the_capture_gap_to_the_operator_reading_it():
+    """This page is exactly where someone lands when ID-JAG is failing, so the reason has to be
+    on it; a diagnostic that cannot reach the surface the operator is staring at has a hole."""
+    body = await _render_debug_page({"GOOGLE_CLIENT_ID": _GOOGLE_DEBUG_CLIENT_ID}, id_jag_registered=True)
+
+    assert "id_jag_assertion_capture" in body
+    assert "google" in body
+    assert "GENERIC_CLIENT_ID" in body
+
+
+@pytest.mark.asyncio
+async def test_debug_page_gap_text_carries_no_configuration_values():
+    """Only the provider name and the remedy belong in rendered HTML; the client id the operator
+    configured is not ours to echo back onto a page."""
+    body = await _render_debug_page({"GOOGLE_CLIENT_ID": _GOOGLE_DEBUG_CLIENT_ID}, id_jag_registered=True)
+
+    assert _GOOGLE_DEBUG_CLIENT_ID not in body
+
+
+@pytest.mark.asyncio
+async def test_debug_page_is_byte_identical_when_the_provider_captures():
+    """A deployment with no gap must get the page it got before this change, to the byte. The
+    comparison is against the endpoint with the diagnostic forced inert, not against a guess."""
+    with_feature = await _render_debug_page(
+        {"GENERIC_CLIENT_ID": _GENERIC_DEBUG_CLIENT_ID}, id_jag_registered=True
+    )
+    pre_change = await _render_debug_page(
+        {"GENERIC_CLIENT_ID": _GENERIC_DEBUG_CLIENT_ID},
+        id_jag_registered=True,
+        gap_override={"value": None},
+    )
+
+    assert with_feature == pre_change
+    assert "id_jag" not in with_feature
+
+
+@pytest.mark.asyncio
+async def test_debug_page_is_byte_identical_when_no_id_jag_server_is_registered():
+    """Most deployments run Google SSO and no id_jag server at all; their debug page must not
+    grow an ID-JAG section about a feature they do not use."""
+    with_feature = await _render_debug_page(
+        {"GOOGLE_CLIENT_ID": _GOOGLE_DEBUG_CLIENT_ID}, id_jag_registered=False
+    )
+    pre_change = await _render_debug_page(
+        {"GOOGLE_CLIENT_ID": _GOOGLE_DEBUG_CLIENT_ID},
+        id_jag_registered=False,
+        gap_override={"value": None},
+    )
+
+    assert with_feature == pre_change
+    assert "id_jag" not in with_feature
+
+
+@pytest.mark.asyncio
+async def test_debug_page_survives_a_store_outage():
+    """The page's job is to render claims; an unreachable MCP table must cost it the annotation,
+    not the page."""
+    from litellm.proxy.management_endpoints.ui_sso import id_jag_capture_gap_to_surface
+
+    with (
+        patch.dict(os.environ, {"GOOGLE_CLIENT_ID": _GOOGLE_DEBUG_CLIENT_ID}, clear=False),
+        patch(
+            "litellm.proxy.management_endpoints.ui_sso.ema_assertion_retention_enabled",
+            AsyncMock(side_effect=Exception("db down")),
+        ),
+    ):
+        assert await id_jag_capture_gap_to_surface() is None
+
+
 async def _render_legacy_login_page(env_overrides, general_settings):
     from litellm.proxy.management_endpoints.ui_sso import google_login
 
@@ -8679,7 +8805,6 @@ async def test_cli_completion_persists_assertion_under_db_user_id():
 
     retain_mock.assert_awaited_once_with(user_id="cli-user-id", assertion=assertion)
     assert response.status_code == 200
-# ── Diagnosing a login that captured nothing an id_jag server can spend ───────
 
 
 def _id_jag_gap_warnings(logger_mock) -> list: