From 66f012a06bb448f459bb8f7c90a044abfaabc8d3 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Tue, 14 Jul 2026 18:16:45 -0700
Subject: [PATCH 1/6] fix(mcp): discover missing OAuth scopes and token_url
 when authorization_url is set manually

---
 .../mcp_server/mcp_server_manager.py          |  5 +-
 .../mcp_server/test_mcp_server_manager.py     | 60 ++++++++++++++++++-
 2 files changed, 62 insertions(+), 3 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index e6e265abb61..fda6ae9bef5 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -1447,8 +1447,9 @@ class MCPServerManager:
 
         auth_type = cast(MCPAuthType, mcp_server.auth_type)
         server_url = mcp_server.url
+        has_all_upstream_oauth_fields = bool(mcp_server.authorization_url and mcp_server.token_url and scopes)
         needs_discovery = bool(server_url) and (
-            (auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES and not mcp_server.authorization_url)
+            (auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES and not has_all_upstream_oauth_fields)
             or self._obo_needs_endpoint_discovery(
                 auth_type,
                 mcp_server.token_exchange_endpoint
@@ -1467,7 +1468,7 @@ class MCPServerManager:
         if needs_discovery and mcp_oauth_metadata is None:
             verbose_logger.warning(
                 "MCP OAuth discovery yielded no metadata for server %s (%s); "
-                "OAuth endpoints stay unresolved until a rebuild succeeds",
+                "OAuth endpoints/scopes stay unresolved until a rebuild succeeds",
                 mcp_server.server_id,
                 server_url,
             )
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index adcfff6fe9d..6af86b9bda3 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -1026,7 +1026,6 @@ class TestMCPServerManager:
         """The gateway's relayed authorize flow (used by the browser-only Authorize) needs the
         upstream's authorization_url on the registry entry, and these rows never persist one, so
         the DB build must discover it the same way oauth2 rows do."""
-        from types import SimpleNamespace
 
         manager = MCPServerManager()
         row = LiteLLM_MCPServerTable(
@@ -1053,6 +1052,65 @@ class TestMCPServerManager:
         assert built.authorization_url == "https://idp.example.com/authorize"
         assert built.token_url == "https://idp.example.com/token"
 
+    @pytest.mark.asyncio
+    async def test_build_from_table_discovers_scopes_when_authorization_url_is_manual(self):
+        """An admin-typed authorization_url must not switch off discovery for the fields left
+        blank: without the scopes_supported backfill the authorize redirect goes out scope-less
+        and IdPs like Google hard-fail it with 400 "Missing required parameter: scope"."""
+        manager = MCPServerManager()
+        row = LiteLLM_MCPServerTable(
+            server_id="manual-auth-url-1",
+            alias="manual_auth_url",
+            description="manual authorization_url, blank scopes",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://idp.example.com/manual-authorize",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+        )
+
+        metadata = MCPOAuthMetadata(
+            authorization_url="https://idp.example.com/discovered-authorize",
+            token_url="https://idp.example.com/token",
+            registration_url=None,
+            scopes=["calendar.read", "calendar.write"],
+        )
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)) as mock_discovery:
+            built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
+
+        mock_discovery.assert_awaited_once()
+        assert built.authorization_url == "https://idp.example.com/manual-authorize"
+        assert built.token_url == "https://idp.example.com/token"
+        assert built.scopes == ["calendar.read", "calendar.write"]
+
+    @pytest.mark.asyncio
+    async def test_build_from_table_skips_discovery_when_all_upstream_oauth_fields_present(self):
+        """A fully hand-configured server (authorization_url, token_url, and scopes all set) has
+        nothing left for discovery to fill, so the build must not fetch upstream metadata."""
+        manager = MCPServerManager()
+        row = LiteLLM_MCPServerTable(
+            server_id="fully-manual-1",
+            alias="fully_manual",
+            description="all upstream oauth fields set by the admin",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://idp.example.com/manual-authorize",
+            token_url="https://idp.example.com/manual-token",
+            credentials={"scopes": ["calendar.read"]},
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+        )
+
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)) as mock_discovery:
+            built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
+
+        mock_discovery.assert_not_awaited()
+        assert built.authorization_url == "https://idp.example.com/manual-authorize"
+        assert built.token_url == "https://idp.example.com/manual-token"
+        assert built.scopes == ["calendar.read"]
+
     async def _capture_subject_token(self, call) -> Optional[str]:
         """Run a manager method (via ``call(manager)``) and return the subject_token it threaded
         into ``_create_mcp_client``."""

From a81c6ce350c2fd38d43a15c22edc5c4078f19675 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 15 Jul 2026 10:22:24 -0700
Subject: [PATCH 2/6] fix(mcp): reject discovered token endpoints
 uncorroborated by the manual authorization_url

Discovery is rooted at the MCP resource, so a compromised upstream can
advertise an attacker-run authorization server. When authorization_url is
manually configured and another field is blank, the per-field merge would
combine the trusted authorize endpoint with the advertised token_url, and
the gateway would redeem authorization codes (with the stored client secret
and PKCE verifier) at that endpoint, then persist it. Discovered token_url
and registration_url are now accepted only when the same metadata document
advertises an authorization_endpoint matching the configured value
(scheme+host+path). Scope backfill is unaffected. Applies to both the DB
and config build paths.
---
 .../mcp_server/mcp_server_manager.py          |  90 +++++++++--
 .../mcp_server/test_mcp_server_manager.py     | 144 +++++++++++++++++-
 2 files changed, 221 insertions(+), 13 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index fda6ae9bef5..284bf6a0926 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -213,6 +213,57 @@ def _carry_forward_resolved_oauth_endpoints(new_server: MCPServer, previous_serv
         new_server.scopes = previous_server.scopes
 
 
+def _normalized_authorize_endpoint(url: str) -> str:
+    parsed = urlparse(url)
+    return f"{parsed.scheme.lower()}://{parsed.netloc.lower()}{parsed.path.rstrip('/')}"
+
+
+def _gate_discovered_endpoints_against_manual_authorization_url(
+    metadata: MCPOAuthMetadata | None,
+    manual_authorization_url: str | None,
+    server_identifier: str,
+    is_dcr_bridge: bool,
+) -> MCPOAuthMetadata | None:
+    """Refuse discovered token/registration endpoints the pinned authorize endpoint cannot vouch for.
+
+    Discovery is rooted at the MCP resource (RFC 9728), so a compromised upstream can advertise an
+    attacker-run authorization server. When the admin manually configured ``authorization_url``,
+    filling a blank ``token_url`` from that advertisement recreates the RFC 9700 mix-up attack at
+    configuration time: users sign in at the trusted authorize endpoint while the gateway redeems
+    the code, with the stored client secret and PKCE verifier, at the attacker's token endpoint.
+    Endpoints from one metadata document are only trustworthy together, so the discovered
+    ``token_url`` and ``registration_url`` are accepted only when that same document's
+    ``authorization_endpoint`` matches the pinned value (scheme+host+path; query and trailing slash
+    are not identity). Scope discovery stays ungated: scopes steer the redirect to the trusted
+    authorize endpoint and carry no credentials.
+    """
+    if metadata is None or not manual_authorization_url:
+        return metadata
+    if not metadata.token_url and not metadata.registration_url:
+        return metadata
+    if metadata.authorization_url and _normalized_authorize_endpoint(
+        manual_authorization_url
+    ) == _normalized_authorize_endpoint(metadata.authorization_url):
+        return metadata
+    bridge_note = (
+        " The discovered registration_url is rejected with it, so this dcr_bridge server stays on the"
+        " short-circuit registration arm."
+        if is_dcr_bridge and metadata.registration_url
+        else ""
+    )
+    verbose_logger.warning(
+        "MCP OAuth discovery for server %s advertised authorization_endpoint %s, which does not match the "
+        "manually configured authorization_url %s; rejecting the discovered token_url/registration_url so "
+        "authorization codes and client credentials only go to endpoints vouched for by the configured "
+        "authorization server. Configure Token URL manually if the mismatch is intentional.%s",
+        server_identifier,
+        _normalized_authorize_endpoint(metadata.authorization_url) if metadata.authorization_url else "<absent>",
+        _normalized_authorize_endpoint(manual_authorization_url),
+        bridge_note,
+    )
+    return metadata.model_copy(update={"token_url": None, "registration_url": None})
+
+
 def invalidate_user_env_vars_cache(user_id: str, server_id: str) -> None:
     """Drop a cached entry after the user stores or clears their env var values
     so the next request reads the fresh value instead of a stale one."""
@@ -1041,20 +1092,31 @@ class MCPServerManager:
             else:
                 mcp_oauth_metadata = None
 
+            gated_oauth_metadata = (
+                _gate_discovered_endpoints_against_manual_authorization_url(
+                    mcp_oauth_metadata,
+                    server_config.get("authorization_url"),
+                    server_name or server_id,
+                    bool(server_config.get("dcr_bridge")),
+                )
+                if auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
+                else mcp_oauth_metadata
+            )
+
             # Filter blank scopes (e.g. YAML ``scopes: [""]``) the same way the DB-build path does, so
             # an all-blank list normalizes to None rather than a ``("",)`` tuple that skips the
             # entra_obo fail-closed scope precondition and POSTs an empty scope to the IdP.
             resolved_scopes = self._extract_scopes(server_config.get("scopes")) or (
-                mcp_oauth_metadata.scopes if mcp_oauth_metadata else None
+                gated_oauth_metadata.scopes if gated_oauth_metadata else None
             )
             resolved_authorization_url = server_config.get("authorization_url") or (
-                mcp_oauth_metadata.authorization_url if mcp_oauth_metadata else None
+                gated_oauth_metadata.authorization_url if gated_oauth_metadata else None
             )
             resolved_token_url = server_config.get("token_url") or (
-                mcp_oauth_metadata.token_url if mcp_oauth_metadata else None
+                gated_oauth_metadata.token_url if gated_oauth_metadata else None
             )
             resolved_registration_url = server_config.get("registration_url") or (
-                mcp_oauth_metadata.registration_url if mcp_oauth_metadata else None
+                gated_oauth_metadata.registration_url if gated_oauth_metadata else None
             )
 
             config_oauth2_flow = server_config.get("oauth2_flow", None)
@@ -1472,8 +1534,18 @@ class MCPServerManager:
                 mcp_server.server_id,
                 server_url,
             )
