feat(mcp): aggregate gateway DCR discovery front door behind mcp_gateway_dcr

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

@@ -10,6 +10,10 @@ from typing_extensions import assert_never
 
 import litellm
 from litellm._logging import verbose_logger
+from litellm.proxy._experimental.mcp_server.oauth_utils import (
+    get_request_base_url,
+    is_mcp_gateway_dcr_enabled,
+)
 from litellm.proxy._experimental.mcp_server.outbound_credentials.bridge_credentials import (
     BridgeEnvelopeAdmitted,
     BridgeEnvelopeInvalid,
@@ -120,6 +124,96 @@ def _has_client_supplied_mcp_auth(
     return bool(mcp_auth_header) or bool(mcp_server_auth_headers)
 
 
+def _is_aggregate_gateway_dcr_challenge_scope(
+    route: str,
+    mcp_servers: list[str] | None,
+    mcp_auth_header: str | None,
+    mcp_server_auth_headers: dict[str, dict[str, str]] | None,
+    exc: Exception,
+) -> bool:
+    """True when an unauthenticated request to the aggregate ``/mcp`` endpoint
+    should receive the RFC 9728 401 challenge that advertises the gateway as
+    the authorization server (``mcp_gateway_dcr`` front door).
+
+    Fires only for a genuine 401 on the aggregate scope: any named target
+    (path or ``x-mcp-servers``) belongs to the per-server challenge paths, and
+    client-supplied MCP auth headers mean the caller is not a cold-start DCR
+    client. Fails closed to the original admission error otherwise."""
+    if not is_mcp_gateway_dcr_enabled():
+        return False
+    if not _is_litellm_auth_admission_error(exc):
+        return False
+    if mcp_servers:
+        return False
+    if _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers):
+        return False
+    return len(MCPRequestHandler._extract_target_server_names_from_path(route)) == 0
+
+
+def _aggregate_gateway_dcr_challenge(request: Request, invalid_token: bool) -> HTTPException:
+    """The RFC 9728 challenge for the aggregate endpoint: points the client at
+    the gateway's own protected-resource metadata so a DCR client discovers
+    the gateway as its authorization server and starts the sign-in flow.
+
+    ``invalid_token`` adds the RFC 6750 error code for a request that DID
+    present a bearer that failed admission (expired or revoked), telling
+    spec-compliant clients to re-authorize rather than retry; a request with
+    no credentials at all gets the bare challenge per RFC 6750 section 3.1."""
+    error_attr = 'error="invalid_token", ' if invalid_token else ""
+    resource_metadata_url = f"{get_request_base_url(request)}/.well-known/oauth-protected-resource/mcp"
+    return HTTPException(
+        status_code=401,
+        detail={
+            "error": "authentication_required",
+            "message": "Authenticate with the gateway to use the MCP endpoint.",
+        },
+        headers={"WWW-Authenticate": f'Bearer {error_attr}resource_metadata="{resource_metadata_url}"'},
+    )
+
+
+def _admission_failure_fallback(
+    request: Request,
+    request_route: str,
+    mcp_servers: list[str] | None,
+    mcp_auth_header: str | None,
+    mcp_server_auth_headers: dict[str, dict[str, str]] | None,
+    exc: Exception,
+    bearer_presented: bool,
+) -> UserAPIKeyAuth:
+    """Map a failed LiteLLM admission to its anonymous fallback or challenge.
+
+    Two fallbacks exist, both gated on a genuine 401 with no client-supplied
+    MCP auth headers. The pass-through cold start (RFC 9728 / MCP
+    Authorization spec discovery return) admits anonymously so the route's
+    401 emitter can produce the per-server challenge. The aggregate
+    gateway-DCR scope converts the failure into the gateway's own
+    resource_metadata challenge, with the RFC 6750 ``invalid_token`` error
+    code when the caller DID present a bearer (an expired gateway session
+    must re-authorize, not retry a dead token). Anything else re-raises the
+    original admission error unchanged."""
+    mcp_servers_from_path = _parse_mcp_server_names_from_path(request_route, mcp_servers)
+    if (
+        mcp_servers_from_path is not None
+        and not _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers)
+        and _is_litellm_auth_admission_error(exc)
+        and _is_mcp_passthrough_cold_start(
+            mcp_servers_from_path,
+            client_ip=IPAddressUtils.get_mcp_client_ip(request),
+        )
+    ):
+        verbose_logger.debug("MCP pass-through cold start: deferring admission to route 401 emitter")
+        return UserAPIKeyAuth()
+    if _is_aggregate_gateway_dcr_challenge_scope(
+        route=request_route,
+        mcp_servers=mcp_servers,
+        mcp_auth_header=mcp_auth_header,
+        mcp_server_auth_headers=mcp_server_auth_headers,
+        exc=exc,
+    ):
+        raise _aggregate_gateway_dcr_challenge(request, invalid_token=bearer_presented) from exc
+    raise exc
+
+
 class MCPRequestHandler:
     """
     Class to handle MCP request processing, including:
@@ -271,56 +365,32 @@ class MCPRequestHandler:
         elif oauth2_headers:
             # Authorization on a non-delegated server: the bearer must be a real
             # LiteLLM credential, so a failed validation is a genuine 401/403 and
-            # propagates. The sole anonymous fallback is the auth_type=none
-            # pass-through cold-start (RFC 9728 discovery return), gated on a 401
-            # so a recognized-but-forbidden key still fails closed.
-            client_ip = IPAddressUtils.get_mcp_client_ip(request)
+            # propagates unless a fallback in _admission_failure_fallback applies.
             try:
                 validated_user_api_key_auth = await user_api_key_auth(api_key=litellm_api_key, request=request)
             except (HTTPException, ProxyException) as e:
-                # ProxyException.code is normalized to str (possibly "None"), so
-                # compare both int and str forms rather than coercing.
-                status = e.status_code if isinstance(e, HTTPException) else e.code
-                is_unauthenticated = status in (401, "401")
-                mcp_servers_from_path = _parse_mcp_server_names_from_path(request_route, mcp_servers)
-                if (
-                    is_unauthenticated
-                    and mcp_servers_from_path is not None
-                    and not _has_client_supplied_mcp_auth(
-                        mcp_auth_header,
-                        mcp_server_auth_headers,
-                    )
-                    and _is_mcp_passthrough_cold_start(mcp_servers_from_path, client_ip=client_ip)
-                ):
-                    verbose_logger.debug(
-                        "MCP pass-through return: forwarding Authorization as upstream OAuth token for delegated auth"
-                    )
-                    validated_user_api_key_auth = UserAPIKeyAuth()
-                else:
-                    raise
+                validated_user_api_key_auth = _admission_failure_fallback(
+                    request=request,
+                    request_route=request_route,
+                    mcp_servers=mcp_servers,
+                    mcp_auth_header=mcp_auth_header,
+                    mcp_server_auth_headers=mcp_server_auth_headers,
+                    exc=e,
+                    bearer_presented=True,
+                )
         else:
             try:
                 validated_user_api_key_auth = await user_api_key_auth(api_key=litellm_api_key, request=request)
             except (HTTPException, ProxyException) as exc:
-                # Cold-start MCP OAuth discovery: RFC 9728 / MCP Authorization spec
-                # require unauthenticated requests to protected resources to receive
-                # 401 + WWW-Authenticate. Defer to _raise_preemptive_401_for_unauthenticated_servers
-                # for pass-through servers instead of surfacing a generic admission error.
-                mcp_servers_from_path = _parse_mcp_server_names_from_path(request_route, mcp_servers)
-                client_ip = IPAddressUtils.get_mcp_client_ip(request)
-                if (
-                    mcp_servers_from_path is not None
-                    and not _has_client_supplied_mcp_auth(
-                        mcp_auth_header,
-                        mcp_server_auth_headers,
-                    )
-                    and _is_litellm_auth_admission_error(exc)
-                    and _is_mcp_passthrough_cold_start(mcp_servers_from_path, client_ip=client_ip)
-                ):
-                    verbose_logger.debug("MCP pass-through cold start: deferring admission to route 401 emitter")
-                    validated_user_api_key_auth = UserAPIKeyAuth()
-                else:
-                    raise
+                validated_user_api_key_auth = _admission_failure_fallback(
+                    request=request,
+                    request_route=request_route,
+                    mcp_servers=mcp_servers,
+                    mcp_auth_header=mcp_auth_header,
+                    mcp_server_auth_headers=mcp_server_auth_headers,
+                    exc=exc,
+                    bearer_presented=False,
+                )
 
         return (
             validated_user_api_key_auth,

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -42,6 +42,7 @@ from litellm.proxy._experimental.mcp_server.faults import (
 from litellm.proxy._experimental.mcp_server.oauth_utils import (
     TOKEN_NO_CACHE_HEADERS,
     get_request_base_url,
+    is_mcp_gateway_dcr_enabled,
     validate_trusted_redirect_uri,
 )
 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
@@ -1708,6 +1709,12 @@ async def _build_oauth_protected_resource_response(
         global_mcp_server_manager,
     )
 
+    # With the gateway-level DCR front door enabled, unnamed discovery
+    # describes the gateway itself as the authorization server for the
+    # aggregate /mcp resource instead of narrowing to one server.
+    if mcp_server_name is None and is_mcp_gateway_dcr_enabled():
+        return _build_aggregate_protected_resource_response(request)
+
     request_base_url = get_request_base_url(request)
     client_ip = IPAddressUtils.get_mcp_client_ip(request)
 
@@ -1838,6 +1845,92 @@ def _jwt_auth_issuers() -> list:
     return issuers
 
 
+def _build_aggregate_protected_resource_response(request: Request) -> dict:
+    """RFC 9728 metadata for the aggregate /mcp resource: the gateway itself is
+    the authorization server. No per-server names or scopes leak here; access
+    is resolved after sign-in from the authenticated user's grants.
+
+    The advertised authorization server is ``{base}/mcp`` (not the bare
+    origin) so RFC 8414 path-insertion resolves its metadata at
+    ``/.well-known/oauth-authorization-server/mcp``, a route this module
+    owns. The bare-origin well-known is registered first by the BYOK OAuth
+    feature and describes the BYOK flow, so it must not be the aggregate
+    discovery entry point (same pattern as the per-server documents, which
+    advertise ``{base}/{server_name}``)."""
+    request_base_url = get_request_base_url(request)
+    return {
+        "authorization_servers": [f"{request_base_url}/mcp"],
+        "resource": f"{request_base_url}/mcp",
+        "scopes_supported": [],
+    }
+
+
+def _build_aggregate_authorization_server_response(request: Request) -> dict:
+    """RFC 8414 metadata for the gateway as the aggregate authorization server.
+
+    The issuer is ``{base}/mcp`` and must stay equal to the value the
+    aggregate protected-resource document advertises: spec clients verify the
+    issuer in the metadata matches the one that derived the well-known URL.
+    Advertises the root /authorize, /token, and /register endpoints and
+    ``token_endpoint_auth_methods_supported: ["none", ...]`` because DCR
+    clients (Claude Desktop, MCP Inspector) register as public clients; PKCE
+    S256 is mandatory in the gateway's authorize flow."""
+    request_base_url = get_request_base_url(request)
+    return {
+        "issuer": f"{request_base_url}/mcp",
+        "authorization_endpoint": f"{request_base_url}/authorize",
+        "token_endpoint": f"{request_base_url}/token",
+        "registration_endpoint": f"{request_base_url}/register",
+        "response_types_supported": ["code"],
+        "scopes_supported": [],
+        "grant_types_supported": ["authorization_code", "refresh_token"],
+        "code_challenge_methods_supported": ["S256"],
+        "token_endpoint_auth_methods_supported": ["none", "client_secret_post"],
+    }
+
+
+def _raise_404_unless_gateway_dcr_enabled() -> None:
+    """The aggregate well-known routes exist only under the gateway-level DCR
+    front door; flag-off they 404 exactly like the previously-absent routes so
+    discovery behavior is byte-identical for existing deployments."""
+    if is_mcp_gateway_dcr_enabled():
+        return
+    raise HTTPException(status_code=404, detail="Not Found")
+
+
+# RFC 9728 path-appended discovery for the aggregate /mcp endpoint. A client
+# pointed at {base}/mcp inserts the well-known segment before the resource
+# path, so this exact route must exist for aggregate discovery to work at all.
+# Declared before the parameterized well-known routes below: Starlette matches
+# in registration order, and /.well-known/oauth-authorization-server/{name}
+# would otherwise capture the "/mcp" suffix as a server name.
+@router.get(
+    f"/.well-known/oauth-protected-resource{'' if get_server_root_path() == '/' else get_server_root_path()}/mcp"
+)
+async def oauth_protected_resource_aggregate(request: Request):
+    """
+    OAuth protected resource discovery for the aggregate /mcp endpoint
+    (gateway-level DCR front door; 404 when the flag is off).
+    """
+    _raise_404_unless_gateway_dcr_enabled()
+    return _build_aggregate_protected_resource_response(request)
+
+
+@router.get(
+    f"/.well-known/oauth-authorization-server{'' if get_server_root_path() == '/' else get_server_root_path()}/mcp"
+)
+async def oauth_authorization_server_aggregate(request: Request):
+    """
+    OAuth authorization server discovery for the aggregate /mcp endpoint, the
+    RFC 8414 path-inserted form for a client that treats {base}/mcp as its
+    authorization base URL (gateway-level DCR front door; 404 when the flag
+    is off, indistinguishable from an unknown server name on the
+    parameterized route below).
+    """
+    _raise_404_unless_gateway_dcr_enabled()
+    return _build_aggregate_authorization_server_response(request)
+
+
 # Standard MCP pattern: /.well-known/oauth-protected-resource/mcp/{server_name}
 # This is the pattern expected by standard MCP clients (mcp-inspector, VSCode Copilot)
 @router.get(
@@ -1897,6 +1990,11 @@ def _build_oauth_authorization_server_response(
         global_mcp_server_manager,
     )
 
+    # With the gateway-level DCR front door enabled, unnamed discovery keeps
+    # advertising the gateway's own /authorize, /token, and /register.
+    if mcp_server_name is None and is_mcp_gateway_dcr_enabled():
+        return _build_aggregate_authorization_server_response(request)
+
     request_base_url = get_request_base_url(request)
     client_ip = IPAddressUtils.get_mcp_client_ip(request)
 

litellm/proxy/_experimental/mcp_server/oauth_utils.py

@@ -70,6 +70,29 @@ def _origin_label(scheme: str, netloc: str) -> str:
     return f"{scheme}://{netloc}" if netloc else f"{scheme}://"
 
 
+MCP_GATEWAY_DCR_SETTING = "mcp_gateway_dcr"
+
+
+def is_mcp_gateway_dcr_enabled() -> bool:
+    """True when ``general_settings.mcp_gateway_dcr`` opts this deployment into
+    the gateway-level DCR front door for the aggregate ``/mcp`` endpoint: root
+    OAuth discovery advertises the gateway itself as the authorization server
+    (instead of resolving the single configured oauth2 server), and the
+    anonymous aggregate 401 carries the RFC 9728 ``resource_metadata``
+    challenge so DCR clients (Claude Desktop, MCP Inspector) can start the
+    sign-in flow. Off by default; flag-off behavior is unchanged."""
+    from litellm.proxy.proxy_server import general_settings  # noqa: PLC0415  # circular import at module load
+
+    if not isinstance(general_settings, dict):
+        return False
+    raw = general_settings.get(MCP_GATEWAY_DCR_SETTING)
+    if isinstance(raw, bool):
+        return raw
+    if isinstance(raw, str):
+        return raw.strip().lower() == "true"
+    return False
+
+
 def _resolve_proxy_base_url_env() -> Optional[str]:
     global _warned_invalid_proxy_base_url
     configured = os.environ.get("PROXY_BASE_URL", "").strip()

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

@@ -6131,3 +6131,136 @@ class TestMCPDcrBridgeDelegateAdmission:
                     route="/mcp/bridge_delegate_server",
                 )
         assert exc_info.value.status_code == 500
