fix(user_management): honor client_id in PKCE authorization URL

src/workos/sso/_resource.py

@@ -145,6 +145,7 @@ def get_authorization_url(
         *,
         provider_scopes: Optional[List[str]] = None,
         provider_query_params: Optional[Dict[str, str]] = None,
+        client_id: Optional[str] = None,
         domain: Optional[str] = None,
         provider: Optional[Union[SSOProvider, str]] = None,
         redirect_uri: str,
@@ -164,6 +165,7 @@ def get_authorization_url(
         Args:
             provider_scopes: Additional scopes to request from the identity provider. Applicable when using OAuth or OpenID Connect connections.
             provider_query_params: Key/value pairs of query parameters to pass to the OAuth provider. Only applicable when using OAuth connections.
+            client_id: The unique identifier of the WorkOS environment client. Defaults to the client's configured client_id.
             domain: (deprecated) Deprecated. Use `connection` or `organization` instead. Used to initiate SSO for a connection by domain. The domain must be associated with a connection in your WorkOS environment.
             provider: Used to initiate OAuth authentication with various providers.
             redirect_uri: Where to redirect the user after they complete the authentication process. You must use one of the redirect URIs configured via the [Redirects](https://dashboard.workos.com/redirects) page on the dashboard.
@@ -193,6 +195,7 @@ def get_authorization_url(
                 if provider_scopes is not None
                 else None,
                 "provider_query_params": provider_query_params,
+                "client_id": client_id,
                 "domain": domain,
                 "provider": enum_value(provider) if provider is not None else None,
                 "redirect_uri": redirect_uri,
@@ -207,7 +210,7 @@ def get_authorization_url(
             if v is not None
         }
         params["response_type"] = "code"
-        if self._client.client_id is not None:
+        if "client_id" not in params and self._client.client_id is not None:
             params["client_id"] = self._client.client_id
         return self._client.build_url(("sso", "authorize"), params)
 
@@ -565,6 +568,7 @@ def get_authorization_url(
         *,
         provider_scopes: Optional[List[str]] = None,
         provider_query_params: Optional[Dict[str, str]] = None,
+        client_id: Optional[str] = None,
         domain: Optional[str] = None,
         provider: Optional[Union[SSOProvider, str]] = None,
         redirect_uri: str,
@@ -584,6 +588,7 @@ def get_authorization_url(
         Args:
             provider_scopes: Additional scopes to request from the identity provider. Applicable when using OAuth or OpenID Connect connections.
             provider_query_params: Key/value pairs of query parameters to pass to the OAuth provider. Only applicable when using OAuth connections.
+            client_id: The unique identifier of the WorkOS environment client. Defaults to the client's configured client_id.
             domain: (deprecated) Deprecated. Use `connection` or `organization` instead. Used to initiate SSO for a connection by domain. The domain must be associated with a connection in your WorkOS environment.
             provider: Used to initiate OAuth authentication with various providers.
             redirect_uri: Where to redirect the user after they complete the authentication process. You must use one of the redirect URIs configured via the [Redirects](https://dashboard.workos.com/redirects) page on the dashboard.
@@ -613,6 +618,7 @@ def get_authorization_url(
                 if provider_scopes is not None
                 else None,
                 "provider_query_params": provider_query_params,
+                "client_id": client_id,
                 "domain": domain,
                 "provider": enum_value(provider) if provider is not None else None,
                 "redirect_uri": redirect_uri,
@@ -627,7 +633,7 @@ def get_authorization_url(
             if v is not None
         }
         params["response_type"] = "code"
-        if self._client.client_id is not None:
+        if "client_id" not in params and self._client.client_id is not None:
             params["client_id"] = self._client.client_id
         return self._client.build_url(("sso", "authorize"), params)
 

src/workos/user_management/_resource.py

@@ -565,6 +565,7 @@ def get_authorization_url(
         state: Optional[str] = None,
         organization_id: Optional[str] = None,
         redirect_uri: str,
+        client_id: Optional[str] = None,
         request_options: Optional[RequestOptions] = None,
     ) -> str:
         """Get an authorization URL
@@ -587,6 +588,7 @@ def get_authorization_url(
             state: An opaque value used to maintain state between the request and the callback.
             organization_id: The ID of the organization to authenticate the user against.
             redirect_uri: The callback URI where the authorization code will be sent after authentication.
+            client_id: The unique identifier of the WorkOS environment client. Defaults to the client's configured client_id.
             request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.
 
         Returns:
@@ -619,11 +621,12 @@ def get_authorization_url(
                 "state": state,
                 "organization_id": organization_id,
                 "redirect_uri": redirect_uri,
+                "client_id": client_id,
             }.items()
             if v is not None
         }
         params["response_type"] = "code"
-        if self._client.client_id is not None:
+        if "client_id" not in params and self._client.client_id is not None:
             params["client_id"] = self._client.client_id
         return self._client.build_url(("user_management", "authorize"), params)
 
@@ -2373,6 +2376,7 @@ def get_authorization_url_with_pkce(
 
         url = self.get_authorization_url(
             redirect_uri=redirect_uri,
+            client_id=client_id,
             code_challenge=pair.code_challenge,
             code_challenge_method="S256",
             state=state,
@@ -2924,6 +2928,7 @@ def get_authorization_url(
         state: Optional[str] = None,
         organization_id: Optional[str] = None,
         redirect_uri: str,
+        client_id: Optional[str] = None,
         request_options: Optional[RequestOptions] = None,
     ) -> str:
         """Get an authorization URL
@@ -2946,6 +2951,7 @@ def get_authorization_url(
             state: An opaque value used to maintain state between the request and the callback.
             organization_id: The ID of the organization to authenticate the user against.
             redirect_uri: The callback URI where the authorization code will be sent after authentication.
+            client_id: The unique identifier of the WorkOS environment client. Defaults to the client's configured client_id.
             request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.
 
         Returns:
@@ -2978,11 +2984,12 @@ def get_authorization_url(
                 "state": state,
                 "organization_id": organization_id,
                 "redirect_uri": redirect_uri,
+                "client_id": client_id,
             }.items()
             if v is not None
         }
         params["response_type"] = "code"
-        if self._client.client_id is not None:
+        if "client_id" not in params and self._client.client_id is not None:
             params["client_id"] = self._client.client_id
         return self._client.build_url(("user_management", "authorize"), params)
 
@@ -4732,6 +4739,7 @@ async def get_authorization_url_with_pkce(
 
         url = self.get_authorization_url(
             redirect_uri=redirect_uri,
+            client_id=client_id,
             code_challenge=pair.code_challenge,
             code_challenge_method="S256",
             state=state,

tests/test_inline_helpers.py

@@ -94,6 +94,19 @@ def test_with_organization_id(self, workos):
         )
         assert "organization_id=org_01" in result["url"]
 
+    def test_uses_configured_client_id(self, workos):
+        result = workos.user_management.get_authorization_url_with_pkce(
+            redirect_uri="https://example.com/callback"
+        )
+        assert "client_id=client_test" in result["url"]
+
+    def test_explicit_client_id_overrides_configured(self, workos):
+        result = workos.user_management.get_authorization_url_with_pkce(
+            redirect_uri="https://example.com/callback", client_id="client_override"
+        )
+        assert "client_id=client_override" in result["url"]
+        assert "client_id=client_test" not in result["url"]
+
 
 @pytest.mark.asyncio
 class TestAsyncAuthKitPKCEAuthorizationUrl:
@@ -112,6 +125,13 @@ async def test_url_contains_pkce_params(self, async_workos):
         assert "code_challenge=" in result["url"]
         assert "code_challenge_method=S256" in result["url"]
 
+    async def test_explicit_client_id_overrides_configured(self, async_workos):
+        result = await async_workos.user_management.get_authorization_url_with_pkce(
+            redirect_uri="https://example.com/callback", client_id="client_override"
+        )
+        assert "client_id=client_override" in result["url"]
+        assert "client_id=client_test" not in result["url"]
+
 
 class TestAuthKitPKCECodeExchange:
     def test_sends_code_verifier(self, workos, httpx_mock):