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
@@ -1640,6 +1641,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)
 
@@ -1770,6 +1777,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(
@@ -1829,6 +1922,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

@@ -5950,3 +5950,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

@@ -7130,3 +7130,168 @@ async def test_token_exchange_unreadable_body_still_renders_oauth_fault():
     assert response.status_code == 502
     body = json.loads(response.body)
     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_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 (
+        _build_oauth_authorization_server_response,
+        _build_oauth_protected_resource_response,
+    )
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
+
+    global_mcp_server_manager.registry.clear()
+    oauth2_server = _create_oauth2_server()
+    global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://llm.example.com/"
+    mock_request.headers = {}
+
+    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 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"] == []
+
+        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_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 (
+        _build_oauth_authorization_server_response,
+    )
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
+
+    global_mcp_server_manager.registry.clear()
+    oauth2_server = _create_oauth2_server()
+    global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://llm.example.com/"
+    mock_request.headers = {}
+
+    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()
+
+
+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
+
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import router
+
+    app = FastAPI()
+    app.include_router(router)
+    client = TestClient(app)
+
+    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
+
+
+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
+
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import router
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
+
+    global_mcp_server_manager.registry.clear()
+    app = FastAPI()
+    app.include_router(router)
+    client = TestClient(app)
+
+    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"
+
+
+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
+
+    for raw, expected in (
+        (True, True),
+        (False, False),
+        ("true", True),
+        ("True", True),
+        ("false", False),
+        ("yes", False),
+        (1, False),
+        (None, False),
+    ):
+        with patch.dict(general_settings, {"mcp_gateway_dcr": raw}):
+            assert is_mcp_gateway_dcr_enabled() is expected, f"raw={raw!r}"
+
+    with patch.dict(general_settings, {}, clear=True):
+        assert is_mcp_gateway_dcr_enabled() is False