+        gated_oauth_metadata = (
+            _gate_discovered_endpoints_against_manual_authorization_url(
+                mcp_oauth_metadata,
+                mcp_server.authorization_url,
+                mcp_server.server_id,
+                bool(getattr(mcp_server, "dcr_bridge", None)),
+            )
+            if auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
+            else mcp_oauth_metadata
+        )
 
-        resolved_scopes = scopes or (mcp_oauth_metadata.scopes if mcp_oauth_metadata else None)
+        resolved_scopes = scopes or (gated_oauth_metadata.scopes if gated_oauth_metadata else None)
 
         new_server = MCPServer(
             server_id=mcp_server.server_id,
@@ -1493,9 +1565,9 @@ class MCPServerManager:
             client_secret=client_secret_value or getattr(mcp_server, "client_secret", None),
             oauth2_flow=self._explicit_oauth2_flow(getattr(mcp_server, "oauth2_flow", None)),
             scopes=resolved_scopes,
-            authorization_url=mcp_server.authorization_url or getattr(mcp_oauth_metadata, "authorization_url", None),
-            token_url=mcp_server.token_url or getattr(mcp_oauth_metadata, "token_url", None),
-            registration_url=mcp_server.registration_url or getattr(mcp_oauth_metadata, "registration_url", None),
+            authorization_url=mcp_server.authorization_url or getattr(gated_oauth_metadata, "authorization_url", None),
+            token_url=mcp_server.token_url or getattr(gated_oauth_metadata, "token_url", None),
+            registration_url=mcp_server.registration_url or getattr(gated_oauth_metadata, "registration_url", None),
             token_endpoint_auth_method=(
                 credentials_dict.get("token_endpoint_auth_method") if credentials_dict else None
             ),
@@ -1555,7 +1627,7 @@ class MCPServerManager:
                 existing_authorization_url=mcp_server.authorization_url,
                 existing_token_url=mcp_server.token_url,
                 existing_scopes=scopes,
-                metadata=mcp_oauth_metadata,
+                metadata=gated_oauth_metadata,
             )
         return new_server
 
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index 6af86b9bda3..8091675090e 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -356,6 +356,55 @@ class TestMCPServerManager:
         assert server.oauth2_flow == "authorization_code"
         assert server.needs_user_oauth_token is True
 
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_rejects_discovered_token_url_on_authorization_endpoint_mismatch(self):
+        """The config loader always runs discovery and or-merges per field, so a yaml server with a
+        manual authorization_url has the same config-time mix-up exposure as a DB row: a discovered
+        token_url from a document advertising a different authorize endpoint must not be combined
+        with the pinned one. Scopes still backfill."""
+        manager = MCPServerManager()
+
+        metadata = MCPOAuthMetadata(
+            authorization_url="https://attacker.example.com/authorize",
+            token_url="https://attacker.example.com/token",
+            scopes=["read"],
+        )
+        config = self._oauth2_config(
+            oauth2_flow="authorization_code",
+            authorization_url="https://idp.example.com/authorize",
+            token_url=None,
+        )
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
+            await manager.load_servers_from_config(config)
+
+        server = next(iter(manager.config_mcp_servers.values()))
+        assert server.authorization_url == "https://idp.example.com/authorize"
+        assert server.token_url is None
+        assert server.scopes == ["read"]
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_fills_token_url_when_metadata_corroborates_manual_authorization_url(self):
+        """Corroborated metadata keeps the self-heal on the config path: when the discovered
+        document advertises the same authorize endpoint the admin pinned, its token_url fills the
+        blank field."""
+        manager = MCPServerManager()
+
+        metadata = MCPOAuthMetadata(
+            authorization_url="https://idp.example.com/authorize",
+            token_url="https://idp.example.com/token",
+            scopes=["read"],
+        )
+        config = self._oauth2_config(
+            oauth2_flow="authorization_code",
+            authorization_url="https://idp.example.com/authorize/",
+            token_url=None,
+        )
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
+            await manager.load_servers_from_config(config)
+
+        server = next(iter(manager.config_mcp_servers.values()))
+        assert server.token_url == "https://idp.example.com/token"
+
     @pytest.mark.asyncio
     async def test_load_servers_from_config_non_oauth2_needs_no_flow(self):
         manager = MCPServerManager()
@@ -1056,7 +1105,9 @@ class TestMCPServerManager:
     async def test_build_from_table_discovers_scopes_when_authorization_url_is_manual(self):
         """An admin-typed authorization_url must not switch off discovery for the fields left
         blank: without the scopes_supported backfill the authorize redirect goes out scope-less
-        and IdPs like Google hard-fail it with 400 "Missing required parameter: scope"."""
+        and IdPs like Google hard-fail it with 400 "Missing required parameter: scope". Scope
+        backfill works even when the advertised authorization_endpoint differs from the manual
+        value, because scopes only steer the redirect to the trusted authorize endpoint."""
         manager = MCPServerManager()
         row = LiteLLM_MCPServerTable(
             server_id="manual-auth-url-1",
@@ -1081,9 +1132,90 @@ class TestMCPServerManager:
 
         mock_discovery.assert_awaited_once()
         assert built.authorization_url == "https://idp.example.com/manual-authorize"
-        assert built.token_url == "https://idp.example.com/token"
+        assert built.token_url is None
         assert built.scopes == ["calendar.read", "calendar.write"]
 
+    @pytest.mark.asyncio
+    async def test_build_from_table_fills_endpoints_when_metadata_corroborates_manual_authorization_url(self):
+        """A discovered token_url is only trusted next to a manual authorization_url when the same
+        metadata document advertises that authorize endpoint, and the comparison must tolerate
+        formatting-only differences (host case, trailing slash, query params like ?prompt=consent)
+        so hand-copied URLs still self-heal."""
+        manager = MCPServerManager()
+        row = LiteLLM_MCPServerTable(
+            server_id="manual-auth-url-2",
+            alias="manual_auth_url_match",
+            description="manual authorization_url matching discovery, blank token_url",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://IDP.example.com/authorize/?prompt=consent",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+        )
+
+        metadata = MCPOAuthMetadata(
+            authorization_url="https://idp.example.com/authorize",
+            token_url="https://idp.example.com/token",
+            registration_url="https://idp.example.com/register",
+            scopes=["read"],
+        )
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
+            built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
+
+        assert built.authorization_url == "https://IDP.example.com/authorize/?prompt=consent"
+        assert built.token_url == "https://idp.example.com/token"
+        assert built.registration_url == "https://idp.example.com/register"
+        assert built.scopes == ["read"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "advertised_authorization_url",
+        ["https://attacker.example.com/authorize", None],
+    )
+    async def test_build_from_table_rejects_discovered_token_url_on_authorization_endpoint_mismatch(
+        self, advertised_authorization_url
+    ):
+        """Resource-rooted discovery lets a compromised upstream advertise its own authorization
+        server. With a manual authorization_url pinned, accepting that document's token_url would
+        send the authorization code, stored client secret, and PKCE verifier to the attacker's
+        token endpoint (config-time RFC 9700 mix-up), and the persist hook would make the hostile
+        endpoint durable. Both the in-memory merge and the persisted metadata must drop the
+        uncorroborated token_url and registration_url."""
+        manager = MCPServerManager()
+        row = LiteLLM_MCPServerTable(
+            server_id="manual-auth-url-3",
+            alias="manual_auth_url_mismatch",
+            description="manual authorization_url, hostile discovery document",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://idp.example.com/authorize",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+        )
+
+        metadata = MCPOAuthMetadata(
+            authorization_url=advertised_authorization_url,
+            token_url="https://attacker.example.com/token",
+            registration_url="https://attacker.example.com/register",
+            scopes=["read"],
+        )
+        with (
+            patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)),
+            patch.object(manager, "_persist_discovered_oauth_endpoints", new=AsyncMock()) as mock_persist,
+        ):
+            built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
+
+        assert built.authorization_url == "https://idp.example.com/authorize"
+        assert built.token_url is None
+        assert built.registration_url is None
+        assert built.scopes == ["read"]
+        persisted_metadata = mock_persist.await_args.kwargs["metadata"]
+        assert persisted_metadata.token_url is None
+        assert persisted_metadata.registration_url is None
+        assert persisted_metadata.scopes == ["read"]
+
     @pytest.mark.asyncio
     async def test_build_from_table_skips_discovery_when_all_upstream_oauth_fields_present(self):
         """A fully hand-configured server (authorization_url, token_url, and scopes all set) has
@@ -2310,6 +2442,10 @@ class TestMCPServerManager:
 
     @pytest.mark.asyncio
     async def test_load_servers_from_config_overrides_discovery_metadata(self):
+        """Config values win per field. The discovered token_url/registration_url do NOT fill the
+        blanks here: the document advertises a different authorization_endpoint than the manually
+        configured one, so combining its endpoints with the pinned authorize URL would be the
+        config-time mix-up the discovery gate exists to prevent."""
         manager = MCPServerManager()
 
         discovered_metadata = MCPOAuthMetadata(
@@ -2343,8 +2479,8 @@ class TestMCPServerManager:
         server = next(iter(manager.config_mcp_servers.values()))
         assert server.scopes == ["config"]  # config overrides discovery
         assert server.authorization_url == "https://config.example.com/auth"
-        assert server.token_url == "https://discovered.example.com/token"
-        assert server.registration_url == "https://discovered.example.com/register"
+        assert server.token_url is None
+        assert server.registration_url is None
 
     @pytest.mark.asyncio
     async def test_load_servers_from_config_filters_blank_scopes(self):

From 447d50fa4041c3d6dcbfdaa6fa4b6f89ea2eaaee Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 15 Jul 2026 10:39:04 -0700
Subject: [PATCH 3/6] fix(mcp): enforce the OAuth endpoint trust rule at
 carry-forward too, elide default port

The corroboration check belongs to adopting a token_url from any non-manual
source, not to discovery alone. Carry-forward is the other such source: it
copied a prior registry entry's token_url/registration_url onto a rebuild
whose authorization_url had been re-pointed to a different server, reviving an
uncorroborated token endpoint the discovery gate would reject. Both sites now
share one predicate, _endpoints_corroborate_authorization_url: previous
endpoints carry forward only when the previous authorization_url corroborates
the authorize endpoint the build will use (absent -> the previous one is
adopted too, a consistent group; else it must match). Endpoint comparison now
elides the default port so :443 and formatting-only differences still match.
---
 .../mcp_server/mcp_server_manager.py          | 95 ++++++++++++------
 .../mcp_server/test_mcp_server_manager.py     | 96 +++++++++++++++++++
 2 files changed, 160 insertions(+), 31 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index 284bf6a0926..b84c5601560 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -186,6 +186,46 @@ _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES: tuple[MCPAuth, ...] = (
 )
 
 
+def _normalized_authorize_endpoint(url: str) -> str:
+    """Compare authorize endpoints on scheme, host, and path only. The default port is elided and
+    the host is lowercased so ``https://IDP.example.com:443/authorize/`` and
+    ``https://idp.example.com/authorize`` are the same identity; query and trailing slash are not."""
+    parsed = urlparse(url)
+    scheme = parsed.scheme.lower()
+    host = (parsed.hostname or "").lower()
+    default_port = {"https": 443, "http": 80}.get(scheme)
+    try:
+        port = parsed.port
+    except ValueError:
+        port = None
+    authority = host if port is None or port == default_port else f"{host}:{port}"
+    return f"{scheme}://{authority}{parsed.path.rstrip('/')}"
+
+
+def _endpoints_corroborate_authorization_url(
+    source_authorization_url: str | None,
+    trusted_authorization_url: str | None,
+) -> bool:
+    """Whether a source's ``token_url``/``registration_url`` may be paired with a trusted authorize
+    endpoint. This is the single trust rule for adopting OAuth endpoints from any non-manual source.
+
+    Discovery is rooted at the MCP resource (RFC 9728), so a compromised upstream can advertise an
+    attacker-run authorization server. When ``authorization_url`` is admin-pinned, pairing it with a
+    ``token_url`` from a different source is the RFC 9700 authorization-server mix-up: the user signs
+    in at the trusted authorize endpoint while the gateway redeems the code, with the stored client
+    secret and PKCE verifier, at the attacker's token endpoint. Endpoints are trustworthy together
+    only when they share an authorization server, so a source's endpoints are adopted only when the
+    same source advertised an ``authorization_endpoint`` matching the pinned value. With no pinned
+    value (``trusted_authorization_url is None``) there is nothing to protect: the authorize endpoint
+    comes from the same source as the token endpoint, so they corroborate each other by construction.
+    """
+    if trusted_authorization_url is None:
+        return True
+    return bool(source_authorization_url) and _normalized_authorize_endpoint(
+        source_authorization_url
+    ) == _normalized_authorize_endpoint(trusted_authorization_url)
+
+
 def _carry_forward_resolved_oauth_endpoints(new_server: MCPServer, previous_server: MCPServer | None) -> None:
     """Keep the last known good OAuth endpoints when a rebuild's re-discovery comes back empty.
 
@@ -193,57 +233,50 @@ def _carry_forward_resolved_oauth_endpoints(new_server: MCPServer, previous_serv
     during re-discovery downgrades a working server (``authorization_url`` set) to a broken one
     (``None``, /authorize 400s) with no configuration change. Mirrors the ``short_prefix``
     carry-forward. Skipped when the server's ``url`` or ``auth_type`` changed, since the previous
-    endpoints may then belong to a different upstream. ``registration_url`` IS carried here even
-    though ``_persist_discovered_oauth_endpoints`` refuses to write it to the row: carrying only
-    restores the same in-memory value the previous build already ran with, while persisting it
-    would flip ``_dcr_bridge_relays_client_registration`` (which keys off the stored column) for
-    dcr_bridge servers that never had one configured.
+    endpoints may then belong to a different upstream. ``registration_url`` IS carried even though
+    ``_persist_discovered_oauth_endpoints`` refuses to write it to the row: carrying only restores
+    the same in-memory value the previous build already ran with, while persisting it would flip
+    ``_dcr_bridge_relays_client_registration`` (which keys off the stored column) for dcr_bridge
+    servers that never had one configured.
+
+    Carry-forward is a non-manual endpoint source, so the same trust rule as discovery applies: the
+    previous ``token_url``/``registration_url`` are carried only when the previous
+    ``authorization_url`` corroborates the authorize endpoint this build will use, i.e. when the
+    incoming build has no pinned authorize endpoint (``None`` -> we adopt the previous one too, a
+    consistent group) or pins the same one. An admin re-pointing ``authorization_url`` to a different
+    server must not keep serving the old token endpoint.
     """
     if previous_server is None:
         return
     if previous_server.url != new_server.url or previous_server.auth_type != new_server.auth_type:
         return
+    may_carry_endpoints = _endpoints_corroborate_authorization_url(
+        previous_server.authorization_url, new_server.authorization_url
+    )
     if new_server.authorization_url is None and previous_server.authorization_url:
         new_server.authorization_url = previous_server.authorization_url
-    if new_server.token_url is None and previous_server.token_url:
+    if may_carry_endpoints and new_server.token_url is None and previous_server.token_url:
         new_server.token_url = previous_server.token_url
-    if new_server.registration_url is None and previous_server.registration_url:
+    if may_carry_endpoints and new_server.registration_url is None and previous_server.registration_url:
         new_server.registration_url = previous_server.registration_url
     if not new_server.scopes and previous_server.scopes:
         new_server.scopes = previous_server.scopes
 
 
-def _normalized_authorize_endpoint(url: str) -> str:
-    parsed = urlparse(url)
-    return f"{parsed.scheme.lower()}://{parsed.netloc.lower()}{parsed.path.rstrip('/')}"
-
-
 def _gate_discovered_endpoints_against_manual_authorization_url(
     metadata: MCPOAuthMetadata | None,
     manual_authorization_url: str | None,
     server_identifier: str,
     is_dcr_bridge: bool,
 ) -> MCPOAuthMetadata | None:
-    """Refuse discovered token/registration endpoints the pinned authorize endpoint cannot vouch for.
-
-    Discovery is rooted at the MCP resource (RFC 9728), so a compromised upstream can advertise an
-    attacker-run authorization server. When the admin manually configured ``authorization_url``,
-    filling a blank ``token_url`` from that advertisement recreates the RFC 9700 mix-up attack at
-    configuration time: users sign in at the trusted authorize endpoint while the gateway redeems
-    the code, with the stored client secret and PKCE verifier, at the attacker's token endpoint.
-    Endpoints from one metadata document are only trustworthy together, so the discovered
-    ``token_url`` and ``registration_url`` are accepted only when that same document's
-    ``authorization_endpoint`` matches the pinned value (scheme+host+path; query and trailing slash
-    are not identity). Scope discovery stays ungated: scopes steer the redirect to the trusted
-    authorize endpoint and carry no credentials.
+    """Apply :func:`_endpoints_corroborate_authorization_url` to freshly discovered metadata, the
+    other non-manual endpoint source. Drops the discovered ``token_url``/``registration_url`` (never
+    the scopes, which carry no credentials) when they cannot be vouched for by the pinned authorize
+    endpoint, logging why so an intentional mismatch can be resolved by setting Token URL by hand.
     """
-    if metadata is None or not manual_authorization_url:
+    if metadata is None or (not metadata.token_url and not metadata.registration_url):
         return metadata
-    if not metadata.token_url and not metadata.registration_url:
-        return metadata
-    if metadata.authorization_url and _normalized_authorize_endpoint(
-        manual_authorization_url
-    ) == _normalized_authorize_endpoint(metadata.authorization_url):
+    if _endpoints_corroborate_authorization_url(metadata.authorization_url, manual_authorization_url):
         return metadata
     bridge_note = (
         " The discovered registration_url is rejected with it, so this dcr_bridge server stays on the"
@@ -258,7 +291,7 @@ def _gate_discovered_endpoints_against_manual_authorization_url(
         "authorization server. Configure Token URL manually if the mismatch is intentional.%s",
         server_identifier,
         _normalized_authorize_endpoint(metadata.authorization_url) if metadata.authorization_url else "<absent>",
-        _normalized_authorize_endpoint(manual_authorization_url),
+        _normalized_authorize_endpoint(manual_authorization_url) if manual_authorization_url else "<absent>",
         bridge_note,
     )
     return metadata.model_copy(update={"token_url": None, "registration_url": None})
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index 8091675090e..229b1bfb014 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -5287,6 +5287,102 @@ class TestMCPServerTimestamps:
         _carry_forward_resolved_oauth_endpoints(new_server=explicit, previous_server=previous)
         assert explicit.authorization_url == "https://configured.example.com/auth"
 
+    def test_carry_forward_does_not_revive_token_url_across_authorization_url_change(self):
+        """Carry-forward is a non-manual endpoint source, so it obeys the same trust rule as
+        discovery: a previous token_url/registration_url belongs to the previous authorization
+        server, so it must not be pinned to a NEW authorization_url the admin re-pointed to. Without
+        this, re-pointing authorize to server B while the same MCP url keeps serving A's token
+        endpoint recreates the RFC 9700 mix-up, durably, and the discovery gate alone cannot catch
+        it because the stale endpoint comes from the registry, not from discovery."""
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            _carry_forward_resolved_oauth_endpoints,
+        )
+
+        previous = MCPServer(
+            server_id="s1",
+            name="s1",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://idp-a.example.com/authorize",
+            token_url="https://idp-a.example.com/token",
+            registration_url="https://idp-a.example.com/register",
+        )
+        repointed = MCPServer(
+            server_id="s1",
+            name="s1",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://idp-b.example.com/authorize",
+        )
+
+        _carry_forward_resolved_oauth_endpoints(new_server=repointed, previous_server=previous)
+
+        assert repointed.authorization_url == "https://idp-b.example.com/authorize"
+        assert repointed.token_url is None
+        assert repointed.registration_url is None
+
+    def test_carry_forward_restores_endpoints_when_authorization_url_unchanged(self):
+        """The last-known-good path still works: a rebuild whose discovery blipped (no authorize
+        endpoint) adopts the previous authorize endpoint AND its token endpoint together as a
+        consistent group, and a rebuild that re-pins the same authorize endpoint (formatting aside)
+        keeps carrying the corroborated token endpoint."""
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            _carry_forward_resolved_oauth_endpoints,
+        )
+
+        def previous() -> MCPServer:
+            return MCPServer(
+                server_id="s1",
+                name="s1",
+                url="https://up.example.com/mcp",
+                transport=MCPTransport.http,
+                auth_type=MCPAuth.oauth2,
+                authorization_url="https://idp.example.com/authorize",
+                token_url="https://idp.example.com/token",
+                registration_url="https://idp.example.com/register",
+            )
+
+        blipped = MCPServer(
+            server_id="s1",
+            name="s1",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url=None,
+        )
+        _carry_forward_resolved_oauth_endpoints(new_server=blipped, previous_server=previous())
+        assert blipped.authorization_url == "https://idp.example.com/authorize"
+        assert blipped.token_url == "https://idp.example.com/token"
+        assert blipped.registration_url == "https://idp.example.com/register"
+
+        same_authorize = MCPServer(
+            server_id="s1",
+            name="s1",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="https://IDP.example.com:443/authorize/",
+        )
+        _carry_forward_resolved_oauth_endpoints(new_server=same_authorize, previous_server=previous())
+        assert same_authorize.token_url == "https://idp.example.com/token"
+        assert same_authorize.registration_url == "https://idp.example.com/register"
+
+    def test_normalized_authorize_endpoint_treats_default_port_and_slash_as_identity(self):
+        """The corroboration check must not fail on formatting-only differences an IdP legitimately
+        emits: default port, trailing slash, host case, and query string are not identity, but a
+        non-default port is."""
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            _normalized_authorize_endpoint,
+        )
+
+        canonical = _normalized_authorize_endpoint("https://idp.example.com/authorize")
+        assert _normalized_authorize_endpoint("https://idp.example.com:443/authorize") == canonical
+        assert _normalized_authorize_endpoint("https://IDP.example.com/authorize/") == canonical
+        assert _normalized_authorize_endpoint("https://idp.example.com/authorize?prompt=consent") == canonical
+        assert _normalized_authorize_endpoint("https://idp.example.com:8443/authorize") != canonical
+
     def test_build_mcp_server_table_preserves_timestamps(self):
         """_build_mcp_server_table must use the MCPServer's stored timestamps, not datetime.now()."""
         manager = MCPServerManager()

From 8650f6c7d352ca079fe82fba3a84f1f5adcf3c91 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 15 Jul 2026 10:57:40 -0700
Subject: [PATCH 4/6] fix(mcp): bound pinned-config discovery to the
 corroborated authorization server, scopes included

Provenance is a property of the whole discovered metadata document, not per
field. Waving scopes through while gating endpoints left a second inflation
vector: a compromised upstream advertises broad scopes via the resource
metadata (RFC 9728 / WWW-Authenticate), the gateway requests them from the
trusted authorization server, and the resulting token flows back to the
upstream. Both that and the token-endpoint mix-up are now one rule: when
authorization_url is admin-pinned, discovered token_url/registration_url are
kept only if the document corroborates the pin, and scopes come from the
authorization server's own scopes_supported (a new authorization_server_scopes
field, trusted tier) rather than the resource-advertised scopes. A document
that does not corroborate backfills nothing. Blank (empty-string)
authorization_url is treated as unpinned so the merge and the gate agree.
Carry-forward, the other non-manual source, drops the same three across an
authorization_url change.
---
 .../mcp_server/mcp_server_manager.py          |  51 ++++---
 .../types/mcp_server/mcp_server_manager.py    |   9 ++
 .../mcp_server/test_mcp_server_manager.py     | 137 ++++++++++++++----
 3 files changed, 147 insertions(+), 50 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index b84c5601560..2ace0d9b16a 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -219,7 +219,7 @@ def _endpoints_corroborate_authorization_url(
     value (``trusted_authorization_url is None``) there is nothing to protect: the authorize endpoint
     comes from the same source as the token endpoint, so they corroborate each other by construction.
     """
-    if trusted_authorization_url is None:
+    if not (trusted_authorization_url and trusted_authorization_url.strip()):
         return True
     return bool(source_authorization_url) and _normalized_authorize_endpoint(
         source_authorization_url
@@ -240,43 +240,53 @@ def _carry_forward_resolved_oauth_endpoints(new_server: MCPServer, previous_serv
     servers that never had one configured.
 
     Carry-forward is a non-manual endpoint source, so the same trust rule as discovery applies: the
-    previous ``token_url``/``registration_url`` are carried only when the previous
+    previous ``token_url``/``registration_url``/``scopes`` are carried only when the previous
     ``authorization_url`` corroborates the authorize endpoint this build will use, i.e. when the
     incoming build has no pinned authorize endpoint (``None`` -> we adopt the previous one too, a
     consistent group) or pins the same one. An admin re-pointing ``authorization_url`` to a different
-    server must not keep serving the old token endpoint.
+    server must not keep serving the old server's token endpoint or granted scopes.
     """
     if previous_server is None:
         return
     if previous_server.url != new_server.url or previous_server.auth_type != new_server.auth_type:
         return
-    may_carry_endpoints = _endpoints_corroborate_authorization_url(
+    may_carry = _endpoints_corroborate_authorization_url(
         previous_server.authorization_url, new_server.authorization_url
     )
     if new_server.authorization_url is None and previous_server.authorization_url:
         new_server.authorization_url = previous_server.authorization_url
-    if may_carry_endpoints and new_server.token_url is None and previous_server.token_url:
+    if may_carry and new_server.token_url is None and previous_server.token_url:
         new_server.token_url = previous_server.token_url
-    if may_carry_endpoints and new_server.registration_url is None and previous_server.registration_url:
+    if may_carry and new_server.registration_url is None and previous_server.registration_url:
         new_server.registration_url = previous_server.registration_url
-    if not new_server.scopes and previous_server.scopes:
+    if may_carry and not new_server.scopes and previous_server.scopes:
         new_server.scopes = previous_server.scopes
 
 
-def _gate_discovered_endpoints_against_manual_authorization_url(
+def _restrict_discovery_to_corroborated_authorization_server(
     metadata: MCPOAuthMetadata | None,
     manual_authorization_url: str | None,
     server_identifier: str,
     is_dcr_bridge: bool,
 ) -> MCPOAuthMetadata | None:
-    """Apply :func:`_endpoints_corroborate_authorization_url` to freshly discovered metadata, the
-    other non-manual endpoint source. Drops the discovered ``token_url``/``registration_url`` (never
-    the scopes, which carry no credentials) when they cannot be vouched for by the pinned authorize
-    endpoint, logging why so an intentional mismatch can be resolved by setting Token URL by hand.
+    """Bound what freshly discovered metadata may backfill into a manually pinned config.
+
+    Discovery is rooted at the MCP resource, so provenance is a property of the whole metadata
+    document, not per field: a compromised upstream can advertise both an attacker ``token_endpoint``
+    (the RFC 9700 mix-up) and inflated ``scopes`` (tricking the user into granting a broader token
+    that then flows to the upstream). Both are closed by one rule. When ``authorization_url`` is
+    admin-pinned, the discovered ``token_url`` and ``registration_url`` are kept only if the document
+    corroborates the pin (its ``authorization_endpoint`` matches), and scopes are taken from the
+    authorization server's own ``scopes_supported`` (``authorization_server_scopes``, trusted tier)
+    rather than the resource-advertised ``scopes`` a compromised upstream controls. A document that
+    does not corroborate backfills nothing. With no pin there is no trust anchor to protect and the
+    authorize endpoint comes from the same chain as everything else, so discovery is returned as-is.
     """
-    if metadata is None or (not metadata.token_url and not metadata.registration_url):
+    if metadata is None or not (manual_authorization_url and manual_authorization_url.strip()):
         return metadata
     if _endpoints_corroborate_authorization_url(metadata.authorization_url, manual_authorization_url):
+        return metadata.model_copy(update={"scopes": metadata.authorization_server_scopes})
+    if not metadata.token_url and not metadata.registration_url and not metadata.scopes:
         return metadata
     bridge_note = (
         " The discovered registration_url is rejected with it, so this dcr_bridge server stays on the"
@@ -286,15 +296,15 @@ def _gate_discovered_endpoints_against_manual_authorization_url(
     )
     verbose_logger.warning(
         "MCP OAuth discovery for server %s advertised authorization_endpoint %s, which does not match the "
-        "manually configured authorization_url %s; rejecting the discovered token_url/registration_url so "
-        "authorization codes and client credentials only go to endpoints vouched for by the configured "
-        "authorization server. Configure Token URL manually if the mismatch is intentional.%s",
+        "manually configured authorization_url %s; rejecting the discovered token_url/registration_url/scopes "
+        "so authorization codes, client credentials, and granted scopes only follow the configured "
+        "authorization server. Configure Token URL and Scopes manually if the mismatch is intentional.%s",
         server_identifier,
         _normalized_authorize_endpoint(metadata.authorization_url) if metadata.authorization_url else "<absent>",
-        _normalized_authorize_endpoint(manual_authorization_url) if manual_authorization_url else "<absent>",
+        _normalized_authorize_endpoint(manual_authorization_url),
         bridge_note,
     )
-    return metadata.model_copy(update={"token_url": None, "registration_url": None})
+    return metadata.model_copy(update={"token_url": None, "registration_url": None, "scopes": None})
 
 
 def invalidate_user_env_vars_cache(user_id: str, server_id: str) -> None:
@@ -1126,7 +1136,7 @@ class MCPServerManager:
                 mcp_oauth_metadata = None
 
             gated_oauth_metadata = (
-                _gate_discovered_endpoints_against_manual_authorization_url(
+                _restrict_discovery_to_corroborated_authorization_server(
                     mcp_oauth_metadata,
                     server_config.get("authorization_url"),
                     server_name or server_id,
@@ -1568,7 +1578,7 @@ class MCPServerManager:
                 server_url,
             )
         gated_oauth_metadata = (
-            _gate_discovered_endpoints_against_manual_authorization_url(
+            _restrict_discovery_to_corroborated_authorization_server(
                 mcp_oauth_metadata,
                 mcp_server.authorization_url,
                 mcp_server.server_id,
@@ -3365,6 +3375,7 @@ class MCPServerManager:
                 authorization_url=data.get("authorization_endpoint"),
                 token_url=data.get("token_endpoint"),
                 registration_url=data.get("registration_endpoint"),
+                authorization_server_scopes=scopes,
             )
 
             if any(
diff --git a/litellm/types/mcp_server/mcp_server_manager.py b/litellm/types/mcp_server/mcp_server_manager.py
index 82ff15303f3..79a523f7c8b 100644
--- a/litellm/types/mcp_server/mcp_server_manager.py
+++ b/litellm/types/mcp_server/mcp_server_manager.py
@@ -17,9 +17,18 @@ MCPInfo = Dict[str, Any]
 
 class MCPOAuthMetadata(BaseModel):
     scopes: Optional[List[str]] = None
+    """Effective scopes, resource-preferred: the RFC 9728 protected-resource advertisement or the
+    WWW-Authenticate challenge when the resource supplied one, else the authorization server's
+    ``scopes_supported``. A compromised resource server can influence this, so it must not expand a
+    manually pinned ``authorization_url`` (see ``authorization_server_scopes``)."""
     authorization_url: Optional[str] = None
     token_url: Optional[str] = None
     registration_url: Optional[str] = None
+    authorization_server_scopes: Optional[List[str]] = None
+    """The ``scopes_supported`` enumerated by the authorization-server metadata document itself
+    (RFC 8414), independent of anything the resource server advertised. This is the only scope
+    source trusted to backfill a manually pinned ``authorization_url``, because it shares provenance
+    with the ``authorization_endpoint`` used to corroborate that pin."""
     from_origin_fallback: bool = False
     """True when the metadata came from guessing the resource origin as its authorization
     server rather than from an RFC 9728/8414-advertised document. Guessed endpoints are
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index 229b1bfb014..d14ae561f71 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -357,17 +357,18 @@ class TestMCPServerManager:
         assert server.needs_user_oauth_token is True
 
     @pytest.mark.asyncio
-    async def test_load_servers_from_config_rejects_discovered_token_url_on_authorization_endpoint_mismatch(self):
+    async def test_load_servers_from_config_rejects_uncorroborated_discovery_including_scopes(self):
         """The config loader always runs discovery and or-merges per field, so a yaml server with a
-        manual authorization_url has the same config-time mix-up exposure as a DB row: a discovered
-        token_url from a document advertising a different authorize endpoint must not be combined
-        with the pinned one. Scopes still backfill."""
+        manual authorization_url has the same config-time mix-up exposure as a DB row: a document
+        advertising a different authorize endpoint backfills nothing, neither its token_url nor its
+        scopes."""
         manager = MCPServerManager()
 
         metadata = MCPOAuthMetadata(
             authorization_url="https://attacker.example.com/authorize",
             token_url="https://attacker.example.com/token",
-            scopes=["read"],
+            scopes=["read", "admin"],
+            authorization_server_scopes=["read", "admin"],
         )
         config = self._oauth2_config(
             oauth2_flow="authorization_code",
@@ -380,19 +381,20 @@ class TestMCPServerManager:
         server = next(iter(manager.config_mcp_servers.values()))
         assert server.authorization_url == "https://idp.example.com/authorize"
         assert server.token_url is None
-        assert server.scopes == ["read"]
+        assert server.scopes is None
 
     @pytest.mark.asyncio
     async def test_load_servers_from_config_fills_token_url_when_metadata_corroborates_manual_authorization_url(self):
         """Corroborated metadata keeps the self-heal on the config path: when the discovered
         document advertises the same authorize endpoint the admin pinned, its token_url fills the
-        blank field."""
+        blank field and scopes come from the authorization server's own scopes_supported."""
         manager = MCPServerManager()
 
         metadata = MCPOAuthMetadata(
             authorization_url="https://idp.example.com/authorize",
             token_url="https://idp.example.com/token",
-            scopes=["read"],
+            scopes=["read", "admin"],
+            authorization_server_scopes=["read"],
         )
         config = self._oauth2_config(
             oauth2_flow="authorization_code",
@@ -404,6 +406,34 @@ class TestMCPServerManager:
 
         server = next(iter(manager.config_mcp_servers.values()))
         assert server.token_url == "https://idp.example.com/token"
+        assert server.scopes == ["read"]
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_blank_authorization_url_is_not_a_pin(self):
+        """A blank (empty-string) authorization_url is not a trust anchor, so discovery backfills
+        the whole set — authorize endpoint, token_url, and its resource-preferred scopes — from the
+        same chain, exactly as if the field had been omitted. The corroboration gate must treat
+        empty-string as unpinned so it does not strand the token_url the merge still fills."""
+        manager = MCPServerManager()
+
+        metadata = MCPOAuthMetadata(
+            authorization_url="https://idp.example.com/authorize",
+            token_url="https://idp.example.com/token",
+            scopes=["read"],
+            authorization_server_scopes=["read"],
+        )
+        config = self._oauth2_config(
+            oauth2_flow="authorization_code",
+            authorization_url="",
+            token_url=None,
+        )
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
+            await manager.load_servers_from_config(config)
+
+        server = next(iter(manager.config_mcp_servers.values()))
+        assert server.authorization_url == "https://idp.example.com/authorize"
+        assert server.token_url == "https://idp.example.com/token"
+        assert server.scopes == ["read"]
 
     @pytest.mark.asyncio
     async def test_load_servers_from_config_non_oauth2_needs_no_flow(self):
@@ -1102,12 +1132,12 @@ class TestMCPServerManager:
         assert built.token_url == "https://idp.example.com/token"
 
     @pytest.mark.asyncio
-    async def test_build_from_table_discovers_scopes_when_authorization_url_is_manual(self):
-        """An admin-typed authorization_url must not switch off discovery for the fields left
-        blank: without the scopes_supported backfill the authorize redirect goes out scope-less
-        and IdPs like Google hard-fail it with 400 "Missing required parameter: scope". Scope
-        backfill works even when the advertised authorization_endpoint differs from the manual
-        value, because scopes only steer the redirect to the trusted authorize endpoint."""
+    async def test_build_from_table_backfills_scopes_from_authorization_server_not_resource(self):
+        """When authorization_url is admin-pinned, scopes backfill from the authorization server's
+        own scopes_supported (trusted tier), never from the resource-advertised scopes a compromised
+        upstream controls. Here the corroborating document carries an inflated resource `scopes`
+        (`admin`) alongside the real authorization_server_scopes; only the latter may be requested,
+        otherwise a hostile resource could trick the user into granting a broader token."""
         manager = MCPServerManager()
         row = LiteLLM_MCPServerTable(
             server_id="manual-auth-url-1",
@@ -1116,24 +1146,24 @@ class TestMCPServerManager:
             url="https://up.example.com/mcp",
             transport=MCPTransport.http,
             auth_type=MCPAuth.oauth2,
-            authorization_url="https://idp.example.com/manual-authorize",
+            authorization_url="https://idp.example.com/authorize",
             created_at=datetime.now(),
             updated_at=datetime.now(),
         )
 
         metadata = MCPOAuthMetadata(
-            authorization_url="https://idp.example.com/discovered-authorize",
+            authorization_url="https://idp.example.com/authorize",
             token_url="https://idp.example.com/token",
-            registration_url=None,
-            scopes=["calendar.read", "calendar.write"],
+            scopes=["read", "admin"],
+            authorization_server_scopes=["read", "write"],
         )
         with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)) as mock_discovery:
             built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
 
         mock_discovery.assert_awaited_once()
-        assert built.authorization_url == "https://idp.example.com/manual-authorize"
-        assert built.token_url is None
-        assert built.scopes == ["calendar.read", "calendar.write"]
+        assert built.authorization_url == "https://idp.example.com/authorize"
+        assert built.token_url == "https://idp.example.com/token"
+        assert built.scopes == ["read", "write"]
 
     @pytest.mark.asyncio
     async def test_build_from_table_fills_endpoints_when_metadata_corroborates_manual_authorization_url(self):
@@ -1159,6 +1189,7 @@ class TestMCPServerManager:
             token_url="https://idp.example.com/token",
             registration_url="https://idp.example.com/register",
             scopes=["read"],
+            authorization_server_scopes=["read"],
         )
         with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
             built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
@@ -1173,15 +1204,15 @@ class TestMCPServerManager:
         "advertised_authorization_url",
         ["https://attacker.example.com/authorize", None],
     )
