Change to the updated MCP spec where priority order is 1. Client scope 2. WWW-auth scope 3. PRM scopes

Author: dogacancolak

src/mcp/client/auth.py

@@ -97,6 +97,7 @@ class OAuthContext:
     oauth_metadata: OAuthMetadata | None = None
     auth_server_url: str | None = None
     protocol_version: str | None = None
+    www_authenticate_scope: str | None = None
 
     # Client registration
     client_info: OAuthClientInformationFull | None = None
@@ -228,6 +229,30 @@ def _extract_resource_metadata_from_www_auth(self, init_response: httpx.Response
 
         return None
 
+    def _extract_scope_from_www_auth(self, init_response: httpx.Response) -> str | None:
+        """
+        Extract scope parameter from WWW-Authenticate header as per RFC6750.
+
+        Returns:
+            Scope string if found in WWW-Authenticate header, None otherwise
+        """
+        if not init_response or init_response.status_code != 401:
+            return None
+
+        www_auth_header = init_response.headers.get("WWW-Authenticate")
+        if not www_auth_header:
+            return None
+
+        # Pattern matches: scope="value" or scope=value (unquoted)
+        pattern = r'scope=(?:"([^"]+)"|([^\s,]+))'
+        match = re.search(pattern, www_auth_header)
+
+        if match:
+            # Return quoted value if present, otherwise unquoted value
+            return match.group(1) or match.group(2)
+
+        return None
+
     async def _discover_protected_resource(self, init_response: httpx.Response) -> httpx.Request:
         # RFC9728: Try to extract resource_metadata URL from WWW-Authenticate header of the initial response
         url = self._extract_resource_metadata_from_www_auth(init_response)
@@ -480,16 +505,21 @@ async def _handle_oauth_metadata_response(self, response: httpx.Response) -> Non
         self.context.oauth_metadata = metadata
 
         # Only set scope if client_metadata.scope is None
+        # Per MCP spec, priority order:
+        # 1. Use scope from WWW-Authenticate header (if provided)
+        # 2. Use all scopes from PRM scopes_supported (if available)
+        # 3. Omit scope parameter if neither is available
         if self.context.client_metadata.scope is None:
-            # Priority 1: Use PRM's scopes_supported if available
-            if (
+            if self.context.www_authenticate_scope is not None:
+                # Priority 1: WWW-Authenticate header scope
+                self.context.client_metadata.scope = self.context.www_authenticate_scope
+            elif (
                 self.context.protected_resource_metadata is not None
                 and self.context.protected_resource_metadata.scopes_supported is not None
             ):
+                # Priority 2: PRM scopes_supported
                 self.context.client_metadata.scope = " ".join(self.context.protected_resource_metadata.scopes_supported)
-            # Priority 2: Fall back to OAuth metadata scopes if available
-            elif metadata.scopes_supported is not None:
-                self.context.client_metadata.scope = " ".join(metadata.scopes_supported)
+            # Priority 3: Omit scope parameter
 
     async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]:
         """HTTPX auth flow integration."""
@@ -518,12 +548,15 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                 # Perform full OAuth flow
                 try:
                     # OAuth flow must be inline due to generator constraints
-                    # Step 1: Discover protected resource metadata (RFC9728 with WWW-Authenticate support)
+                    # Step 1: Extract scope from WWW-Authenticate header
+                    self.context.www_authenticate_scope = self._extract_scope_from_www_auth(response)
+
+                    # Step 2: Discover protected resource metadata (RFC9728 with WWW-Authenticate support)
                     discovery_request = await self._discover_protected_resource(response)
                     discovery_response = yield discovery_request
                     await self._handle_protected_resource_response(discovery_response)
 
-                    # Step 2: Discover OAuth metadata (with fallback for legacy servers)
+                    # Step 3: Discover OAuth metadata (with fallback for legacy servers)
                     discovery_urls = self._get_discovery_urls()
                     for url in discovery_urls:
                         oauth_metadata_request = self._create_oauth_metadata_request(url)
@@ -538,16 +571,16 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                         elif oauth_metadata_response.status_code < 400 or oauth_metadata_response.status_code >= 500:
                             break  # Non-4XX error, stop trying
 
-                    # Step 3: Register client if needed
+                    # Step 4: Register client if needed
                     registration_request = await self._register_client()
                     if registration_request:
                         registration_response = yield registration_request
                         await self._handle_registration_response(registration_response)
 
-                    # Step 4: Perform authorization
+                    # Step 5: Perform authorization
                     auth_code, code_verifier = await self._perform_authorization()
 
-                    # Step 5: Exchange authorization code for tokens
+                    # Step 6: Exchange authorization code for tokens
                     token_request = await self._exchange_token(auth_code, code_verifier)
                     token_response = yield token_request
                     await self._handle_token_response(token_response)

tests/client/test_auth.py

@@ -121,21 +121,6 @@ def prm_metadata_without_scopes():
     )
 
 