+
+
+@pytest.mark.asyncio
+class TestAggregateGatewayDcrChallenge:
+    """The mcp_gateway_dcr front door: a 401 on the aggregate /mcp scope must
+    carry the RFC 9728 resource_metadata challenge pointing at the gateway's
+    own protected-resource metadata, and must NOT fire for named-server
+    targets, explicit litellm keys, non-401 failures, or with the flag off."""
+
+    _FLAG_PATCH_TARGET = "litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.is_mcp_gateway_dcr_enabled"
+    _AUTH_PATCH_TARGET = "litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.user_api_key_auth"
+    _EXPECTED_RESOURCE_METADATA = 'resource_metadata="http://testserver/.well-known/oauth-protected-resource/mcp"'
+
+    def _scope(self, path="/mcp", extra_headers=()):
+        return {
+            "type": "http",
+            "method": "POST",
+            "path": path,
+            "headers": [(b"host", b"testserver"), *extra_headers],
+        }
+
+    def _auth_401(self):
+        async def _raise(api_key, request):
+            raise ProxyException(
+                message="Authentication Error: Invalid API key",
+                type="auth_error",
+                param="api_key",
+                code=401,
+            )
+
+        return _raise
+
+    async def test_challenge_on_anonymous_aggregate_mcp(self):
+        """Anonymous request to the aggregate /mcp with the flag on: 401 plus
+        the bare bearer challenge (no error attribute, RFC 6750 section 3.1)."""
+        with (
+            patch(self._AUTH_PATCH_TARGET, side_effect=self._auth_401()),
+            patch(self._FLAG_PATCH_TARGET, return_value=True),
+        ):
+            with pytest.raises(HTTPException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(self._scope())
+        assert exc_info.value.status_code == 401
+        www_authenticate = (exc_info.value.headers or {})["WWW-Authenticate"]
+        assert www_authenticate == f"Bearer {self._EXPECTED_RESOURCE_METADATA}"
+
+    async def test_challenge_invalid_token_on_failed_bearer(self):
+        """A bearer that fails LiteLLM admission at aggregate scope (an expired
+        gateway session, a revoked key) re-challenges with error=invalid_token
+        so a spec client re-authorizes instead of retrying the dead token."""
+        with (
+            patch(self._AUTH_PATCH_TARGET, side_effect=self._auth_401()),
+            patch(self._FLAG_PATCH_TARGET, return_value=True),
+        ):
+            with pytest.raises(HTTPException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(
+                    self._scope(extra_headers=((b"authorization", b"Bearer expired-session-token"),))
+                )
+        assert exc_info.value.status_code == 401
+        www_authenticate = (exc_info.value.headers or {})["WWW-Authenticate"]
+        assert www_authenticate == f'Bearer error="invalid_token", {self._EXPECTED_RESOURCE_METADATA}'
+
+    async def test_no_challenge_when_flag_off(self):
+        """Flag off: the original admission error propagates untouched, both
+        with and without a bearer."""
+        for extra_headers in ((), ((b"authorization", b"Bearer some-token"),)):
+            with (
+                patch(self._AUTH_PATCH_TARGET, side_effect=self._auth_401()),
+                patch(self._FLAG_PATCH_TARGET, return_value=False),
+            ):
+                with pytest.raises(ProxyException) as exc_info:
+                    await MCPRequestHandler.process_mcp_request(self._scope(extra_headers=extra_headers))
+            assert str(exc_info.value.code) == "401"
+
+    async def test_no_challenge_for_explicit_litellm_key(self):
+        """An explicit x-litellm-api-key declares a litellm-key client; a typo
+        there must surface the real auth error, never a DCR challenge that
+        would send SDKs into a sign-in flow."""
+        with (
+            patch(self._AUTH_PATCH_TARGET, side_effect=self._auth_401()),
+            patch(self._FLAG_PATCH_TARGET, return_value=True),
+        ):
+            with pytest.raises(ProxyException):
+                await MCPRequestHandler.process_mcp_request(
+                    self._scope(extra_headers=((b"x-litellm-api-key", b"sk-typo"),))
+                )
+
+    async def test_no_challenge_for_named_servers_header(self):
+        """x-mcp-servers names explicit targets; the per-server challenge paths
+        own those, so the aggregate challenge must not fire."""
+        with (
+            patch(self._AUTH_PATCH_TARGET, side_effect=self._auth_401()),
+            patch(self._FLAG_PATCH_TARGET, return_value=True),
+        ):
+            with pytest.raises(ProxyException):
+                await MCPRequestHandler.process_mcp_request(
+                    self._scope(extra_headers=((b"x-mcp-servers", b"github"),))
+                )
+
+    async def test_no_challenge_for_path_named_server(self):
+        """/mcp/{server} targets one server; the aggregate challenge must not
+        fire even when that server does not resolve."""
+        with (
+            patch(self._AUTH_PATCH_TARGET, side_effect=self._auth_401()),
+            patch(self._FLAG_PATCH_TARGET, return_value=True),
+        ):
+            with pytest.raises(ProxyException):
+                await MCPRequestHandler.process_mcp_request(self._scope(path="/mcp/github"))
+
+    async def test_no_challenge_for_client_supplied_mcp_auth(self):
+        """Per-server x-mcp-{alias}-authorization headers mean the caller is
+        not a cold-start DCR client; keep the original error."""
+        with (
+            patch(self._AUTH_PATCH_TARGET, side_effect=self._auth_401()),
+            patch(self._FLAG_PATCH_TARGET, return_value=True),
+        ):
+            with pytest.raises(ProxyException):
+                await MCPRequestHandler.process_mcp_request(
+                    self._scope(extra_headers=((b"x-mcp-github-authorization", b"Bearer upstream"),))
+                )
+
+    async def test_no_challenge_for_non_401_failure(self):
+        """Only genuine 401s convert to a challenge; a 500 stays a 500."""
+
+        async def _raise_500(api_key, request):
+            raise ProxyException(message="boom", type="server_error", param=None, code=500)
+
+        with (
+            patch(self._AUTH_PATCH_TARGET, side_effect=_raise_500),
+            patch(self._FLAG_PATCH_TARGET, return_value=True),
+        ):
+            with pytest.raises(ProxyException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(self._scope())
+        assert str(exc_info.value.code) == "500"

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

@@ -7132,371 +7132,166 @@ async def test_token_exchange_unreadable_body_still_renders_oauth_fault():
     assert body == {"error": "server_error", "error_description": "upstream token endpoint returned HTTP 400"}
 
 
+def _patch_gateway_dcr_flag(enabled: bool):
+    return patch(
+        "litellm.proxy._experimental.mcp_server.discoverable_endpoints.is_mcp_gateway_dcr_enabled",
+        return_value=enabled,
+    )
+
+
 @pytest.mark.asyncio
-async def test_persist_dcr_client_for_config_server_uses_side_store():
-    """A config.yaml-declared OAuth2 DCR server has no LiteLLM_MCPServerTable row, so
-    update_mcp_server returns None. The minted client must then persist to the server-scoped
-    OAuth-client store keyed by server_id (never a shadow server row), overlay onto the in-memory
-    server so refresh can authenticate this process, and never call update_server(None) (which
-    previously raised AttributeError on .approval_status, was swallowed, and reported a 200 that
-    persisted nothing)."""
+async def test_gateway_dcr_root_discovery_describes_gateway_not_single_server():
+    """Flag on: root discovery must keep describing the gateway as the
+    authorization server for the aggregate /mcp resource even when exactly one
+    OAuth2 server exists (flag off, resolution narrows to that server; that
+    behavior is pinned by test_discovery_root_includes_server_name_prefix)."""
+    from fastapi import Request
+
     from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
-        _persist_dcr_client_registration,
+        _build_oauth_authorization_server_response,
+        _build_oauth_protected_resource_response,
     )
     from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
         global_mcp_server_manager,
     )
-    from litellm.proxy._types import MCPTransport
-    from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
-    config_server = MCPServer(
-        server_id="config_faros",
-        name="config_faros",
-        server_name="config_faros",
-        transport=MCPTransport.http,
-        auth_type=MCPAuth.oauth2,
-        client_id=None,
-        client_secret=None,
-        authorization_url="https://provider.example/oauth/authorize",
-        token_url="https://provider.example/oauth/token",
-        registration_url="https://provider.example/oauth/register",
-    )
+    global_mcp_server_manager.registry.clear()
+    oauth2_server = _create_oauth2_server()
+    global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
-    mock_upsert = AsyncMock()
-    mock_update_server = AsyncMock()
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://llm.example.com/"
+    mock_request.headers = {}
 
-    with (
-        patch.object(global_mcp_server_manager, "is_config_declared_server", return_value=True),
-        patch("litellm.proxy.utils.get_prisma_client_or_throw", return_value=MagicMock()),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.update_mcp_server",
-            new=AsyncMock(return_value=None),
-        ),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server",
-            new=AsyncMock(return_value=None),
-        ),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server_oauth_client_credentials",
-            new=AsyncMock(return_value=None),
-        ),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.upsert_mcp_server_oauth_client_credentials",
-            new=mock_upsert,
-        ),
-        patch.object(global_mcp_server_manager, "update_server", new=mock_update_server),
-    ):
-        result = await _persist_dcr_client_registration(
-            mcp_server=config_server,
-            registration_response={
-                "client_id": "minted-client",
-                "client_secret": "minted-secret",
-                "token_endpoint_auth_method": "client_secret_basic",
-            },
-            current_redirect_uri="https://proxy.litellm.example/callback",
-        )
+    try:
+        with _patch_gateway_dcr_flag(True):
+            authorization_response = _build_oauth_authorization_server_response(
+                request=mock_request,
+                mcp_server_name=None,
+            )
+            resource_response = await _build_oauth_protected_resource_response(
+                request=mock_request,
+                mcp_server_name=None,
+                use_standard_pattern=True,
+            )
 