-    async def test_build_from_table_rejects_discovered_token_url_on_authorization_endpoint_mismatch(
+    async def test_build_from_table_rejects_uncorroborated_discovery_including_scopes(
         self, advertised_authorization_url
     ):
         """Resource-rooted discovery lets a compromised upstream advertise its own authorization
-        server. With a manual authorization_url pinned, accepting that document's token_url would
-        send the authorization code, stored client secret, and PKCE verifier to the attacker's
-        token endpoint (config-time RFC 9700 mix-up), and the persist hook would make the hostile
-        endpoint durable. Both the in-memory merge and the persisted metadata must drop the
-        uncorroborated token_url and registration_url."""
+        server. With a manual authorization_url pinned, a document that does not corroborate it
+        backfills nothing: accepting its token_url would send the code, client secret, and PKCE
+        verifier to the attacker (config-time RFC 9700 mix-up), and accepting its scopes would let
+        the upstream inflate the granted token. Both the in-memory merge and the persisted metadata
+        must drop the uncorroborated token_url, registration_url, and scopes."""
         manager = MCPServerManager()
         row = LiteLLM_MCPServerTable(
             server_id="manual-auth-url-3",
@@ -1199,7 +1230,8 @@ class TestMCPServerManager:
             authorization_url=advertised_authorization_url,
             token_url="https://attacker.example.com/token",
             registration_url="https://attacker.example.com/register",
-            scopes=["read"],
+            scopes=["read", "admin"],
+            authorization_server_scopes=["read", "admin"],
         )
         with (
             patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)),
@@ -1210,11 +1242,11 @@ class TestMCPServerManager:
         assert built.authorization_url == "https://idp.example.com/authorize"
         assert built.token_url is None
         assert built.registration_url is None
-        assert built.scopes == ["read"]
+        assert built.scopes is None
         persisted_metadata = mock_persist.await_args.kwargs["metadata"]
         assert persisted_metadata.token_url is None
         assert persisted_metadata.registration_url is None
-        assert persisted_metadata.scopes == ["read"]
+        assert persisted_metadata.scopes is None
 
     @pytest.mark.asyncio
     async def test_build_from_table_skips_discovery_when_all_upstream_oauth_fields_present(self):
@@ -2317,6 +2349,48 @@ class TestMCPServerManager:
         assert result.scopes == ["api://some-scope/.default"]
         assert result.from_origin_fallback is False
 
+    @pytest.mark.asyncio
+    async def test_descovery_metadata_preserves_authorization_server_scopes_under_resource_override(self):
+        """The effective `scopes` field is resource-preferred (RFC 9728 / WWW-Authenticate), but the
+        authorization server's own scopes_supported must survive on `authorization_server_scopes` so
+        the pinned-config backfill can request the trusted-tier scopes instead of resource-advertised
+        ones. This is the provenance split the scope-inflation defense depends on."""
+        manager = MCPServerManager()
+
+        mock_response = MagicMock()
+        mock_response.raise_for_status = MagicMock()
+        mock_client = MagicMock()
+        mock_client.get = AsyncMock(return_value=mock_response)
+
+        authorization_server_metadata = MCPOAuthMetadata(
+            scopes=["as.read", "as.write"],
+            authorization_url="https://idp.example.com/authorize",
+            token_url="https://idp.example.com/token",
+            authorization_server_scopes=["as.read", "as.write"],
+        )
+
+        with (
+            patch(
+                "litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
+                return_value=mock_client,
+            ),
+            patch.object(
+                manager,
+                "_attempt_well_known_discovery",
+                AsyncMock(return_value=(["https://idp.example.com"], ["resource.only"])),
+            ),
+            patch.object(
+                manager,
+                "_fetch_authorization_server_metadata",
+                AsyncMock(return_value=authorization_server_metadata),
+            ),
+        ):
+            result = await manager._descovery_metadata("https://up.example.com/mcp")
+
+        assert result is not None
+        assert result.scopes == ["resource.only"]
+        assert result.authorization_server_scopes == ["as.read", "as.write"]
+
     @pytest.mark.asyncio
     async def test_fetch_single_authorization_server_metadata_supports_azure_issuer_path(
         self,
@@ -2357,6 +2431,9 @@ class TestMCPServerManager:
         assert result.authorization_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/authorize"
         assert result.token_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/token"
         assert result.scopes == ["api://some-scope/.default"]
+        # The authorization server's own scopes_supported is retained under a dedicated field so a
+        # later resource-scope override cannot erase the trusted-tier value used to backfill a pin.
+        assert result.authorization_server_scopes == ["api://some-scope/.default"]
 
     @pytest.mark.asyncio
     async def test_fetch_single_authorization_server_metadata_derives_azure_metadata(

From feedab214ee4cdbb19b0a24200004f9e5db1f080 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 15 Jul 2026 11:12:54 -0700
Subject: [PATCH 5/6] fix(mcp): normalize blank OAuth endpoint fields to None
 at build entry points

A whitespace-only authorization_url was truthy to the row/config merges and
has_all check but blank to the corroboration gate, so discovery and
carry-forward adopted token_url/registration_url/scopes as if unpinned while
the broken whitespace value was still used for redirects. Rather than add
another strip() at each site, the pinned authorization_url/token_url/
registration_url are normalized once per build path (DB and config) via
_blank_to_none, so the merge, has_all gate, discovery gate, persist hook, and
carry-forward all see a single notion of blank. Empty and whitespace pins now
behave identically to an omitted field.
---
 .../mcp_server/mcp_server_manager.py          | 51 +++++++++++++------
 .../mcp_server/test_mcp_server_manager.py     | 46 ++++++++++++++---
 2 files changed, 75 insertions(+), 22 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index 2ace0d9b16a..e011c8f8e88 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -186,6 +186,21 @@ _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES: tuple[MCPAuth, ...] = (
 )
 
 
+def _blank_to_none(value: str | None) -> str | None:
+    """Collapse an absent, empty, or whitespace-only string to ``None``.
+
+    OAuth endpoint fields are consumed by truthiness-based merges (``row or discovered``) and by the
+    corroboration gate. A whitespace-only value is truthy to ``or`` but is not a usable endpoint, so
+    without this the merge would keep the blank value for redirects while the gate treats it as
+    unpinned and backfills the other fields, yielding a broken half-discovered config. Normalizing
+    the pinned fields once, at each build entry point, gives every downstream consumer a single
+    notion of "blank" so those code paths cannot disagree.
+    """
+    if not isinstance(value, str):
+        return None
+    return value.strip() or None
+
+
 def _normalized_authorize_endpoint(url: str) -> str:
     """Compare authorize endpoints on scheme, host, and path only. The default port is elided and
     the host is lowercased so ``https://IDP.example.com:443/authorize/`` and
@@ -1120,12 +1135,15 @@ class MCPServerManager:
             )
 
             auth_type = server_config.get("auth_type", None)
