fix(auth): preserve configured OAuth scopes

Author: daleselaji-dev

src/mcp/client/auth/oauth2.py

@@ -687,12 +687,15 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         self.context.client_info = None
                         self.context.clear_tokens()
 
-                    # Step 3: Apply scope selection strategy
+                    # Step 3: Apply scope selection strategy, preserving an explicitly
+                    # configured scope. An explicit empty string intentionally requests no
+                    # scopes, so only ``None`` opts into discovery-based selection.
                     self.context.client_metadata.scope = get_client_metadata_scopes(
                         extract_scope_from_www_auth(response),
                         self.context.protected_resource_metadata,
                         self.context.oauth_metadata,
                         self.context.client_metadata.grant_types,
+                        configured_scope=self.context.client_metadata.scope,
                     )
 
                     # Step 4: Register client or use URL-based client ID (CIMD)

src/mcp/client/auth/utils.py

@@ -102,21 +102,27 @@ def get_client_metadata_scopes(
     protected_resource_metadata: ProtectedResourceMetadata | None,
     authorization_server_metadata: OAuthMetadata | None = None,
     client_grant_types: list[str] | None = None,
+    configured_scope: str | None = None,
 ) -> str | None:
-    """Select effective scopes and augment for refresh token support."""
-    selected_scope: str | None = None
+    """Select effective scopes and augment for refresh token support.
+
+    An explicitly configured scope is preferred over discovered scopes. ``None``
+    continues to opt into the discovery-based selection strategy.
+    """
+    selected_scope: str | None = configured_scope
 
     # MCP spec scope selection priority:
     #   1. WWW-Authenticate header scope
     #   2. PRM scopes_supported
     #   3. AS scopes_supported (SDK fallback)
     #   4. Omit scope parameter
-    if www_authenticate_scope is not None:
-        selected_scope = www_authenticate_scope
-    elif protected_resource_metadata is not None and protected_resource_metadata.scopes_supported is not None:
-        selected_scope = " ".join(protected_resource_metadata.scopes_supported)
-    elif authorization_server_metadata is not None and authorization_server_metadata.scopes_supported is not None:
-        selected_scope = " ".join(authorization_server_metadata.scopes_supported)
+    if selected_scope is None:
+        if www_authenticate_scope is not None:
+            selected_scope = www_authenticate_scope
+        elif protected_resource_metadata is not None and protected_resource_metadata.scopes_supported is not None:
+            selected_scope = " ".join(protected_resource_metadata.scopes_supported)
+        elif authorization_server_metadata is not None and authorization_server_metadata.scopes_supported is not None:
+            selected_scope = " ".join(authorization_server_metadata.scopes_supported)
 
     # SEP-2207: append offline_access when the AS supports it and the client can use refresh tokens
     if (
@@ -126,6 +132,7 @@ def get_client_metadata_scopes(
         and "offline_access" in authorization_server_metadata.scopes_supported
         and client_grant_types is not None
         and "refresh_token" in client_grant_types
+        and selected_scope
         and "offline_access" not in selected_scope.split()
     ):
         selected_scope = f"{selected_scope} offline_access"

tests/client/test_auth.py

@@ -268,6 +268,55 @@ def test_clear_tokens(self, oauth_provider: OAuthClientProvider, valid_tokens: O
 class TestOAuthFlow:
     """Test OAuth flow methods."""
 
+    @pytest.mark.anyio
+    async def test_explicit_client_scope_is_not_overwritten_by_discovery(self, oauth_provider: OAuthClientProvider):
+        oauth_provider.context.client_metadata.scope = "explicit:read"
+        oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="existing-client",
+            client_secret="client-secret",
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        )
+        oauth_provider._initialized = True
+        oauth_provider._perform_authorization = mock.AsyncMock(
+            return_value=httpx2.Request("POST", "https://auth.example.com/token")
+        )
+
+        auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+        request = await auth_flow.__anext__()
+        prm_request = await auth_flow.asend(
+            httpx2.Response(
+                401,
+                headers={"WWW-Authenticate": 'Bearer resource_metadata="https://api.example.com/.well-known/prm"'},
+                request=request,
+            )
+        )
+        prm_response = httpx2.Response(
+            200,
+            json={
+                "resource": "https://api.example.com/v1/mcp",
+                "authorization_servers": ["https://auth.example.com"],
+                "scopes_supported": ["discovered:read", "discovered:write"],
+            },
+            request=prm_request,
+        )
+        asm_request = await auth_flow.asend(prm_response)
+        asm_response = httpx2.Response(
+            200,
+            json={
+                "issuer": "https://auth.example.com",
+                "authorization_endpoint": "https://auth.example.com/authorize",
+                "token_endpoint": "https://auth.example.com/token",
+                "scopes_supported": ["discovered:read", "discovered:write"],
+            },
+            request=asm_request,
+        )
+
+        token_request = await auth_flow.asend(asm_response)
+
+        assert oauth_provider.context.client_metadata.scope == "explicit:read"
+        assert token_request.method == "POST"
+        await auth_flow.aclose()
+
     @pytest.mark.anyio
     async def test_build_protected_resource_discovery_urls(
         self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage