fix: allow passing query params in OAuthProxy upstream authorization url (#1630)

src/fastmcp/server/auth/oauth_proxy.py

@@ -474,9 +474,8 @@ class OAuthProxy(OAuthProvider):
             query_params["scope"] = " ".join(scopes_to_use)
 
         # Build the upstream authorization URL
-        upstream_url = (
-            f"{self._upstream_authorization_endpoint}?{urlencode(query_params)}"
-        )
+        separator = "&" if "?" in self._upstream_authorization_endpoint else "?"
+        upstream_url = f"{self._upstream_authorization_endpoint}{separator}{urlencode(query_params)}"
 
         logger.debug(
             "Starting OAuth transaction %s for client %s, redirecting to IdP",

tests/server/auth/test_oauth_proxy.py

@@ -114,6 +114,46 @@ class TestOAuthProxyComprehensive:
         )
         assert proxy2._redirect_path == "/auth/callback"
 
+    async def test_authorize_url_with_ampersand_separator(self, jwt_verifier):
+        """Test that authorize builds URLs with & separator when upstream endpoint has existing query parameters."""
+        # Test case: upstream endpoint with existing query parameters
+        proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://auth.example.com/authorize?version=2.0",
+            upstream_token_endpoint="https://auth.example.com/token",
+            upstream_client_id="client-123",
+            upstream_client_secret="secret-456",
+            token_verifier=jwt_verifier,
+            base_url="https://myserver.com",
+        )
+
+        client = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+        )
+
+        params = AuthorizationParams(
+            redirect_uri=AnyUrl("http://localhost:54321/callback"),
+            redirect_uri_provided_explicitly=True,
+            state="client-state",
+            code_challenge="challenge",
+            scopes=["read"],
+        )
+
+        # Should use "&" separator
+        redirect_url = await proxy.authorize(client, params)
+        parsed = urlparse(redirect_url)
+        query_params = parse_qs(parsed.query)
+
+        assert parsed.scheme == "https"
+        assert parsed.netloc == "auth.example.com"
+        assert parsed.path == "/authorize"
+        # Params in the original url are kept
+        assert "version" in query_params
+        assert query_params["version"] == ["2.0"]
+        # New params added correctly
+        assert query_params["response_type"] == ["code"]
+
     def test_dcr_always_enabled(self, jwt_verifier):
         """Test that DCR is always enabled for OAuth Proxy."""
         proxy = OAuthProxy(