Preserve query components in PRM resource URLs

Author: mayankbohradev

src/mcp/client/auth/utils.py

@@ -1,5 +1,5 @@
 import re
-from urllib.parse import urljoin, urlparse
+from urllib.parse import urljoin, urlparse, urlunparse
 
 from httpx import Request, Response
 from mcp_types import LATEST_PROTOCOL_VERSION
@@ -63,7 +63,7 @@ def build_protected_resource_metadata_discovery_urls(www_auth_url: str | None, s
 
     Per SEP-985, the client MUST:
     1. Try resource_metadata from WWW-Authenticate header (if present)
-    2. Fall back to path-based well-known URI: /.well-known/oauth-protected-resource/{path}
+    2. Fall back to path/query-based well-known URI: /.well-known/oauth-protected-resource/{path}?{query}
     3. Fall back to root-based well-known URI: /.well-known/oauth-protected-resource
 
     Args:
@@ -83,9 +83,11 @@ def build_protected_resource_metadata_discovery_urls(www_auth_url: str | None, s
     parsed = urlparse(server_url)
     base_url = f"{parsed.scheme}://{parsed.netloc}"
 
-    # Priority 2: Path-based well-known URI (if server has a path component)
-    if parsed.path and parsed.path != "/":
-        path_based_url = urljoin(base_url, f"/.well-known/oauth-protected-resource{parsed.path}")
+    # Priority 2: Path/query-based well-known URI (if server has a path or query component)
+    if (parsed.path and parsed.path != "/") or parsed.query:
+        resource_path = parsed.path if parsed.path != "/" else ""
+        metadata_path = f"/.well-known/oauth-protected-resource{resource_path}"
+        path_based_url = urlunparse((parsed.scheme, parsed.netloc, metadata_path, parsed.params, parsed.query, ""))
         urls.append(path_based_url)
 
     # Priority 3: Root-based well-known URI

src/mcp/server/auth/routes.py

@@ -1,6 +1,6 @@
 from collections.abc import Awaitable, Callable
 from typing import Any
-from urllib.parse import urlparse
+from urllib.parse import urlparse, urlunparse
 
 from pydantic import AnyHttpUrl
 from starlette.middleware.cors import CORSMiddleware
@@ -202,7 +202,7 @@ def build_metadata(
 def build_resource_metadata_url(resource_server_url: AnyHttpUrl) -> AnyHttpUrl:
     """Build RFC 9728 compliant protected resource metadata URL.
 
-    Inserts /.well-known/oauth-protected-resource between host and resource path
+    Inserts /.well-known/oauth-protected-resource between host and resource path/query
     as specified in RFC 9728 §3.1.
 
     Args:
@@ -214,7 +214,8 @@ def build_resource_metadata_url(resource_server_url: AnyHttpUrl) -> AnyHttpUrl:
     parsed = urlparse(str(resource_server_url))
     # Handle trailing slash: if path is just "/", treat as empty
     resource_path = parsed.path if parsed.path != "/" else ""
-    return AnyHttpUrl(f"{parsed.scheme}://{parsed.netloc}/.well-known/oauth-protected-resource{resource_path}")
+    metadata_path = f"/.well-known/oauth-protected-resource{resource_path}"
+    return AnyHttpUrl(urlunparse((parsed.scheme, parsed.netloc, metadata_path, parsed.params, parsed.query, "")))
 
 
 def create_protected_resource_routes(

src/mcp/shared/auth_utils.py

@@ -32,9 +32,10 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) ->
     """Check if a requested resource URL matches a configured resource URL.
 
     A requested resource matches if it has the same scheme, domain, port,
-    and its path starts with the configured resource's path. This allows
-    hierarchical matching where a token for a parent resource can be used
-    for child resources.
+    the same query component, and its path starts with the configured
+    resource's path. This allows hierarchical matching where a token for a
+    parent resource can be used for child resources without collapsing
+    query-routed resource identifiers.
 
     Args:
         requested_resource: The resource URL being requested
@@ -51,6 +52,9 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) ->
     if requested.scheme.lower() != configured.scheme.lower() or requested.netloc.lower() != configured.netloc.lower():
         return False
 
+    if requested.query != configured.query:
+        return False
+
     # Normalize trailing slashes before comparison so that
     # "/foo" and "/foo/" are treated as equivalent.
     requested_path = requested.path

tests/client/test_auth.py

@@ -863,6 +863,26 @@ async def test_validate_resource_rejects_mismatched_resource(
         await provider._validate_resource_match(prm)
 
 
+@pytest.mark.anyio
+async def test_validate_resource_rejects_mismatched_query_component(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+) -> None:
+    """Client rejects PRM resources with a different query-routed resource identifier."""
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp?tenant=a",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+    )
+    provider._initialized = True
+
+    prm = ProtectedResourceMetadata(
+        resource=AnyHttpUrl("https://api.example.com/v1/mcp?tenant=b"),
+        authorization_servers=[AnyHttpUrl("https://auth.example.com")],
+    )
+    with pytest.raises(OAuthFlowError, match="does not match expected"):
+        await provider._validate_resource_match(prm)
+
+
 @pytest.mark.anyio
 async def test_validate_resource_accepts_matching_resource(
     client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
@@ -1827,6 +1847,23 @@ async def callback_handler() -> AuthorizationCodeResult:
         assert discovery_urls[0] == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
         assert discovery_urls[1] == "https://api.example.com/.well-known/oauth-protected-resource"
 
+    def test_prm_discovery_preserves_server_url_query_component(self):
+        """Fallback PRM discovery preserves RFC 9728 query components."""
+        init_response = httpx.Response(
+            status_code=401, headers={}, request=httpx.Request("GET", "https://api.example.com/v1/mcp?tenant=a")
+        )
+
+        discovery_urls = build_protected_resource_metadata_discovery_urls(
+            extract_resource_metadata_from_www_auth(init_response), "https://api.example.com/v1/mcp?tenant=a"
+        )
+
+        assert discovery_urls == snapshot(
+            [
+                "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp?tenant=a",
+                "https://api.example.com/.well-known/oauth-protected-resource",
+            ]
+        )
+
     @pytest.mark.anyio
     async def test_root_based_fallback_after_path_based_404(
         self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage

tests/server/auth/test_protected_resource.py

@@ -122,6 +122,13 @@ def test_metadata_url_construction_url_with_path_component():
     assert str(result) == "https://example.com/.well-known/oauth-protected-resource/mcp"
 
 
+def test_metadata_url_construction_preserves_query_component():
+    """Resource metadata URL construction preserves RFC 9728 query components."""
+    resource_url = AnyHttpUrl("https://example.com/mcp?tenant=a")
+    result = build_resource_metadata_url(resource_url)
+    assert str(result) == "https://example.com/.well-known/oauth-protected-resource/mcp?tenant=a"
+
+
 def test_metadata_url_construction_url_with_trailing_slash_only():
     """Test URL construction for resource with trailing slash only."""
     resource_url = AnyHttpUrl("https://example.com/")
@@ -136,6 +143,8 @@ def test_metadata_url_construction_url_with_trailing_slash_only():
         ("https://example.com", "https://example.com/.well-known/oauth-protected-resource"),
         ("https://example.com/", "https://example.com/.well-known/oauth-protected-resource"),
         ("https://example.com/mcp", "https://example.com/.well-known/oauth-protected-resource/mcp"),
+        ("https://example.com/mcp?tenant=a", "https://example.com/.well-known/oauth-protected-resource/mcp?tenant=a"),
+        ("https://example.com?tenant=a", "https://example.com/.well-known/oauth-protected-resource?tenant=a"),
         ("http://localhost:8001/mcp", "http://localhost:8001/.well-known/oauth-protected-resource/mcp"),
     ],
 )

tests/shared/test_auth_utils.py

@@ -100,6 +100,24 @@ def test_check_resource_allowed_path_boundary_matching():
     assert check_resource_allowed("https://example.com/api/v1", "https://example.com/api/") is True
 
 
+def test_check_resource_allowed_rejects_different_queries():
+    """Query-routed resources with different query components are distinct."""
+    assert (
+        check_resource_allowed(
+            "https://example.com/api?tenant=a",
+            "https://example.com/api?tenant=b",
+        )
+        is False
+    )
+    assert (
+        check_resource_allowed(
+            "https://example.com/api?tenant=a",
+            "https://example.com/api",
+        )
+        is False
+    )
+
+
 def test_check_resource_allowed_trailing_slash_handling():
     """Trailing slashes should be handled correctly."""
     # With and without trailing slashes