-@pytest.fixture
-def oauth_metadata_response_without_scopes():
-    """OAuth metadata response without scopes."""
-    return httpx.Response(
-        200,
-        content=(
-            b'{"issuer": "https://auth.example.com", '
-            b'"authorization_endpoint": "https://auth.example.com/authorize", '
-            b'"token_endpoint": "https://auth.example.com/token", '
-            b'"registration_endpoint": "https://auth.example.com/register"}'
-            # No scopes_supported field
-        ),
-    )
-
-
 class TestPKCEParameters:
     """Test PKCE parameter generation."""
 
@@ -449,60 +434,63 @@ async def test_handle_metadata_response_success(self, oauth_provider: OAuthClien
         assert str(oauth_provider.context.oauth_metadata.issuer) == "https://auth.example.com/"
 
     @pytest.mark.anyio
-    async def test_prioritize_prm_scopes_over_oauth_metadata(
+    async def test_prioritize_www_auth_scope_over_prm(
         self,
         oauth_provider_without_scope: OAuthClientProvider,
         oauth_metadata_response: httpx.Response,
         prm_metadata: ProtectedResourceMetadata,
     ):
-        """Test that PRM scopes are prioritized over auth server metadata scopes."""
+        """Test that WWW-Authenticate scope is prioritized over PRM scopes."""
         provider = oauth_provider_without_scope
 
-        # Set up PRM metadata with specific scopes
+        # Set up PRM metadata with scopes
         provider.context.protected_resource_metadata = prm_metadata
 
+        # Set WWW-Authenticate scope (priority 1)
+        provider.context.www_authenticate_scope = "special:scope from:www-authenticate"
+
         # Process the OAuth metadata
         await provider._handle_oauth_metadata_response(oauth_metadata_response)
 
-        # Verify that PRM scopes are used (not OAuth metadata scopes)
-        assert provider.context.client_metadata.scope == "resource:read resource:write"
+        # Verify that WWW-Authenticate scope is used (not PRM scopes)
+        assert provider.context.client_metadata.scope == "special:scope from:www-authenticate"
 
     @pytest.mark.anyio
-    async def test_fallback_to_oauth_metadata_scopes_when_no_prm_scopes(
+    async def test_prioritize_prm_scopes_when_no_www_auth_scope(
         self,
         oauth_provider_without_scope: OAuthClientProvider,
         oauth_metadata_response: httpx.Response,
-        prm_metadata_without_scopes: ProtectedResourceMetadata,
+        prm_metadata: ProtectedResourceMetadata,
     ):
-        """Test fallback to OAuth metadata scopes when PRM has no scopes."""
+        """Test that PRM scopes are prioritized when WWW-Authenticate header has no scopes."""
         provider = oauth_provider_without_scope
 
-        # Set up PRM metadata without scopes
-        provider.context.protected_resource_metadata = prm_metadata_without_scopes
+        # Set up PRM metadata with specific scopes
+        provider.context.protected_resource_metadata = prm_metadata
 
-        # Process the OAuth metadata
+        # Process the OAuth metadata (no WWW-Authenticate scope)
         await provider._handle_oauth_metadata_response(oauth_metadata_response)
 
-        # Verify that OAuth metadata scopes are used as fallback
-        assert provider.context.client_metadata.scope == "read write admin"
+        # Verify that PRM scopes are used
+        assert provider.context.client_metadata.scope == "resource:read resource:write"
 
     @pytest.mark.anyio
-    async def test_no_scope_changes_when_both_missing(
+    async def test_omit_scope_when_no_prm_scopes_or_www_auth(
         self,
         oauth_provider_without_scope: OAuthClientProvider,
+        oauth_metadata_response: httpx.Response,
         prm_metadata_without_scopes: ProtectedResourceMetadata,
-        oauth_metadata_response_without_scopes: httpx.Response,
     ):
-        """Test that no scope changes occur when both PRM and OAuth metadata lack scopes."""
+        """Test that scope is omitted when PRM has no scopes and WWW-Authenticate doesn't specify scope."""
         provider = oauth_provider_without_scope
 
         # Set up PRM metadata without scopes
         provider.context.protected_resource_metadata = prm_metadata_without_scopes
 
-        # Process the OAuth metadata
-        await provider._handle_oauth_metadata_response(oauth_metadata_response_without_scopes)
+        # Process the OAuth metadata (no WWW-Authenticate scope set)
+        await provider._handle_oauth_metadata_response(oauth_metadata_response)
 
-        # Verify that scope remains None
+        # Verify that scope is omitted
         assert provider.context.client_metadata.scope is None
 
     @pytest.mark.anyio
@@ -515,6 +503,9 @@ async def test_preserve_existing_client_scope(
         """Test that existing client scope is preserved regardless of metadata."""
         provider = oauth_provider
 
+        # Set WWW-Authenticate scope
+        provider.context.www_authenticate_scope = "special:scope from:www-authenticate"
+
         # Set up PRM metadata with scopes
         provider.context.protected_resource_metadata = prm_metadata
 
@@ -1092,3 +1083,98 @@ async def callback_handler() -> tuple[str, str | None]:
 
         result = provider._extract_resource_metadata_from_www_auth(init_response)
         assert result is None, f"Should return None for {description}"
+
+    @pytest.mark.parametrize(
+        "www_auth_header,expected_scope",
+        [
+            # Quoted scope
+            ('Bearer scope="read write"', "read write"),
+            # Unquoted scope
+            ("Bearer scope=read", "read"),
+            # Multiple parameters with quoted scope
+            ('Bearer realm="api", scope="admin:write resource:read"', "admin:write resource:read"),
+            # Multiple parameters with unquoted scope
+            ('Bearer realm="api", scope=basic', "basic"),
+            # Scope with special characters (colons, underscores)
+            ('Bearer scope="resource:read resource:write user_profile"', "resource:read resource:write user_profile"),
+        ],
+    )
+    def test_extract_scope_from_www_auth_valid_cases(
+        self,
+        client_metadata: OAuthClientMetadata,
+        mock_storage: MockTokenStorage,
+        www_auth_header: str,
+        expected_scope: str,
+    ):
+        """Test extraction of scope from various valid WWW-Authenticate headers."""
+
+        async def redirect_handler(url: str) -> None:
+            pass
+
+        async def callback_handler() -> tuple[str, str | None]:
+            return "test_auth_code", "test_state"
+
+        provider = OAuthClientProvider(
+            server_url="https://api.example.com/v1/mcp",
+            client_metadata=client_metadata,
+            storage=mock_storage,
+            redirect_handler=redirect_handler,
+            callback_handler=callback_handler,
+        )
+
+        init_response = httpx.Response(
+            status_code=401,
+            headers={"WWW-Authenticate": www_auth_header},
+            request=httpx.Request("GET", "https://api.example.com/test"),
+        )
+
+        result = provider._extract_scope_from_www_auth(init_response)
+        assert result == expected_scope
+
+    @pytest.mark.parametrize(
+        "status_code,www_auth_header,description",
+        [
+            # No header
+            (401, None, "no WWW-Authenticate header"),
+            # Empty header
+            (401, "", "empty WWW-Authenticate header"),
+            # Header without scope
+            (401, 'Bearer realm="api", error="insufficient_scope"', "no scope parameter"),
+            # Malformed header
+            (401, "Bearer scope=", "malformed scope parameter"),
+            # Non-401 status code
+            (200, 'Bearer scope="read write"', "200 OK response"),
+            (500, 'Bearer scope="read write"', "500 error response"),
+        ],
+    )
+    def test_extract_scope_from_www_auth_invalid_cases(
+        self,
+        client_metadata: OAuthClientMetadata,
+        mock_storage: MockTokenStorage,
+        status_code: int,
+        www_auth_header: str | None,
+        description: str,
+    ):
+        """Test extraction returns None for invalid cases."""
+
+        async def redirect_handler(url: str) -> None:
+            pass
+
+        async def callback_handler() -> tuple[str, str | None]:
+            return "test_auth_code", "test_state"
+
+        provider = OAuthClientProvider(
+            server_url="https://api.example.com/v1/mcp",
+            client_metadata=client_metadata,
+            storage=mock_storage,
+            redirect_handler=redirect_handler,
+            callback_handler=callback_handler,
+        )
+
+        headers = {"WWW-Authenticate": www_auth_header} if www_auth_header is not None else {}
+        init_response = httpx.Response(
+            status_code=status_code, headers=headers, request=httpx.Request("GET", "https://api.example.com/test")
+        )
+
+        result = provider._extract_scope_from_www_auth(init_response)
+        assert result is None, f"Should return None for {description}"