feat: add Custom Token Exchange support (RFC 8693)

Implements get_token_by_exchange_profile() method to support Custom Token
Exchange via Token Exchange Profiles. This enables applications to exchange
custom tokens (from MCP servers, legacy systems, or partner services) for
Auth0 tokens while preserving user identity.

Key features:
- RFC 8693 compliant token exchange implementation
- Subject token validation (no whitespace, no "Bearer" prefix)
- Extra parameters support with security controls (denylist, DoS protection)
- Full type hints and comprehensive error handling
- Matches auth0-auth-js ApiClient.getTokenByExchangeProfile() functionality

Changes:
- Add get_token_by_exchange_profile() to ApiClient
- Add GetTokenByExchangeProfileError exception class
- Export GetTokenByExchangeProfileError in public API
- Add README documentation with usage examples

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: btiernay

README.md

@@ -113,6 +113,67 @@ asyncio.run(main())
 
 More info https://auth0.com/docs/secure/tokens/token-vault
 
+### 5. Custom Token Exchange (Early Access)
+
+> [!NOTE]
+> This feature is currently available in [Early Access](https://auth0.com/docs/troubleshoot/product-lifecycle/product-release-stages#early-access) for Enterprise customers. Please reach out to Auth0 support to get it enabled for your tenant.
+
+Custom Token Exchange allows you to exchange custom tokens for Auth0 tokens using RFC 8693. This is useful for:
+- Getting Auth0 tokens for another audience
+- Integrating external identity providers
+- Migrating to Auth0
+
+```python
+import asyncio
+
+from auth0_api_python import ApiClient, ApiClientOptions
+
+async def main():
+    api_client = ApiClient(ApiClientOptions(
+        domain="<AUTH0_DOMAIN>",
+        audience="<AUTH0_AUDIENCE>",
+        client_id="<AUTH0_CLIENT_ID>",
+        client_secret="<AUTH0_CLIENT_SECRET>",
+    ))
+
+    # The subject_token_type must match a Token Exchange Profile configured in Auth0
+    custom_token = "..."  # Your custom token from legacy system or external source
+
+    result = await api_client.get_token_by_exchange_profile(
+        subject_token=custom_token,
+        subject_token_type="urn:example:custom-token",  # Your custom token type URI
+        audience="https://api.example.com",
+        scope="openid profile read:data"
+    )
+
+    # Result contains access_token, expires_at, and optionally id_token, refresh_token
+    print(f"Access Token: {result['access_token']}")
+    print(f"Expires At: {result['expires_at']}")
+    if "id_token" in result:
+        print(f"ID Token: {result['id_token']}")
+
+asyncio.run(main())
+```
+
+#### Custom Parameters
+
+You can pass custom parameters to your Auth0 Action using the `extra` parameter:
+
+```python
+result = await api_client.get_token_by_exchange_profile(
+    subject_token=custom_token,
+    subject_token_type="urn:example:custom-token",
+    audience="https://api.example.com",
+    extra={
+        "device_id": "device-12345",
+        "session_token": "sess-abc",
+        "roles": ["admin", "user"]  # Arrays are supported
+    }
+)
+```
+
+More info: https://auth0.com/docs/authenticate/custom-token-exchange
+
 #### Requiring Additional Claims
 
 If your application demands extra claims, specify them with `required_claims`:
@@ -126,7 +187,7 @@ decoded_and_verified_token = await api_client.verify_access_token(
 
 If the token lacks `my_custom_claim` or fails any standard check (issuer mismatch, expired token, invalid signature), the method raises a `VerifyAccessTokenError`.
 
-### 5. DPoP Authentication
+### 6. DPoP Authentication
 
 > [!NOTE]  
 > This feature is currently available in [Early Access](https://auth0.com/docs/troubleshoot/product-lifecycle/product-release-stages#early-access). Please reach out to Auth0 support to get it enabled for your tenant.

src/auth0_api_python/__init__.py

@@ -7,8 +7,10 @@
 
 from .api_client import ApiClient
 from .config import ApiClientOptions
+from .errors import GetTokenByExchangeProfileError
 
 __all__ = [
     "ApiClient",
-    "ApiClientOptions"
+    "ApiClientOptions",
+    "GetTokenByExchangeProfileError"
 ]

src/auth0_api_python/api_client.py

@@ -1,5 +1,5 @@
 import time
-from typing import Any, Optional
+from typing import Any, Optional, Union
 
 import httpx
 from authlib.jose import JsonWebKey, JsonWebToken
@@ -9,6 +9,7 @@
     ApiError,
     BaseAuthError,
     GetAccessTokenForConnectionError,
+    GetTokenByExchangeProfileError,
     InvalidAuthSchemeError,
     InvalidDpopProofError,
     MissingAuthorizationError,
@@ -501,6 +502,221 @@ async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict
                 exc
             )
 
+    async def get_token_by_exchange_profile(
+        self,
+        subject_token: str,
+        subject_token_type: str,
+        audience: Optional[str] = None,
+        scope: Optional[str] = None,
+        requested_token_type: Optional[str] = None,
+        extra: Optional[dict[str, Union[str, list[str]]]] = None
+    ) -> dict[str, Any]:
+        """
+        Exchanges a token via a Custom Token Exchange Profile for Auth0 tokens (RFC 8693).
+
+        This method supports Custom Token Exchange for custom token types via a configured
+        Token Exchange Profile. It exchanges custom tokens (from MCP servers, legacy systems,
+        or partner services) for Auth0 tokens targeting a specific API audience while
+        preserving user identity.
+
+        **Note**: This method requires a confidential client (client_id and client_secret
+        must be configured).
+
+        Args:
+            subject_token: The raw token to be exchanged (without "Bearer " prefix)
+            subject_token_type: URI identifying the token type (must match a Token Exchange Profile)
+            audience: Optional target API identifier for the exchanged tokens
+            scope: Optional space-separated OAuth 2.0 scopes to request
+            requested_token_type: Optional type of token to issue (defaults to access token)
+            extra: Optional custom parameters accessible in Auth0 Actions. Cannot override
+                   reserved OAuth parameters. Array values limited to 20 items per key.
+
+        Returns:
+            Dictionary containing:
+            - access_token (str): The Auth0 access token
+            - expires_at (int): Unix timestamp when token expires
+            - id_token (str, optional): OpenID Connect ID token
+            - refresh_token (str, optional): Refresh token
+            - scope (str, optional): Granted scopes
+            - token_type (str, optional): Token type (typically "Bearer")
+            - issued_token_type (str, optional): RFC 8693 issued token type identifier
+
+        Raises:
+            MissingRequiredArgumentError: If required parameters are missing
+            GetTokenByExchangeProfileError: If client credentials not configured or exchange fails
+            ApiError: If the token endpoint returns an error
+
+        Example:
+            >>> result = await api_client.get_token_by_exchange_profile(
+            ...     subject_token=custom_token,
+            ...     subject_token_type="urn:example:custom-token",
+            ...     audience="https://api.backend.com",
+            ...     scope="openid profile read:data"
+            ... )
+            >>> print(result["access_token"])
+
+        References:
+            - Custom Token Exchange Documentation: https://auth0.com/docs/authenticate/custom-token-exchange
+            - RFC 8693 OAuth 2.0 Token Exchange: https://datatracker.ietf.org/doc/html/rfc8693
+        """
+        # Constants
+        TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
+        MAX_ARRAY_VALUES_PER_KEY = 20
+
+        # OAuth parameter denylist
+        PARAM_DENYLIST = frozenset([
+            "grant_type", "client_id", "client_secret", "client_assertion",
+            "client_assertion_type", "subject_token", "subject_token_type",
+            "requested_token_type", "actor_token", "actor_token_type",
+            "audience", "aud", "resource", "resources", "resource_indicator",
+            "scope", "connection", "login_hint", "organization", "assertion",
+        ])
+
+        # Validate required parameters
+        if not subject_token:
+            raise MissingRequiredArgumentError("subject_token")
+        if not subject_token_type:
+            raise MissingRequiredArgumentError("subject_token_type")
+
+        # Validate subject token format
+        if not isinstance(subject_token, str):
+            raise GetTokenByExchangeProfileError("subject_token must be a string")
+        if not subject_token.strip():
+            raise GetTokenByExchangeProfileError("subject_token cannot be blank or whitespace")
+        if subject_token != subject_token.strip():
+            raise GetTokenByExchangeProfileError(
+                "subject_token must not include leading or trailing whitespace"
+            )
+        if subject_token.lower().startswith("bearer "):
+            raise GetTokenByExchangeProfileError(
+                "subject_token must not include the 'Bearer ' prefix"
+            )
+
+        # Require client credentials
+        client_id = self.options.client_id
+        client_secret = self.options.client_secret
+        if not client_id or not client_secret:
+            raise GetTokenByExchangeProfileError(
+                "Client credentials are required to use get_token_by_exchange_profile"
+            )
+
+        # Discover token endpoint
+        metadata = await self._discover()
+        token_endpoint = metadata.get("token_endpoint")
+        if not token_endpoint:
+            raise GetTokenByExchangeProfileError("Token endpoint missing in OIDC metadata")
+
+        # Build request parameters
+        params = {
+            "grant_type": TOKEN_EXCHANGE_GRANT_TYPE,
+            "client_id": client_id,
+            "subject_token": subject_token,
+            "subject_token_type": subject_token_type,
+        }
+
+        # Add optional parameters
+        if audience:
+            params["audience"] = audience
+        if scope:
+            params["scope"] = scope
+        if requested_token_type:
+            params["requested_token_type"] = requested_token_type
+
+        # Append extra parameters with validation
+        if extra:
+            for parameter_key, parameter_value in extra.items():
+                # Silently ignore denylisted parameters
+                if parameter_key in PARAM_DENYLIST:
+                    continue
+
+                if isinstance(parameter_value, list):
+                    # Validate array size for DoS protection
+                    if len(parameter_value) > MAX_ARRAY_VALUES_PER_KEY:
+                        raise GetTokenByExchangeProfileError(
+                            f"Parameter '{parameter_key}' exceeds maximum array size of {MAX_ARRAY_VALUES_PER_KEY}"
+                        )
+                    # Store as list - httpx will encode as multiple key=value pairs
+                    params[parameter_key] = parameter_value
+                else:
+                    params[parameter_key] = str(parameter_value)
+
+        # Make token exchange request
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    token_endpoint,
+                    data=params,
+                    auth=(client_id, client_secret)
+                )
+
+                if response.status_code != 200:
+                    error_data = response.json() if "json" in response.headers.get(
+                        "content-type", "").lower() else {}
+                    raise ApiError(
+                        error_data.get("error", "token_exchange_error"),
+                        error_data.get(
+                            "error_description",
+                            f"Failed to exchange token of type '{subject_token_type}'"
+                            + (f" for audience '{audience}'" if audience else "")
+                        ),
+                        response.status_code
+                    )
+
+                try:
+                    token_response = response.json()
+                except Exception:
+                    raise ApiError("invalid_json", "Token endpoint returned invalid JSON.", 502)
+
+                # Validate required fields
+                access_token = token_response.get("access_token")
+                if not isinstance(access_token, str) or not access_token:
+                    raise ApiError(
+                        "invalid_response",
+                        "Missing or invalid access_token in response.",
+                        502
+                    )
+
+                expires_in_raw = token_response.get("expires_in", 3600)
+                try:
+                    expires_in = int(expires_in_raw)
+                except (TypeError, ValueError):
+                    raise ApiError("invalid_response", "expires_in is not an integer.", 502)
+
+                # Build response (match JS SDK structure)
+                result = {
+                    "access_token": access_token,
+                    "expires_at": int(time.time()) + expires_in,
+                }
+
+                # Add optional fields if present (conditional spreading like JS)
+                if "scope" in token_response and token_response["scope"]:
+                    result["scope"] = token_response["scope"]
+                if "id_token" in token_response and token_response["id_token"]:
+                    result["id_token"] = token_response["id_token"]
+                if "refresh_token" in token_response and token_response["refresh_token"]:
+                    result["refresh_token"] = token_response["refresh_token"]
+                if "token_type" in token_response and token_response["token_type"]:
+                    result["token_type"] = token_response["token_type"]
+                if "issued_token_type" in token_response and token_response["issued_token_type"]:
+                    result["issued_token_type"] = token_response["issued_token_type"]
+
+                return result
+
+        except httpx.TimeoutException as exc:
+            raise ApiError(
+                "timeout_error",
+                f"Request to token endpoint timed out: {str(exc)}",
+                504,
+                exc
+            )
+        except httpx.HTTPError as exc:
+            raise ApiError(
+                "network_error",
+                f"Network error occurred: {str(exc)}",
+                502,
+                exc
+            )
+
     # ===== Private Methods =====
 
     async def _discover(self) -> dict[str, Any]:

src/auth0_api_python/errors.py

@@ -106,6 +106,16 @@ def get_error_code(self) -> str:
         return "get_access_token_for_connection_error"
 
 
+class GetTokenByExchangeProfileError(BaseAuthError):
+    """Error raised when getting a token via exchange profile fails."""
+
+    def get_status_code(self) -> int:
+        return 400
+
+    def get_error_code(self) -> str:
+        return "get_token_by_exchange_profile_error"
+
+
 class ApiError(BaseAuthError):
     """
     Error raised when an API request to Auth0 fails.