[v1.x] Preserve the exact OAuth resource indicator (backport #2925)

src/mcp/client/auth/oauth2.py

@@ -155,7 +155,7 @@ def get_resource_url(self) -> str:
 
         # If PRM provides a resource that's a valid parent, use it
         if self.protected_resource_metadata and self.protected_resource_metadata.resource:
-            prm_resource = str(self.protected_resource_metadata.resource)
+            prm_resource = self.protected_resource_metadata.resource_str
             if check_resource_allowed(requested_resource=resource, configured_resource=prm_resource):
                 resource = prm_resource
 
@@ -280,7 +280,7 @@ def __init__(
 
     async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None:
         """Validate that PRM resource matches the server URL per RFC 8707."""
-        prm_resource = str(prm.resource) if prm.resource else None
+        prm_resource = prm.resource_str if prm.resource else None
         if not prm_resource:
             return  # pragma: no cover
         default_resource = resource_url_from_server_url(self.context.server_url)

src/mcp/shared/auth.py

@@ -1,6 +1,23 @@
-from typing import Any, Literal
-
-from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, field_validator
+from typing import Any, Literal, cast
+
+from pydantic import (
+    AnyHttpUrl,
+    AnyUrl,
+    BaseModel,
+    ConfigDict,
+    Field,
+    PrivateAttr,
+    ValidatorFunctionWrapHandler,
+    field_validator,
+    model_validator,
+)
+
+# `url_preserve_empty_path` (pydantic >= 2.12) keeps a path-less URL from gaining a trailing slash when
+# parsed from the wire, so issuer/resource identifiers round-trip as transmitted (RFC 3986 §6.2.1 simple
+# string comparison). Older pydantic ignores unknown config keys at runtime; the cast keeps the 2.11 type
+# stubs (which do not know the key) quiet. See `ProtectedResourceMetadata.resource_str` for the
+# version-independent path used for the RFC 8707 `resource` parameter.
+_PRESERVE_EMPTY_PATH = cast(ConfigDict, {"url_preserve_empty_path": True})
 
 
 class OAuthToken(BaseModel):
@@ -41,6 +58,8 @@ class OAuthClientMetadata(BaseModel):
     for the full specification.
     """
 
+    model_config = _PRESERVE_EMPTY_PATH
+
     redirect_uris: list[AnyUrl] | None = Field(..., min_length=1)
     # supported auth methods for the token endpoint
     token_endpoint_auth_method: (
@@ -132,6 +151,8 @@ class OAuthMetadata(BaseModel):
     See https://datatracker.ietf.org/doc/html/rfc8414#section-2
     """
 
+    model_config = _PRESERVE_EMPTY_PATH
+
     issuer: AnyHttpUrl
     authorization_endpoint: AnyHttpUrl
     token_endpoint: AnyHttpUrl
@@ -162,8 +183,31 @@ class ProtectedResourceMetadata(BaseModel):
     See https://datatracker.ietf.org/doc/html/rfc9728#section-2
     """
 
+    model_config = _PRESERVE_EMPTY_PATH
+
     resource: AnyHttpUrl
     authorization_servers: list[AnyHttpUrl] = Field(..., min_length=1)
+    # The `resource` value exactly as received. `url_preserve_empty_path` only takes effect on
+    # pydantic >= 2.12; on older pydantic a path-less URL still renders with a trailing slash, which
+    # breaks the byte-exact RFC 8707 `resource` parameter. Kept alongside the parsed URL so callers
+    # can echo the server's identifier verbatim regardless of pydantic version.
+    _resource_raw: str | None = PrivateAttr(default=None)
+
+    @model_validator(mode="wrap")
+    @classmethod
+    def _capture_raw_resource(cls, data: Any, handler: ValidatorFunctionWrapHandler) -> "ProtectedResourceMetadata":
+        raw: Any = cast(dict[str, Any], data).get("resource") if isinstance(data, dict) else None
+        model = cast("ProtectedResourceMetadata", handler(data))
+        if isinstance(raw, str):
+            model._resource_raw = raw
+        return model
+
+    @property
+    def resource_str(self) -> str:
+        """The resource identifier as a string, exactly as the server published it when parsed from
+        JSON/dict input (RFC 8707 requires clients to send it byte-for-byte); otherwise the rendered URL."""
+        return self._resource_raw if self._resource_raw is not None else str(self.resource)
+
     jwks_uri: AnyHttpUrl | None = None
     scopes_supported: list[str] | None = None
     bearer_methods_supported: list[str] | None = Field(default=["header"])  # MCP only supports header method

tests/client/test_auth.py

@@ -522,10 +522,11 @@ async def test_handle_metadata_response_success(self, oauth_provider: OAuthClien
         }"""
         response = httpx.Response(200, content=content)
 
-        # Should set metadata
+        # Should set metadata. On pydantic >= 2.12 the empty path is preserved (no trailing slash);
+        # older pydantic ignores `url_preserve_empty_path` and still normalises to ".../".
         await oauth_provider._handle_oauth_metadata_response(response)
         assert oauth_provider.context.oauth_metadata is not None
-        assert str(oauth_provider.context.oauth_metadata.issuer) == "https://auth.example.com/"
+        assert str(oauth_provider.context.oauth_metadata.issuer).rstrip("/") == "https://auth.example.com"
 
     @pytest.mark.anyio
     async def test_prioritize_www_auth_scope_over_prm(
@@ -2212,6 +2213,28 @@ async def test_get_resource_url_falls_back_when_prm_mismatches(
     assert provider.context.get_resource_url() == "https://api.example.com/v1/mcp"
 
 
+@pytest.mark.anyio
+async def test_get_resource_url_echoes_pathless_prm_resource_verbatim(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+) -> None:
+    """RFC 8707: the `resource` parameter is the PRM `resource` byte-for-byte. A path-less identifier
+    parsed from the wire must not gain a trailing slash, on any supported pydantic version."""
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+    )
+    provider._initialized = True
+
+    prm = ProtectedResourceMetadata.model_validate_json(
+        '{"resource": "https://api.example.com", "authorization_servers": ["https://auth.example.com"]}'
+    )
+    assert prm.resource_str == "https://api.example.com"
+    provider.context.protected_resource_metadata = prm
+
+    assert provider.context.get_resource_url() == "https://api.example.com"
+
+
 def _prepare_full_flow(provider: OAuthClientProvider, client_info: OAuthClientInformationFull | None) -> list[str]:
     """Reset `provider` for a full flow with `client_info` as the stored registration, and wire a
     redirect/callback pair that echoes the `state` of the last authorization URL it was sent to.