-    assert result == "persisted"
+        assert authorization_response["issuer"] == "https://llm.example.com/mcp"
+        assert authorization_response["authorization_endpoint"] == "https://llm.example.com/authorize"
+        assert authorization_response["token_endpoint"] == "https://llm.example.com/token"
+        assert authorization_response["registration_endpoint"] == "https://llm.example.com/register"
+        assert "none" in authorization_response["token_endpoint_auth_methods_supported"]
+        assert authorization_response["code_challenge_methods_supported"] == ["S256"]
+        assert authorization_response["scopes_supported"] == []
 
-    mock_upsert.assert_called_once()
-    assert mock_upsert.call_args.kwargs["server_id"] == "config_faros"
-    stored = mock_upsert.call_args.kwargs["credentials"]
-    assert stored["client_id"] == "minted-client"
-    assert stored["client_secret"] == "minted-secret"
-    assert stored["token_endpoint_auth_method"] == "client_secret_basic"
-    assert stored["redirect_uris"] == ["https://proxy.litellm.example/callback"]
-
-    assert config_server.client_id == "minted-client"
-    assert config_server.client_secret == "minted-secret"
-    assert config_server.token_endpoint_auth_method == "client_secret_basic"
-
-    mock_update_server.assert_not_called()
+        assert resource_response["resource"] == "https://llm.example.com/mcp"
+        assert resource_response["authorization_servers"] == ["https://llm.example.com/mcp"]
+        assert resource_response["scopes_supported"] == []
+    finally:
+        global_mcp_server_manager.registry.clear()
 
 
 @pytest.mark.asyncio
