Enhance OAuth metadata discovery in client

- Added a new method to discover OAuth Protected Resource Metadata, improving the handling of authorization server URLs.
- Updated the OAuthClientProvider to utilize the discovered protected resource metadata when fetching OAuth metadata.
- Refactored tests to validate the new discovery logic and ensure correct URL calls for protected resource and authorization server metadata.

Signed-off-by: Xin Fu <[email protected]>

Author: imfing

src/mcp/client/auth.py

@@ -12,7 +12,7 @@
 import time
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import Protocol
-from urllib.parse import urlencode, urljoin
+from urllib.parse import urlencode, urljoin, urlparse, urlunparse
 
 import anyio
 import httpx
@@ -21,6 +21,7 @@
     OAuthClientInformationFull,
     OAuthClientMetadata,
     OAuthMetadata,
+    OAuthProtectedResourceMetadata,
     OAuthToken,
 )
 from mcp.types import LATEST_PROTOCOL_VERSION
@@ -116,19 +117,59 @@ def _get_authorization_base_url(self, server_url: str) -> str:
 
         Per MCP spec 2.3.2: https://api.example.com/v1/mcp -> https://api.example.com
         """
-        from urllib.parse import urlparse, urlunparse
-
         parsed = urlparse(server_url)
-        # Remove path component
         return urlunparse((parsed.scheme, parsed.netloc, "", "", "", ""))
 
+    async def _discover_protected_resource_metadata(
+        self, resource_server_url: str
+    ) -> OAuthProtectedResourceMetadata | None:
+        """
+        Looks up RFC 9728 OAuth 2.0 Protected Resource Metadata.
+
+        If the server returns a 404 for the well-known endpoint, returns None.
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                urljoin(resource_server_url, "/.well-known/oauth-protected-resource")
+            )
+            if response.status_code == 404:
+                return None
+            response.raise_for_status()
+            metadata_json = response.json()
+            logger.debug(
+                f"OAuth protected resource metadata discovered: {metadata_json}"
+            )
+            return OAuthProtectedResourceMetadata.model_validate(metadata_json)
+
     async def _discover_oauth_metadata(self, server_url: str) -> OAuthMetadata | None:
         """
         Discover OAuth metadata from server's well-known endpoint.
+
+        First tries to discover protected resource metadata and use its authorization
+        server URL if available, otherwise falls back to the server's own well-known.
         """
-        # Extract base URL per MCP spec
-        auth_base_url = self._get_authorization_base_url(server_url)
-        url = urljoin(auth_base_url, "/.well-known/oauth-authorization-server")
+        auth_server_url = self._get_authorization_base_url(server_url)
+
+        try:
+            protected_resource_metadata = (
+                await self._discover_protected_resource_metadata(server_url)
+            )
+
+            if (
+                protected_resource_metadata
+                and protected_resource_metadata.authorization_servers
+                and len(protected_resource_metadata.authorization_servers) > 0
+            ):
+                auth_server_url = str(
+                    protected_resource_metadata.authorization_servers[0]
+                )
+        except Exception as e:
+            logger.warning(
+                "Could not load OAuth Protected Resource metadata, "
+                f"falling back to /.well-known/oauth-authorization-server: {e}"
+            )
+
+        url = urljoin(auth_server_url, "/.well-known/oauth-authorization-server")
         headers = {"MCP-Protocol-Version": LATEST_PROTOCOL_VERSION}
 
         async with httpx.AsyncClient() as client:

src/mcp/server/auth/routes.py

@@ -172,12 +172,8 @@ def build_metadata(
     client_registration_options: ClientRegistrationOptions,
     revocation_options: RevocationOptions,
 ) -> OAuthMetadata:
-    authorization_url = AnyHttpUrl(
-        str(issuer_url).rstrip("/") + AUTHORIZATION_PATH
-    )
-    token_url = AnyHttpUrl(
-        str(issuer_url).rstrip("/") + TOKEN_PATH
-    )
+    authorization_url = AnyHttpUrl(str(issuer_url).rstrip("/") + AUTHORIZATION_PATH)
+    token_url = AnyHttpUrl(str(issuer_url).rstrip("/") + TOKEN_PATH)
 
     # Create metadata
     metadata = OAuthMetadata(

tests/client/test_auth.py

@@ -20,6 +20,7 @@
     OAuthClientInformationFull,
     OAuthClientMetadata,
     OAuthMetadata,
+    OAuthProtectedResourceMetadata,
     OAuthToken,
 )
 
@@ -74,6 +75,16 @@ def oauth_metadata():
     )
 
 
+@pytest.fixture
+def oauth_protected_resource_metadata():
+    return OAuthProtectedResourceMetadata(
+        resource="https://api.example.com/v1/mcp",
+        authorization_servers=[AnyHttpUrl("https://auth.example.com")],
+        scopes_supported=["read", "write"],
+        bearer_methods_supported=["header"],
+    )
+
+
 @pytest.fixture
 def oauth_client_info():
     return OAuthClientInformationFull(
@@ -210,10 +221,13 @@ async def test_discover_oauth_metadata_success(
             assert result.token_endpoint == oauth_metadata.token_endpoint
 
             # Verify correct URL was called
-            mock_client.get.assert_called_once()
-            call_args = mock_client.get.call_args[0]
+            assert mock_client.get.call_count == 2
             assert (
-                call_args[0]
+                mock_client.get.call_args_list[0][0][0]
+                == "https://api.example.com/.well-known/oauth-protected-resource"
+            )
+            assert (
+                mock_client.get.call_args_list[1][0][0]
                 == "https://api.example.com/.well-known/oauth-authorization-server"
             )
 
@@ -262,6 +276,62 @@ async def test_discover_oauth_metadata_cors_fallback(
             assert result is not None
             assert mock_client.get.call_count == 2
 
+    @pytest.mark.anyio
+    async def test_discover_oauth_metadata_from_protected_resource(
+        self, oauth_provider, oauth_metadata, oauth_protected_resource_metadata
+    ):
+        """Test OAuth metadata discovery using protected resource metadata."""
+        protected_resource_response = oauth_protected_resource_metadata.model_dump(
+            by_alias=True, mode="json"
+        )
+        oauth_metadata_response = oauth_metadata.model_dump(by_alias=True, mode="json")
+
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client_class.return_value.__aenter__.return_value = mock_client
+
+            # First call returns protected resource metadata
+            protected_resource_mock = Mock()
+            protected_resource_mock.status_code = 200
+            protected_resource_mock.json.return_value = protected_resource_response
+
+            # Second call returns OAuth metadata from authorization server
+            oauth_metadata_mock = Mock()
+            oauth_metadata_mock.status_code = 200
+            oauth_metadata_mock.json.return_value = oauth_metadata_response
+
+            mock_client.get.side_effect = [
+                protected_resource_mock,  # Protected resource metadata call
+                oauth_metadata_mock,  # OAuth metadata from auth server call
+            ]
+
+            result = await oauth_provider._discover_oauth_metadata(
+                "https://api.example.com/v1/mcp"
+            )
+
+            assert result is not None
+            assert (
+                result.authorization_endpoint == oauth_metadata.authorization_endpoint
+            )
+            assert result.token_endpoint == oauth_metadata.token_endpoint
+
+            # Verify correct URLs were called in order
+            assert mock_client.get.call_count == 2
+
+            # First call should be to protected resource metadata endpoint
+            first_call_args = mock_client.get.call_args_list[0][0]
+            assert (
+                first_call_args[0]
+                == "https://api.example.com/.well-known/oauth-protected-resource"
+            )
+
+            # Second call should be to authorization server's OAuth metadata endpoint
+            second_call_args = mock_client.get.call_args_list[1][0]
+            assert (
+                second_call_args[0]
+                == "https://auth.example.com/.well-known/oauth-authorization-server"
+            )
+
     @pytest.mark.anyio
     async def test_register_oauth_client_success(
         self, oauth_provider, oauth_metadata, oauth_client_info