fix(client/auth): preserve an authorization endpoint's existing query parameters

RFC 6749 §3.1 allows the authorization endpoint URI to carry a query
component. The client built the authorization URL with a bare
f"{endpoint}?{params}", producing a second "?" and a broken URL for
any discovered authorization_endpoint that already has one (e.g.
Salesforce's "...?prompt=select_account").

Merge the flow's parameters into the endpoint's existing query instead.

Fixes #3505, fixes #2776.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: dgilman-perplexity

src/mcp/client/auth/oauth2.py

@@ -12,7 +12,7 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from dataclasses import dataclass, field
 from typing import Any, Protocol, get_args
-from urllib.parse import quote, urlencode, urljoin, urlparse
+from urllib.parse import parse_qsl, quote, urlencode, urljoin, urlparse, urlunparse
 
 import anyio
 import httpx2
@@ -61,6 +61,18 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _build_authorization_url(auth_endpoint: str, auth_params: dict[str, str]) -> str:
+    """Append authorization parameters to the endpoint, preserving any query it already has.
+
+    RFC 6749 §3.1 allows the authorization endpoint URI to include a query component; the
+    discovered `authorization_endpoint` therefore cannot be extended with a bare `?`.
+    """
+    parsed = urlparse(auth_endpoint)
+    query = urlencode(parse_qsl(parsed.query, keep_blank_values=True) + list(auth_params.items()))
+    return urlunparse(parsed._replace(query=query))
+
+
 # Methods a registered client's record may carry without a token request being an error,
 # derived from the set the SDK is willing to request so the two cannot drift. `None`/"none"
 # send no client secret. `private_key_jwt` sends none from here either: only
@@ -424,7 +436,7 @@ async def _perform_authorization_code_grant(self) -> tuple[str, str]:
             if "offline_access" in self.context.client_metadata.scope.split():
                 auth_params["prompt"] = "consent"
 
-        authorization_url = f"{auth_endpoint}?{urlencode(auth_params)}"
+        authorization_url = _build_authorization_url(auth_endpoint, auth_params)
         await self.context.redirect_handler(authorization_url)
 
         # Wait for callback

tests/client/test_auth.py

@@ -3667,3 +3667,45 @@ async def echo_callback() -> AuthorizationCodeResult:
         await auth_flow.asend(httpx2.Response(200, request=final_req))
     except StopAsyncIteration:
         pass
+
+
+@pytest.mark.anyio
+async def test_authorization_url_preserves_existing_endpoint_query(
+    oauth_provider: OAuthClientProvider,
+):
+    """RFC 6749 §3.1: the authorization endpoint URI may include a query component, so the
+    flow's parameters must be merged into it rather than appended after a second `?`."""
+    oauth_provider.context.oauth_metadata = OAuthMetadata(
+        issuer=AnyHttpUrl("https://auth.example.com"),
+        authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize?audience=mcp&prompt="),
+        token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+    )
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="test_client_id",
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+    )
+
+    captured_url: str | None = None
+    captured_state: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_url, captured_state
+        captured_url = url
+        captured_state = parse_qs(urlparse(url).query)["state"][0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    oauth_provider.context.redirect_handler = capture_redirect
+    oauth_provider.context.callback_handler = mock_callback
+
+    auth_code, _ = await oauth_provider._perform_authorization_code_grant()
+
+    assert auth_code == "auth_code"
+    assert captured_url is not None
+    assert captured_url.count("?") == 1
+    params = parse_qs(urlparse(captured_url).query, keep_blank_values=True)
+    assert params["audience"] == ["mcp"]  # the endpoint's own parameters survive
+    assert params["prompt"] == [""]  # including blank-valued ones
+    assert params["client_id"] == ["test_client_id"]
+    assert params["response_type"] == ["code"]