-async def test_hydrate_config_server_applies_stored_dcr_client(monkeypatch):
-    """On restart a config server's in-memory object has no client_id; hydration overlays the
-    persisted DCR client from the server-scoped store, decrypting the encrypted-at-rest blob, so the
-    refresh_token grant can authenticate as the registered client instead of re-authenticating."""
-    import litellm.proxy.common_utils.encrypt_decrypt_utils as enc
-    from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
-    from litellm.proxy._experimental.mcp_server.db import encrypt_credentials
+async def test_gateway_dcr_named_discovery_unaffected_by_flag():
+    """Flag on must not change named-server discovery: a named oauth2 server
+    still resolves to its own per-server document."""
+    from fastapi import Request
+
     from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
-        hydrate_config_server_dcr_client,
-    )
-    from litellm.proxy._types import MCPTransport
-    from litellm.types.mcp_server.mcp_server_manager import MCPServer
-
-    server = MCPServer(
-        server_id="config_faros",
-        name="config_faros",
-        server_name="config_faros",
-        transport=MCPTransport.http,
-        auth_type=MCPAuth.oauth2,
-        client_id=None,
-    )
-
-    monkeypatch.setattr(enc, "_get_salt_key", lambda: "salt-hydrate-key")
-    stored_blob = safe_dumps(
-        encrypt_credentials(
-            credentials={
-                "client_id": "stored-client",
-                "client_secret": "stored-secret",
-                "token_endpoint_auth_method": "client_secret_basic",
-                "redirect_uris": ["https://proxy.litellm.example/callback"],
-            },
-            encryption_key="salt-hydrate-key",
-        )
-    )
-    assert "stored-client" not in stored_blob and "stored-secret" not in stored_blob
-
-    with (
-        patch("litellm.proxy.utils.get_prisma_client_or_throw", return_value=MagicMock()),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server_oauth_client_credentials",
-            new=AsyncMock(return_value=stored_blob),
-        ),
-    ):
-        applied = await hydrate_config_server_dcr_client(server)
-
-    assert applied is True
-    assert server.client_id == "stored-client"
-    assert server.client_secret == "stored-secret"
-    assert server.token_endpoint_auth_method == "client_secret_basic"
-
-
-@pytest.mark.asyncio
-async def test_reuse_config_server_reads_store_with_real_crypto(monkeypatch):
-    """A config-declared server (rowless) keeps its DCR client in the store, so the reuse read
-    resolves it from the store and decrypts the encrypted-at-rest client, mirroring the write path so
-    a re-authorize reuses the client instead of re-minting one."""
-    import litellm.proxy.common_utils.encrypt_decrypt_utils as enc
-    from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
-    from litellm.proxy._experimental.mcp_server.db import encrypt_credentials
-    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
-        _reuse_persisted_dcr_client_if_available,
+        _build_oauth_authorization_server_response,
     )
     from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
         global_mcp_server_manager,
     )