+            manual_authorization_url = _blank_to_none(server_config.get("authorization_url"))
+            manual_token_url = _blank_to_none(server_config.get("token_url"))
+            manual_registration_url = _blank_to_none(server_config.get("registration_url"))
             if server_url and (
                 auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
                 or self._obo_needs_endpoint_discovery(
                     auth_type,
                     server_config.get("token_exchange_endpoint"),
-                    server_config.get("token_url"),
+                    manual_token_url,
                 )
             ):
                 mcp_oauth_metadata = await self._descovery_metadata(
@@ -1138,7 +1156,7 @@ class MCPServerManager:
             gated_oauth_metadata = (
                 _restrict_discovery_to_corroborated_authorization_server(
                     mcp_oauth_metadata,
-                    server_config.get("authorization_url"),
+                    manual_authorization_url,
                     server_name or server_id,
                     bool(server_config.get("dcr_bridge")),
                 )
@@ -1152,13 +1170,11 @@ class MCPServerManager:
             resolved_scopes = self._extract_scopes(server_config.get("scopes")) or (
                 gated_oauth_metadata.scopes if gated_oauth_metadata else None
             )
-            resolved_authorization_url = server_config.get("authorization_url") or (
+            resolved_authorization_url = manual_authorization_url or (
                 gated_oauth_metadata.authorization_url if gated_oauth_metadata else None
             )
-            resolved_token_url = server_config.get("token_url") or (
-                gated_oauth_metadata.token_url if gated_oauth_metadata else None
-            )
-            resolved_registration_url = server_config.get("registration_url") or (
+            resolved_token_url = manual_token_url or (gated_oauth_metadata.token_url if gated_oauth_metadata else None)
+            resolved_registration_url = manual_registration_url or (
                 gated_oauth_metadata.registration_url if gated_oauth_metadata else None
             )
 
@@ -1552,14 +1568,17 @@ class MCPServerManager:
 
         auth_type = cast(MCPAuthType, mcp_server.auth_type)
         server_url = mcp_server.url
-        has_all_upstream_oauth_fields = bool(mcp_server.authorization_url and mcp_server.token_url and scopes)
+        manual_authorization_url = _blank_to_none(mcp_server.authorization_url)
+        manual_token_url = _blank_to_none(mcp_server.token_url)
+        manual_registration_url = _blank_to_none(mcp_server.registration_url)
+        has_all_upstream_oauth_fields = bool(manual_authorization_url and manual_token_url and scopes)
         needs_discovery = bool(server_url) and (
             (auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES and not has_all_upstream_oauth_fields)
             or self._obo_needs_endpoint_discovery(
                 auth_type,
                 mcp_server.token_exchange_endpoint
                 or (credentials_dict.get("token_exchange_endpoint") if credentials_dict else None),
-                mcp_server.token_url,
+                manual_token_url,
             )
         )
         mcp_oauth_metadata = (
@@ -1580,7 +1599,7 @@ class MCPServerManager:
         gated_oauth_metadata = (
             _restrict_discovery_to_corroborated_authorization_server(
                 mcp_oauth_metadata,
-                mcp_server.authorization_url,
+                manual_authorization_url,
                 mcp_server.server_id,
                 bool(getattr(mcp_server, "dcr_bridge", None)),
             )
@@ -1608,9 +1627,9 @@ class MCPServerManager:
             client_secret=client_secret_value or getattr(mcp_server, "client_secret", None),
             oauth2_flow=self._explicit_oauth2_flow(getattr(mcp_server, "oauth2_flow", None)),
             scopes=resolved_scopes,
-            authorization_url=mcp_server.authorization_url or getattr(gated_oauth_metadata, "authorization_url", None),
-            token_url=mcp_server.token_url or getattr(gated_oauth_metadata, "token_url", None),
-            registration_url=mcp_server.registration_url or getattr(gated_oauth_metadata, "registration_url", None),
+            authorization_url=manual_authorization_url or getattr(gated_oauth_metadata, "authorization_url", None),
+            token_url=manual_token_url or getattr(gated_oauth_metadata, "token_url", None),
+            registration_url=manual_registration_url or getattr(gated_oauth_metadata, "registration_url", None),
             token_endpoint_auth_method=(
                 credentials_dict.get("token_endpoint_auth_method") if credentials_dict else None
             ),
@@ -1661,14 +1680,14 @@ class MCPServerManager:
             await self._persist_discovered_obo_token_url(
                 server_id=mcp_server.server_id,
                 auth_type=auth_type,
-                existing_token_url=mcp_server.token_url,
+                existing_token_url=manual_token_url,
                 discovered_token_url=new_server.token_url,
             )
             await self._persist_discovered_oauth_endpoints(
                 server_id=mcp_server.server_id,
                 auth_type=auth_type,
-                existing_authorization_url=mcp_server.authorization_url,
-                existing_token_url=mcp_server.token_url,
+                existing_authorization_url=manual_authorization_url,
+                existing_token_url=manual_token_url,
                 existing_scopes=scopes,
                 metadata=gated_oauth_metadata,
             )
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index d14ae561f71..989281c27e9 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -409,11 +409,13 @@ class TestMCPServerManager:
         assert server.scopes == ["read"]
 
     @pytest.mark.asyncio
-    async def test_load_servers_from_config_blank_authorization_url_is_not_a_pin(self):
-        """A blank (empty-string) authorization_url is not a trust anchor, so discovery backfills
-        the whole set — authorize endpoint, token_url, and its resource-preferred scopes — from the
-        same chain, exactly as if the field had been omitted. The corroboration gate must treat
-        empty-string as unpinned so it does not strand the token_url the merge still fills."""
+    @pytest.mark.parametrize("blank_authorization_url", ["", "   "])
+    async def test_load_servers_from_config_blank_authorization_url_is_not_a_pin(self, blank_authorization_url):
+        """A blank authorization_url — empty or whitespace-only — is not a trust anchor, so discovery
+        backfills the whole set (authorize endpoint, token_url, and its resource-preferred scopes)
+        from the same chain, exactly as if the field had been omitted. The merge and the corroboration
+        gate must agree that blank means unpinned; a whitespace value that the merge kept for redirects
+        while the gate treated as unpinned would strand a broken half-discovered config."""
         manager = MCPServerManager()
 
         metadata = MCPOAuthMetadata(
@@ -424,7 +426,7 @@ class TestMCPServerManager:
         )
         config = self._oauth2_config(
             oauth2_flow="authorization_code",
-            authorization_url="",
+            authorization_url=blank_authorization_url,
             token_url=None,
         )
         with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
@@ -1165,6 +1167,38 @@ class TestMCPServerManager:
         assert built.token_url == "https://idp.example.com/token"
         assert built.scopes == ["read", "write"]
 
+    @pytest.mark.asyncio
+    async def test_build_from_table_whitespace_authorization_url_is_not_a_pin(self):
+        """A whitespace-only authorization_url on the row must not be kept for redirects while the
+        gate treats it as unpinned. It is normalized to unpinned everywhere, so the built server
+        takes the discovered authorize endpoint, token_url, and scopes as one consistent group
+        rather than serving the whitespace value with half-discovered fields."""
+        manager = MCPServerManager()
+        row = LiteLLM_MCPServerTable(
+            server_id="whitespace-auth-url",
+            alias="whitespace_auth_url",
+            description="whitespace authorization_url is not a pin",
+            url="https://up.example.com/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.oauth2,
+            authorization_url="   ",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+        )
+
+        metadata = MCPOAuthMetadata(
+            authorization_url="https://idp.example.com/authorize",
+            token_url="https://idp.example.com/token",
+            scopes=["read"],
+            authorization_server_scopes=["read"],
+        )
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
+            built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
+
+        assert built.authorization_url == "https://idp.example.com/authorize"
+        assert built.token_url == "https://idp.example.com/token"
+        assert built.scopes == ["read"]
+
     @pytest.mark.asyncio
     async def test_build_from_table_fills_endpoints_when_metadata_corroborates_manual_authorization_url(self):
         """A discovered token_url is only trusted next to a manual authorization_url when the same

From e4a6516b4916eaae239ba9b147ae71b28b7a0022 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 15 Jul 2026 16:27:15 -0700
Subject: [PATCH 6/6] fix(mcp): keep scope selection resource-driven, not
 authorization-server-driven
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Reverts the over-correction that restricted a pinned-authorization_url server's
discovered scopes to the authorization server's own scopes_supported. Per the MCP
authorization spec Scope Selection Strategy and RFC 9700 §2.3, the scopes a client
requests are resource-driven: the WWW-Authenticate 401 challenge scope, else the
RFC 9728 protected-resource scopes_supported. The authorization server's RFC 8414
scopes_supported is a non-exhaustive capability list (the server MAY omit supported
scopes) and is never the selection source; scope inflation by a compromised resource
is bounded by the authorization server and user consent (RFC 6749 §3.3), not by the
client restricting the request. The corroboration gate now rejects only the
uncorroborated token_url/registration_url (the RFC 9700 endpoint mix-up) and leaves
scopes untouched. Removes the now-unused authorization_server_scopes field.
---
 .../mcp_server/mcp_server_manager.py          | 37 ++++-----
 .../types/mcp_server/mcp_server_manager.py    | 15 ++--
 .../mcp_server/test_mcp_server_manager.py     | 77 +++++++++----------
 3 files changed, 60 insertions(+), 69 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index e011c8f8e88..8b1d00c2855 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -284,24 +284,26 @@ def _restrict_discovery_to_corroborated_authorization_server(
     server_identifier: str,
     is_dcr_bridge: bool,
 ) -> MCPOAuthMetadata | None:
-    """Bound what freshly discovered metadata may backfill into a manually pinned config.
+    """Reject discovered token/registration endpoints a manually pinned authorize endpoint cannot
+    vouch for (the RFC 9700 authorization-server mix-up).
 
-    Discovery is rooted at the MCP resource, so provenance is a property of the whole metadata
-    document, not per field: a compromised upstream can advertise both an attacker ``token_endpoint``
-    (the RFC 9700 mix-up) and inflated ``scopes`` (tricking the user into granting a broader token
-    that then flows to the upstream). Both are closed by one rule. When ``authorization_url`` is
-    admin-pinned, the discovered ``token_url`` and ``registration_url`` are kept only if the document
-    corroborates the pin (its ``authorization_endpoint`` matches), and scopes are taken from the
-    authorization server's own ``scopes_supported`` (``authorization_server_scopes``, trusted tier)
-    rather than the resource-advertised ``scopes`` a compromised upstream controls. A document that
-    does not corroborate backfills nothing. With no pin there is no trust anchor to protect and the
-    authorize endpoint comes from the same chain as everything else, so discovery is returned as-is.
+    Discovery is rooted at the MCP resource, so a compromised upstream can advertise an attacker
+    ``token_endpoint``: with ``authorization_url`` admin-pinned but ``token_url`` blank, the merge
+    would pair the trusted authorize endpoint with that attacker token endpoint, and the gateway would
+    post the authorization code and client secret there. So the discovered ``token_url`` and
+    ``registration_url`` are kept only if the document corroborates the pin (its
+    ``authorization_endpoint`` matches). ``scopes`` are deliberately NOT gated here: per the MCP
+    authorization spec Scope Selection Strategy and RFC 9700 §2.3, the scopes a client requests are
+    resource-driven (the WWW-Authenticate challenge or the RFC 9728 protected-resource
+    ``scopes_supported``), and scope inflation by a compromised resource is bounded by the
+    authorization server and user consent (RFC 6749 §3.3), not by the client second-guessing the
+    request. With no pin there is no trust anchor to protect, so discovery is returned as-is.
     """
     if metadata is None or not (manual_authorization_url and manual_authorization_url.strip()):
         return metadata
     if _endpoints_corroborate_authorization_url(metadata.authorization_url, manual_authorization_url):
-        return metadata.model_copy(update={"scopes": metadata.authorization_server_scopes})
-    if not metadata.token_url and not metadata.registration_url and not metadata.scopes:
+        return metadata
+    if not metadata.token_url and not metadata.registration_url:
         return metadata
     bridge_note = (
         " The discovered registration_url is rejected with it, so this dcr_bridge server stays on the"
@@ -311,15 +313,15 @@ def _restrict_discovery_to_corroborated_authorization_server(
     )
     verbose_logger.warning(
         "MCP OAuth discovery for server %s advertised authorization_endpoint %s, which does not match the "
-        "manually configured authorization_url %s; rejecting the discovered token_url/registration_url/scopes "
-        "so authorization codes, client credentials, and granted scopes only follow the configured "
-        "authorization server. Configure Token URL and Scopes manually if the mismatch is intentional.%s",
+        "manually configured authorization_url %s; rejecting the discovered token_url/registration_url so "
+        "authorization codes and client credentials only follow the configured authorization server. "
+        "Configure Token URL manually if the mismatch is intentional.%s",
         server_identifier,
         _normalized_authorize_endpoint(metadata.authorization_url) if metadata.authorization_url else "<absent>",
         _normalized_authorize_endpoint(manual_authorization_url),
         bridge_note,
     )
-    return metadata.model_copy(update={"token_url": None, "registration_url": None, "scopes": None})
+    return metadata.model_copy(update={"token_url": None, "registration_url": None})
 
 
 def invalidate_user_env_vars_cache(user_id: str, server_id: str) -> None:
@@ -3394,7 +3396,6 @@ class MCPServerManager:
                 authorization_url=data.get("authorization_endpoint"),
                 token_url=data.get("token_endpoint"),
                 registration_url=data.get("registration_endpoint"),
-                authorization_server_scopes=scopes,
             )
 
             if any(
diff --git a/litellm/types/mcp_server/mcp_server_manager.py b/litellm/types/mcp_server/mcp_server_manager.py
index 79a523f7c8b..e5c726296b2 100644
--- a/litellm/types/mcp_server/mcp_server_manager.py
+++ b/litellm/types/mcp_server/mcp_server_manager.py
@@ -17,18 +17,15 @@ MCPInfo = Dict[str, Any]
 
 class MCPOAuthMetadata(BaseModel):
     scopes: Optional[List[str]] = None
-    """Effective scopes, resource-preferred: the RFC 9728 protected-resource advertisement or the
-    WWW-Authenticate challenge when the resource supplied one, else the authorization server's
-    ``scopes_supported``. A compromised resource server can influence this, so it must not expand a
-    manually pinned ``authorization_url`` (see ``authorization_server_scopes``)."""
+    """Resource-driven scopes for the authorization request: the RFC 9728 protected-resource
+    ``scopes_supported``, or the ``scope`` from the WWW-Authenticate 401 challenge when the resource
+    supplied one, else the authorization server's ``scopes_supported``. This is the scope value a
+    client requests per the MCP authorization spec Scope Selection Strategy; scope minimization and
+    inflation control are the authorization server's and user's job at consent (RFC 6749 §3.3), not
+    the client's."""
     authorization_url: Optional[str] = None
     token_url: Optional[str] = None
     registration_url: Optional[str] = None
-    authorization_server_scopes: Optional[List[str]] = None
-    """The ``scopes_supported`` enumerated by the authorization-server metadata document itself
-    (RFC 8414), independent of anything the resource server advertised. This is the only scope
-    source trusted to backfill a manually pinned ``authorization_url``, because it shares provenance
-    with the ``authorization_endpoint`` used to corroborate that pin."""
     from_origin_fallback: bool = False
     """True when the metadata came from guessing the resource origin as its authorization
     server rather than from an RFC 9728/8414-advertised document. Guessed endpoints are
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index 989281c27e9..bba00ed1819 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -357,18 +357,18 @@ class TestMCPServerManager:
         assert server.needs_user_oauth_token is True
 
     @pytest.mark.asyncio
-    async def test_load_servers_from_config_rejects_uncorroborated_discovery_including_scopes(self):
-        """The config loader always runs discovery and or-merges per field, so a yaml server with a
-        manual authorization_url has the same config-time mix-up exposure as a DB row: a document
-        advertising a different authorize endpoint backfills nothing, neither its token_url nor its
-        scopes."""
+    async def test_load_servers_from_config_rejects_uncorroborated_endpoints_but_keeps_resource_scopes(self):
+        """A yaml server with a manual authorization_url has the same config-time mix-up exposure as a
+        DB row: a document advertising a different authorize endpoint has its token_url rejected. The
+        resource-driven scopes are kept, because scope selection is resource-driven (MCP Scope
+        Selection Strategy) and scope inflation is bounded by the authorization server at consent, not
+        by dropping scopes when an endpoint mismatches."""
         manager = MCPServerManager()
 
         metadata = MCPOAuthMetadata(
             authorization_url="https://attacker.example.com/authorize",
             token_url="https://attacker.example.com/token",
             scopes=["read", "admin"],
-            authorization_server_scopes=["read", "admin"],
         )
         config = self._oauth2_config(
             oauth2_flow="authorization_code",
@@ -381,20 +381,20 @@ class TestMCPServerManager:
         server = next(iter(manager.config_mcp_servers.values()))
         assert server.authorization_url == "https://idp.example.com/authorize"
         assert server.token_url is None
-        assert server.scopes is None
+        assert server.scopes == ["read", "admin"]
 
     @pytest.mark.asyncio
     async def test_load_servers_from_config_fills_token_url_when_metadata_corroborates_manual_authorization_url(self):
-        """Corroborated metadata keeps the self-heal on the config path: when the discovered
-        document advertises the same authorize endpoint the admin pinned, its token_url fills the
-        blank field and scopes come from the authorization server's own scopes_supported."""
+        """Corroborated metadata keeps the self-heal on the config path: when the discovered document
+        advertises the same authorize endpoint the admin pinned, its token_url fills the blank field
+        and scopes come through resource-driven (the discovered document's resource-preferred scopes),
+        not the authorization server's own capability list."""
         manager = MCPServerManager()
 
         metadata = MCPOAuthMetadata(
             authorization_url="https://idp.example.com/authorize",
             token_url="https://idp.example.com/token",
             scopes=["read", "admin"],
-            authorization_server_scopes=["read"],
         )
         config = self._oauth2_config(
             oauth2_flow="authorization_code",
@@ -406,7 +406,7 @@ class TestMCPServerManager:
 
         server = next(iter(manager.config_mcp_servers.values()))
         assert server.token_url == "https://idp.example.com/token"
-        assert server.scopes == ["read"]
+        assert server.scopes == ["read", "admin"]
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("blank_authorization_url", ["", "   "])
@@ -422,7 +422,6 @@ class TestMCPServerManager:
             authorization_url="https://idp.example.com/authorize",
             token_url="https://idp.example.com/token",
             scopes=["read"],
-            authorization_server_scopes=["read"],
         )
         config = self._oauth2_config(
             oauth2_flow="authorization_code",
@@ -1134,12 +1133,13 @@ class TestMCPServerManager:
         assert built.token_url == "https://idp.example.com/token"
 
     @pytest.mark.asyncio
-    async def test_build_from_table_backfills_scopes_from_authorization_server_not_resource(self):
-        """When authorization_url is admin-pinned, scopes backfill from the authorization server's
-        own scopes_supported (trusted tier), never from the resource-advertised scopes a compromised
-        upstream controls. Here the corroborating document carries an inflated resource `scopes`
-        (`admin`) alongside the real authorization_server_scopes; only the latter may be requested,
-        otherwise a hostile resource could trick the user into granting a broader token."""
+    async def test_build_from_table_backfills_resource_driven_scopes_for_pinned_authorization_url(self):
+        """When authorization_url is admin-pinned and corroborated, scopes backfill as the
+        resource-driven value (the WWW-Authenticate challenge scope, else the RFC 9728
+        protected-resource scopes_supported), per the MCP authorization spec Scope Selection Strategy.
+        The client does not restrict scopes to the authorization server's own scopes_supported; scope
+        minimization and inflation control are the authorization server's and user's job at consent
+        (RFC 6749 §3.3)."""
         manager = MCPServerManager()
         row = LiteLLM_MCPServerTable(
             server_id="manual-auth-url-1",
@@ -1157,7 +1157,6 @@ class TestMCPServerManager:
             authorization_url="https://idp.example.com/authorize",
             token_url="https://idp.example.com/token",
             scopes=["read", "admin"],
-            authorization_server_scopes=["read", "write"],
         )
         with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)) as mock_discovery:
             built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
@@ -1165,7 +1164,7 @@ class TestMCPServerManager:
         mock_discovery.assert_awaited_once()
         assert built.authorization_url == "https://idp.example.com/authorize"
         assert built.token_url == "https://idp.example.com/token"
-        assert built.scopes == ["read", "write"]
+        assert built.scopes == ["read", "admin"]
 
     @pytest.mark.asyncio
     async def test_build_from_table_whitespace_authorization_url_is_not_a_pin(self):
@@ -1190,7 +1189,6 @@ class TestMCPServerManager:
             authorization_url="https://idp.example.com/authorize",
             token_url="https://idp.example.com/token",
             scopes=["read"],
-            authorization_server_scopes=["read"],
         )
         with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
             built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
@@ -1223,7 +1221,6 @@ class TestMCPServerManager:
             token_url="https://idp.example.com/token",
             registration_url="https://idp.example.com/register",
             scopes=["read"],
-            authorization_server_scopes=["read"],
         )
         with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
             built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
@@ -1238,15 +1235,17 @@ class TestMCPServerManager:
         "advertised_authorization_url",
         ["https://attacker.example.com/authorize", None],
     )
-    async def test_build_from_table_rejects_uncorroborated_discovery_including_scopes(
+    async def test_build_from_table_rejects_uncorroborated_endpoints_but_keeps_resource_scopes(
         self, advertised_authorization_url
     ):
         """Resource-rooted discovery lets a compromised upstream advertise its own authorization
-        server. With a manual authorization_url pinned, a document that does not corroborate it
-        backfills nothing: accepting its token_url would send the code, client secret, and PKCE
-        verifier to the attacker (config-time RFC 9700 mix-up), and accepting its scopes would let
-        the upstream inflate the granted token. Both the in-memory merge and the persisted metadata
-        must drop the uncorroborated token_url, registration_url, and scopes."""
+        server. With a manual authorization_url pinned, a document that does not corroborate it has
+        its token_url and registration_url dropped: accepting them would send the code, client secret,
+        and PKCE verifier to the attacker (config-time RFC 9700 mix-up). The resource-driven scopes
+        are kept, because scope selection is resource-driven (MCP Scope Selection Strategy) and scope
+        inflation is bounded by the authorization server at consent (RFC 6749 §3.3), not by dropping
+        scopes on an endpoint mismatch. Both the in-memory merge and the persisted metadata drop only
+        the uncorroborated endpoints."""
         manager = MCPServerManager()
         row = LiteLLM_MCPServerTable(
             server_id="manual-auth-url-3",
@@ -1265,7 +1264,6 @@ class TestMCPServerManager:
             token_url="https://attacker.example.com/token",
             registration_url="https://attacker.example.com/register",
             scopes=["read", "admin"],
-            authorization_server_scopes=["read", "admin"],
         )
         with (
             patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)),
@@ -1276,11 +1274,11 @@ class TestMCPServerManager:
         assert built.authorization_url == "https://idp.example.com/authorize"
         assert built.token_url is None
         assert built.registration_url is None
-        assert built.scopes is None
+        assert built.scopes == ["read", "admin"]
         persisted_metadata = mock_persist.await_args.kwargs["metadata"]
         assert persisted_metadata.token_url is None
         assert persisted_metadata.registration_url is None
-        assert persisted_metadata.scopes is None
+        assert persisted_metadata.scopes == ["read", "admin"]
 
     @pytest.mark.asyncio
     async def test_build_from_table_skips_discovery_when_all_upstream_oauth_fields_present(self):
@@ -2384,11 +2382,11 @@ class TestMCPServerManager:
         assert result.from_origin_fallback is False
 
     @pytest.mark.asyncio
-    async def test_descovery_metadata_preserves_authorization_server_scopes_under_resource_override(self):
-        """The effective `scopes` field is resource-preferred (RFC 9728 / WWW-Authenticate), but the
-        authorization server's own scopes_supported must survive on `authorization_server_scopes` so
-        the pinned-config backfill can request the trusted-tier scopes instead of resource-advertised
-        ones. This is the provenance split the scope-inflation defense depends on."""
+    async def test_descovery_metadata_scopes_are_resource_driven(self):
+        """The effective `scopes` are resource-driven: the RFC 9728 protected-resource advertisement
+        (or WWW-Authenticate challenge) overrides the authorization server's own scopes_supported. This
+        is the MCP Scope Selection Strategy: the client requests what the resource needs, not the AS's
+        full capability list."""
         manager = MCPServerManager()
 
         mock_response = MagicMock()
@@ -2400,7 +2398,6 @@ class TestMCPServerManager:
             scopes=["as.read", "as.write"],
             authorization_url="https://idp.example.com/authorize",
             token_url="https://idp.example.com/token",
-            authorization_server_scopes=["as.read", "as.write"],
         )
 
         with (
@@ -2423,7 +2420,6 @@ class TestMCPServerManager:
 
         assert result is not None
         assert result.scopes == ["resource.only"]
-        assert result.authorization_server_scopes == ["as.read", "as.write"]
 
     @pytest.mark.asyncio
     async def test_fetch_single_authorization_server_metadata_supports_azure_issuer_path(
@@ -2465,9 +2461,6 @@ class TestMCPServerManager:
         assert result.authorization_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/authorize"
         assert result.token_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/token"
         assert result.scopes == ["api://some-scope/.default"]
-        # The authorization server's own scopes_supported is retained under a dedicated field so a
-        # later resource-scope override cannot erase the trusted-tier value used to backfill a pin.
-        assert result.authorization_server_scopes == ["api://some-scope/.default"]
 
     @pytest.mark.asyncio
     async def test_fetch_single_authorization_server_metadata_derives_azure_metadata(