-    from litellm.proxy._types import MCPTransport
-    from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
-    server = MCPServer(
-        server_id="config_faros",
-        name="config_faros",
-        server_name="config_faros",
-        transport=MCPTransport.http,
-        auth_type=MCPAuth.oauth2,
-        client_id=None,
-    )
+    global_mcp_server_manager.registry.clear()
+    oauth2_server = _create_oauth2_server()
+    global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
-    monkeypatch.setattr(enc, "_get_salt_key", lambda: "salt-reuse-key")
-    blob = safe_dumps(
-        encrypt_credentials(
-            credentials={"client_id": "stored-client", "client_secret": "sec", "redirect_uris": ["https://x/callback"]},
-            encryption_key="salt-reuse-key",
-        )
-    )
-    assert "stored-client" not in blob
-    store_lookup = AsyncMock(return_value=blob)
-    with (
-        patch.object(global_mcp_server_manager, "is_config_declared_server", return_value=True),
-        patch("litellm.proxy.utils.get_prisma_client_or_throw", return_value=MagicMock()),
-        patch("litellm.proxy._experimental.mcp_server.db.get_mcp_server", new=AsyncMock(return_value=None)),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server_oauth_client_credentials",
-            new=store_lookup,
-        ),
-    ):
-        result = await _reuse_persisted_dcr_client_if_available(server, current_redirect_uri="https://x/callback")
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://llm.example.com/"
+    mock_request.headers = {}
 
-    assert result is True
-    assert server.client_id == "stored-client"
-    store_lookup.assert_awaited_once()
+    try:
+        with _patch_gateway_dcr_flag(True):
+            response = _build_oauth_authorization_server_response(
+                request=mock_request,
+                mcp_server_name="test_oauth",
+            )
+        assert "/test_oauth/authorize" in response["authorization_endpoint"]
+        assert response["scopes_supported"] == ["read", "write"]
+    finally:
+        global_mcp_server_manager.registry.clear()
 
 
-@pytest.mark.asyncio
-async def test_temp_server_is_not_persisted_to_store():
-    """A rowless server that is NOT config-declared (a throwaway /server/oauth/session server) must
-    not leave a permanent store row on persist, and the read must never consult the store for it. Its
-    minted client is overlaid in memory for the session only."""
-    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
-        _persist_dcr_client_registration,
-        _reuse_persisted_dcr_client_if_available,
-    )
-    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
-        global_mcp_server_manager,
-    )
-    from litellm.proxy._types import MCPTransport
-    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+def test_aggregate_wellknown_routes_404_when_flag_off():
+    """Flag off, the aggregate well-known routes answer 404 exactly like the
+    previously-absent routes: discovery behavior is byte-identical for
+    existing deployments."""
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
 
-    temp = MCPServer(
-        server_id="temp-uuid",
-        name="temp",
-        server_name="temp",
-        transport=MCPTransport.http,
-        auth_type=MCPAuth.oauth2,
-        client_id=None,
-        authorization_url="https://p.example/authorize",
-        token_url="https://p.example/token",
-        registration_url="https://p.example/register",
-    )
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import router
 
-    upsert = AsyncMock()
-    store_read = AsyncMock(return_value=None)
-    with (
-        patch.object(global_mcp_server_manager, "is_config_declared_server", return_value=False),
-        patch("litellm.proxy.utils.get_prisma_client_or_throw", return_value=MagicMock()),
-        patch("litellm.proxy._experimental.mcp_server.db.update_mcp_server", new=AsyncMock(return_value=None)),
-        patch("litellm.proxy._experimental.mcp_server.db.get_mcp_server", new=AsyncMock(return_value=None)),
-        patch("litellm.proxy._experimental.mcp_server.db.upsert_mcp_server_oauth_client_credentials", new=upsert),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server_oauth_client_credentials",
-            new=store_read,
-        ),
-        patch.object(global_mcp_server_manager, "update_server", new=AsyncMock()),
-    ):
-        result = await _persist_dcr_client_registration(
-            temp, {"client_id": "temp-client", "client_secret": "s"}, "https://x/callback"
-        )
-        reused = await _reuse_persisted_dcr_client_if_available(
-            MCPServer(
-                server_id="temp-uuid",
-                name="temp",
-                server_name="temp",
-                transport=MCPTransport.http,
-                auth_type=MCPAuth.oauth2,
-                client_id=None,
-            ),
-            current_redirect_uri="https://x/callback",
-        )
+    app = FastAPI()
+    app.include_router(router)
+    client = TestClient(app)
 
-    assert result == "persisted"
-    assert temp.client_id == "temp-client"
-    upsert.assert_not_called()
-    store_read.assert_not_called()
-    assert reused is False
+    with _patch_gateway_dcr_flag(False):
+        assert client.get("/.well-known/oauth-protected-resource/mcp").status_code == 404
+        assert client.get("/.well-known/oauth-authorization-server/mcp").status_code == 404
 
 
-@pytest.mark.asyncio
-async def test_hydrate_does_not_overwrite_explicit_config_client_id():
-    """An explicit client_id set in config.yaml wins: hydration must not overwrite it with a stale
-    persisted store client, and must not even read the store when config already supplied a client."""
-    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
-        hydrate_config_server_dcr_client,
-    )
-    from litellm.proxy._types import MCPTransport
-    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+def test_aggregate_wellknown_routes_serve_gateway_metadata_when_flag_on():
+    """Flag on, both path-appended aggregate routes serve the gateway
+    documents. Exercises real routing, so this also pins registration order:
+    /.well-known/oauth-authorization-server/{name} would otherwise capture
+    the /mcp suffix as a server name and 404."""
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
 
-    server = MCPServer(
-        server_id="config_static",
-        name="config_static",
-        server_name="config_static",
-        transport=MCPTransport.http,
-        auth_type=MCPAuth.oauth2,
-        client_id="explicit-from-config",
-    )
-    store_read = AsyncMock(
-        return_value={"client_id": "stale-store-client", "client_secret": "x", "redirect_uris": []}
-    )
-    with (
-        patch("litellm.proxy.utils.get_prisma_client_or_throw", return_value=MagicMock()),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server_oauth_client_credentials",
-            new=store_read,
-        ),
-    ):
-        applied = await hydrate_config_server_dcr_client(server)
-
-    assert applied is False
-    assert server.client_id == "explicit-from-config"
-    store_read.assert_not_called()
-
-
-@pytest.mark.asyncio
-async def test_reuse_does_not_inherit_store_client_when_a_row_exists():
-    """Security: a server that HAS a LiteLLM_MCPServerTable row reads its DCR client only from that
-    row, never from the server-scoped store. server_id is caller-settable on create, so a submitted
-    server whose id collides with a config-declared server must not be able to load that config
-    server's client from the store and send it to its own token endpoint. A row that exists but has
-    no client_id yields no reusable client and must not fall back to the store."""
-    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
-        _reuse_persisted_dcr_client_if_available,
-    )
-    from litellm.proxy._types import MCPTransport
-    from litellm.types.mcp_server.mcp_server_manager import MCPServer
-
-    submitted = MCPServer(
-        server_id="collides_with_config",
-        name="submitted",
-        server_name="submitted",
-        transport=MCPTransport.http,
-        auth_type=MCPAuth.oauth2,
-        client_id=None,
-    )
-
-    row_without_client = MagicMock()
-    row_without_client.credentials = None
-    row_without_client.server_id = "collides_with_config"
-    store_lookup = AsyncMock(
-        return_value={"client_id": "config-secret-client", "client_secret": "leak", "redirect_uris": []}
-    )
-    with (
-        patch("litellm.proxy.utils.get_prisma_client_or_throw", return_value=MagicMock()),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server",
-            new=AsyncMock(return_value=row_without_client),
-        ),
-        patch(
-            "litellm.proxy._experimental.mcp_server.db.get_mcp_server_oauth_client_credentials",
-            new=store_lookup,
-        ),
-    ):
-        result = await _reuse_persisted_dcr_client_if_available(submitted, current_redirect_uri="https://x/callback")
-
-    assert result is False
-    assert submitted.client_id is None
-    store_lookup.assert_not_called()
-
-
-@pytest.mark.asyncio
-async def test_load_servers_from_config_hydrates_dcr_clients():
-    """load_servers_from_config must invoke DCR-client hydration so config servers pick up their
-    persisted client on startup; deleting the call site leaves a restarted server with no client_id
-    and forces re-authentication on every token expiry."""
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import router
     from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
         global_mcp_server_manager,
     )
 
-    hydrate_spy = AsyncMock()
-    with patch.object(global_mcp_server_manager, "_hydrate_config_servers_dcr_clients", new=hydrate_spy):
-        await global_mcp_server_manager.load_servers_from_config({})
+    global_mcp_server_manager.registry.clear()
+    app = FastAPI()
+    app.include_router(router)
+    client = TestClient(app)
 
-    hydrate_spy.assert_awaited_once()
+    with _patch_gateway_dcr_flag(True):
+        prm = client.get("/.well-known/oauth-protected-resource/mcp")
+        asm = client.get("/.well-known/oauth-authorization-server/mcp")
+
+    assert prm.status_code == 200
+    assert prm.json()["resource"] == "http://testserver/mcp"
+    assert prm.json()["authorization_servers"] == ["http://testserver/mcp"]
+
+    assert asm.status_code == 200
+    assert asm.json()["issuer"] == "http://testserver/mcp"
+    assert asm.json()["authorization_endpoint"] == "http://testserver/authorize"
 
 
-@pytest.mark.asyncio
-async def test_reload_servers_from_database_hydrates_dcr_clients():
-    """load_servers_from_config runs before the DB connects at startup, so its hydration no-ops;
-    reload_servers_from_database runs after the DB connects and must hydrate config servers' persisted
-    DCR clients too, or a fresh pod has no client_id for a config server and forces re-authentication
-    on the first token refresh."""
-    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
-        global_mcp_server_manager,
+def test_is_mcp_gateway_dcr_enabled_reads_general_settings():
+    """The flag reader accepts YAML booleans and env-interpolated strings, and
+    fails closed on anything else."""
+    from litellm.proxy._experimental.mcp_server.oauth_utils import (
+        is_mcp_gateway_dcr_enabled,
     )
+    from litellm.proxy.proxy_server import general_settings
 
-    prisma = MagicMock()
-    prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[])
-
-    hydrate_spy = AsyncMock()
-    with (
-        patch(
-            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
-            return_value=prisma,
-        ),
-        patch.object(global_mcp_server_manager, "_hydrate_config_servers_dcr_clients", new=hydrate_spy),
+    for raw, expected in (
+        (True, True),
+        (False, False),
+        ("true", True),
+        ("True", True),
+        ("false", False),
+        ("yes", False),
+        (1, False),
+        (None, False),
     ):
-        await global_mcp_server_manager.reload_servers_from_database()
+        with patch.dict(general_settings, {"mcp_gateway_dcr": raw}):
+            assert is_mcp_gateway_dcr_enabled() is expected, f"raw={raw!r}"
 
-    hydrate_spy.assert_awaited_once()
+    with patch.dict(general_settings, {}, clear=True):
+        assert is_mcp_gateway_dcr_enabled() is False