server_client.py

"""
Main client for auth0-server-python SDK.
Handles authentication flows, token management, and user sessions.
"""

import asyncio
import json
import time
from collections import OrderedDict
from typing import Any, Callable, Generic, Optional, TypeVar, Union
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse

import httpx
import jwt
from authlib.integrations.base_client.errors import OAuthError
from authlib.integrations.httpx_client import AsyncOAuth2Client
from pydantic import ValidationError

from auth0_server_python.auth_server.mfa_client import MfaClient
from auth0_server_python.auth_server.my_account_client import MyAccountClient
from auth0_server_python.auth_types import (
    CompleteConnectAccountRequest,
    CompleteConnectAccountResponse,
    ConnectAccountOptions,
    ConnectAccountRequest,
    CustomTokenExchangeOptions,
    ListConnectedAccountConnectionsResponse,
    ListConnectedAccountsResponse,
    LoginWithCustomTokenExchangeOptions,
    LoginWithCustomTokenExchangeResult,
    LogoutOptions,
    LogoutTokenClaims,
    MfaRequirements,
    StartInteractiveLoginOptions,
    StateData,
    TokenExchangeResponse,
    TokenSet,
    TransactionData,
    UserClaims,
)
from auth0_server_python.error import (
    AccessTokenError,
    AccessTokenErrorCode,
    AccessTokenForConnectionError,
    AccessTokenForConnectionErrorCode,
    ApiError,
    BackchannelLogoutError,
    ConfigurationError,
    CustomTokenExchangeError,
    CustomTokenExchangeErrorCode,
    DomainResolverError,
    InvalidArgumentError,
    IssuerValidationError,
    MfaRequiredError,
    MissingRequiredArgumentError,
    MissingTransactionError,
    OrganizationTokenValidationError,
    PollingApiError,
    StartLinkUserError,
)
from auth0_server_python.telemetry import Telemetry
from auth0_server_python.utils import PKCE, URL, State
from auth0_server_python.utils.helpers import (
    build_domain_resolver_context,
    validate_org_claims,
    validate_resolved_domain_value,
)

# Generic type for store options
TStoreOptions = TypeVar('TStoreOptions')
# redirect_uri is intentionally excluded — in MCD mode it is built
# dynamically from the resolved domain at login time.
INTERNAL_AUTHORIZE_PARAMS = ["client_id", "response_type",
                             "code_challenge", "code_challenge_method", "state", "nonce", "scope"]


class ServerClient(Generic[TStoreOptions]):
    """
    Main client for Auth0 server SDK. Handles authentication flows, session management,
    and token operations using Authlib for OIDC functionality.
    """
    DEFAULT_AUDIENCE_STATE_KEY = "default"

    # ============================================================================
    # INITIALIZATION
    # ============================================================================

    def __init__(
        self,
        domain: Union[str, Callable[[Optional[dict[str, Any]]], str]] = None,
        client_id: str = None,
        client_secret: str = None,
        redirect_uri: Optional[str] = None,
        secret: str = None,
        transaction_store=None,
        state_store=None,
        transaction_identifier: str = "_a0_tx",
        state_identifier: str = "_a0_session",
        authorization_params: Optional[dict[str, Any]] = None,
        pushed_authorization_requests: bool = False,
        organization: Optional[str] = None,
    ):
        """
        Initialize the Auth0 server client.

        Args:
            domain: Auth0 domain - either a static string (e.g., 'tenant.auth0.com') or a callable that resolves domain dynamically.
            client_id: Auth0 client ID
            client_secret: Auth0 client secret
            redirect_uri: Default redirect URI for authentication
            secret: Secret used for encryption
            transaction_store: Custom transaction store (defaults to MemoryTransactionStore)
            state_store: Custom state store (defaults to MemoryStateStore)
            transaction_identifier: Identifier for transaction data
            state_identifier: Identifier for state data
            authorization_params: Default parameters for authorization requests
            pushed_authorization_requests: Whether to use Pushed Authorization Requests
            organization: Default organization for all login flows from this client.
                Can be an org ID (e.g. 'org_abc123') or an org name (e.g. 'acme-corp').
                Per-login values passed in StartInteractiveLoginOptions always override this.
        """
        if not secret:
            raise MissingRequiredArgumentError("secret")

        if domain is None:
            raise ConfigurationError(
                "Domain is required"
            )

        # Validate domain type
        if not isinstance(domain, str) and not callable(domain):
            raise ConfigurationError(
                f"Domain must be either a string or a callable function. "
                f"Got {type(domain).__name__} instead."
            )

        # Determine if domain is static string or dynamic callable
        if callable(domain):
            self._domain = None
            self._domain_resolver = domain
        else:
            # Validate static domain string
            domain_str = str(domain)
            if not domain_str or domain_str.strip() == "":
                raise ConfigurationError("Domain cannot be empty.")
            self._domain = domain_str
            self._domain_resolver = None

        self._client_id = client_id
        self._client_secret = client_secret
        self._redirect_uri = redirect_uri
        self._secret = secret
        self._default_authorization_params = authorization_params or {}
        self._pushed_authorization_requests = pushed_authorization_requests  # store the flag
        self._organization = organization

        # Initialize stores
        self._transaction_store = transaction_store
        self._state_store = state_store
        self._transaction_identifier = transaction_identifier
        self._state_identifier = state_identifier

        # Initialize telemetry
        self._telemetry = Telemetry.default()
        self._telemetry_headers = self._telemetry.headers

        # Initialize OAuth client
        self._oauth = AsyncOAuth2Client(
            client_id=client_id,
            client_secret=client_secret,
            headers=self._telemetry_headers,
        )

        self._my_account_client = MyAccountClient(
            domain=domain, headers=self._telemetry_headers
        )

        # Unified cache for OIDC metadata and JWKS per domain (LRU eviction + TTL)
        self._discovery_cache: OrderedDict[str, dict] = OrderedDict()
        self._cache_ttl = 600     # 10 mins. TTL
        self._cache_max_entries = 100 # Max 100 domains

        # Initialize MFA client
        self._mfa_client = MfaClient(
            domain=domain,
            client_id=self._client_id,
            client_secret=self._client_secret,
            secret=self._secret,
            state_store=self._state_store,
            state_identifier=self._state_identifier,
            headers=self._telemetry_headers,
        )

    def _get_http_client(self, **kwargs) -> httpx.AsyncClient:
        """Return an httpx.AsyncClient with telemetry headers injected."""
        headers = {**kwargs.pop("headers", {}), **self._telemetry_headers}
        return httpx.AsyncClient(headers=headers, **kwargs)

    def _normalize_url(self, value: str) -> str:
        """
        Normalize a URL-like value (domain or issuer) for comparison.
        Ensures https:// scheme, lowercases, and strips trailing slashes.
        """
        if not value:
            return value

        value = value.lower()
        if value.startswith('https://'):
            pass
        elif value.startswith('http://'):
            value = value.replace('http://', 'https://')
        else:
            value = f'https://{value}'

        return value.rstrip('/')


    async def _resolve_current_domain(self, store_options=None) -> str:
        """Resolve domain from resolver function or return static domain."""
        if self._domain_resolver:
            context = build_domain_resolver_context(store_options)
            try:
                resolved = await self._domain_resolver(context)
                return validate_resolved_domain_value(resolved)
            except DomainResolverError:
                raise
            except Exception as e:
                raise DomainResolverError(
                    f"Domain resolver function raised an exception: {str(e)}",
                    original_error=e
                )
        return self._domain

    def _get_session_domain(self, state_data_dict: dict) -> Optional[str]:
        """
        Get session domain with 3-tier fallback for backward compatibility.

        Legacy sessions created before MCD support may not have a 'domain' field.
        Fallback chain matches JS SDK's #getSessionDomain:
        1. state_data.domain — new sessions have this
        2. self._domain — static domain (if configured)
        3. Extract hostname from user.iss — derive from user's issuer claim
        """
        domain = state_data_dict.get('domain')
        if domain:
            return domain

        if self._domain:
            return self._domain

        user = state_data_dict.get('user')
        if isinstance(user, dict):
            iss = user.get('iss')
        else:
            iss = getattr(user, 'iss', None) if user else None

        if iss:
            parsed = urlparse(iss)
            return parsed.hostname

        return None

    async def _verify_and_decode_jwt(
        self, token: str, jwks: dict, audience: Optional[str] = None
    ) -> dict:
        """
        Find signing key in JWKS by kid and decode JWT.

        Verifies signature but disables built-in issuer validation
        (callers perform normalized issuer comparison separately).

        Args:
            token: The JWT to verify and decode
            jwks: The JWKS dict containing signing keys
            audience: Optional expected audience claim

        Returns:
            Decoded claims dictionary

        Raises:
            ValueError: If no matching key found in JWKS for the token's kid
        """
        unverified_header = jwt.get_unverified_header(token)
        kid = unverified_header.get("kid")

        signing_key = None
        for key in jwks.get("keys", []):
            if key.get("kid") == kid:
                signing_key = jwt.PyJWK.from_dict(key)
                break

        if not signing_key:
            raise ValueError(f"No matching key found in JWKS for kid: {kid}")

        decode_options = {"verify_signature": True, "verify_iss": False}
        kwargs = {"algorithms": ["RS256"], "options": decode_options}
        if audience:
            kwargs["audience"] = audience

        return jwt.decode(token, signing_key.key, **kwargs)

    async def _fetch_oidc_metadata(self, domain: str) -> dict:
        """Fetch OIDC metadata from domain."""
        normalized_domain = self._normalize_url(domain)
        metadata_url = f"{normalized_domain}/.well-known/openid-configuration"
        async with self._get_http_client() as client:
            response = await client.get(metadata_url)
            response.raise_for_status()
            return response.json()

    def _purge_expired_cache_entries(self):
        """Purge all expired entries from the discovery cache."""
        now = time.time()
        expired = [k for k, v in self._discovery_cache.items() if v["expires_at"] <= now]
        for k in expired:
            del self._discovery_cache[k]

    def _ensure_cache_capacity(self):
        """Evict least recently used entry if cache is at capacity."""
        if len(self._discovery_cache) >= self._cache_max_entries:
            self._discovery_cache.popitem(last=False)

    async def _get_oidc_metadata_cached(self, domain: str) -> dict:
        """
        Get OIDC metadata with caching (LRU eviction + TTL).

        Uses a unified cache shared with JWKS when metadata expires,
        the corresponding JWKS is also invalidated.

        Args:
            domain: Auth0 domain

        Returns:
            OIDC metadata document
        """
        now = time.time()

        # Check cache
        if domain in self._discovery_cache:
            cached = self._discovery_cache[domain]
            if cached["expires_at"] > now:
                self._discovery_cache.move_to_end(domain)
                return cached["metadata"]
            # Expired — remove entire entry (metadata + jwks)
            del self._discovery_cache[domain]

        # Cache miss/expired - fetch fresh
        metadata = await self._fetch_oidc_metadata(domain)

        # Purge expired entries and ensure capacity
        self._purge_expired_cache_entries()
        self._ensure_cache_capacity()

        # Store in cache with jwks=None (lazily populated when needed)
        self._discovery_cache[domain] = {
            "metadata": metadata,
            "jwks": None,
            "expires_at": now + self._cache_ttl
        }

        return metadata

    async def _fetch_jwks(self, jwks_uri: str) -> dict:
        """
        Fetch JWKS (JSON Web Key Set) from jwks_uri.

        Args:
            jwks_uri: The JWKS endpoint URL

        Returns:
            JWKS document containing public keys

        Raises:
            ApiError: If JWKS fetch fails
        """
        try:
            async with self._get_http_client() as client:
                response = await client.get(jwks_uri)
                response.raise_for_status()
                return response.json()
        except Exception as e:
            raise ApiError("jwks_fetch_error", f"Failed to fetch JWKS from {jwks_uri}", e)

    async def _get_jwks_cached(self, domain: str, metadata: dict = None) -> dict:
        """
        Get JWKS with caching using OIDC discovery (LRU eviction + TTL).

        Uses a unified cache shared with metadata — JWKS is lazily populated
        on first access and invalidated when the cache entry expires.

        Args:
            domain: Auth0 domain
            metadata: Optional OIDC metadata (if already fetched)

        Returns:
            JWKS document

        Raises:
            ApiError: If JWKS fetch fails or jwks_uri missing from metadata
        """
        now = time.time()

        # Check cache — entry exists, not expired, and jwks already fetched
        if domain in self._discovery_cache:
            cached = self._discovery_cache[domain]
            if cached["expires_at"] > now:
                if cached["jwks"] is not None:
                    self._discovery_cache.move_to_end(domain)
                    return cached["jwks"]
                # Entry valid but jwks not yet fetched — use cached metadata
                metadata = cached["metadata"]
            else:
                # Expired — remove entire entry
                del self._discovery_cache[domain]

        # Get metadata if not available from cache or parameter
        if not metadata:
            metadata = await self._get_oidc_metadata_cached(domain)

        jwks_uri = metadata.get('jwks_uri')
        if not jwks_uri:
            raise ApiError(
                "missing_jwks_uri",
                f"OIDC metadata for {domain} does not contain jwks_uri. Provider may be non-RFC-compliant."
            )

        # Fetch JWKS
        jwks = await self._fetch_jwks(jwks_uri)

        # Update existing cache entry with jwks (entry created by _get_oidc_metadata_cached)
        if domain in self._discovery_cache:
            self._discovery_cache[domain]["jwks"] = jwks
            self._discovery_cache.move_to_end(domain)
        else:
            # Edge case: entry was evicted between metadata and jwks fetch
            self._purge_expired_cache_entries()
            self._ensure_cache_capacity()
            self._discovery_cache[domain] = {
                "metadata": metadata,
                "jwks": jwks,
                "expires_at": now + self._cache_ttl
            }

        return jwks

    # ============================================================================
    # INTERACTIVE LOGIN FLOW
    # Handles browser-based authentication using the Authorization Code flow
    # with PKCE for secure token exchange.
    # ============================================================================

    async def start_interactive_login(
        self,
        options: Optional[StartInteractiveLoginOptions] = None,
        store_options: dict = None
    ) -> str:
        """
        Starts the interactive login process and returns a URL to redirect to.

        Args:
            options: Configuration options for the login process
            store_options: Store options containing request/response

        Returns:
            Authorization URL to redirect the user to
        """
        options = options or StartInteractiveLoginOptions()

        # Resolve domain (static or dynamic)
        origin_domain = await self._resolve_current_domain(store_options)

        # Fetch OIDC metadata from resolved domain
        try:
            metadata = await self._get_oidc_metadata_cached(origin_domain)
        except Exception as e:
            raise ApiError("metadata_error",
                           "Failed to fetch OIDC metadata", e)

        # Get effective authorization params (merge defaults with provided ones)
        auth_params = dict(self._default_authorization_params)
        if options.authorization_params:
            auth_params.update(
                {k: v for k, v in options.authorization_params.items(
                ) if k not in INTERNAL_AUTHORIZE_PARAMS}
            )

        # Ensure we have a redirect_uri
        if "redirect_uri" not in auth_params and not self._redirect_uri:
            raise MissingRequiredArgumentError("redirect_uri")

        # Use the default redirect_uri if none is specified
        if "redirect_uri" not in auth_params and self._redirect_uri:
            auth_params["redirect_uri"] = self._redirect_uri

        # Generate PKCE code verifier and challenge
        code_verifier = PKCE.generate_code_verifier()
        code_challenge = PKCE.generate_code_challenge(code_verifier)

        # Add PKCE parameters to the authorization request
        auth_params["code_challenge"] = code_challenge
        auth_params["code_challenge_method"] = "S256"

        # State parameter to prevent CSRF
        state = PKCE.generate_random_string(32)
        auth_params["state"] = state

        # Merge any requested scope with defaults
        requested_scope = options.authorization_params.get("scope", None) if options.authorization_params else None
        audience = auth_params.get("audience", None)
        merged_scope = self._merge_scope_with_defaults(requested_scope, audience)
        auth_params["scope"] = merged_scope

        # Typed org/invitation fields win over anything already in auth_params from authorization_params.
        resolved_org = options.organization or self._organization
        if resolved_org and not resolved_org.strip():
            raise InvalidArgumentError("organization", "organization must not be blank")
        if resolved_org:
            auth_params["organization"] = resolved_org

        if options.invitation:
            auth_params["invitation"] = options.invitation

        # Build the transaction data to store with domain
        transaction_data = TransactionData(
            code_verifier=code_verifier,
            app_state=options.app_state,
            audience=audience,
            domain=origin_domain,
            redirect_uri=auth_params.get("redirect_uri"),
            organization=resolved_org,
        )

        # Store the transaction data
        await self._transaction_store.set(
            f"{self._transaction_identifier}:{state}",
            transaction_data,
            options=store_options
        )

        # Set metadata for OAuth client
        self._oauth.metadata = metadata
        # If PAR is enabled, use the PAR endpoint
        if self._pushed_authorization_requests:
            par_endpoint = self._oauth.metadata.get(
                "pushed_authorization_request_endpoint")
            if not par_endpoint:
                raise ApiError(
                    "configuration_error", "PAR is enabled but pushed_authorization_request_endpoint is missing in metadata")

            auth_params["client_id"] = self._client_id
            # Post the auth_params to the PAR endpoint
            async with self._get_http_client() as client:
                par_response = await client.post(
                    par_endpoint,
                    data=auth_params,
                    auth=(self._client_id, self._client_secret)
                )
                if par_response.status_code not in (200, 201):
                    error_data = par_response.json()
                    raise ApiError(
                        error_data.get("error", "par_error"),
                        error_data.get(
                            "error_description", "Failed to obtain request_uri from PAR endpoint")
                    )
                par_data = par_response.json()
                request_uri = par_data.get("request_uri")
                if not request_uri:
                    raise ApiError(
                        "par_error", "No request_uri returned from PAR endpoint")

            auth_endpoint = self._oauth.metadata.get("authorization_endpoint")
            final_url = f"{auth_endpoint}?request_uri={request_uri}&response_type={auth_params['response_type']}&client_id={self._client_id}"
            return final_url
        else:
            if "authorization_endpoint" not in self._oauth.metadata:
                raise ApiError("configuration_error",
                               "Authorization endpoint missing in OIDC metadata")

            authorization_endpoint = self._oauth.metadata["authorization_endpoint"]

            try:
                auth_url, state = self._oauth.create_authorization_url(
                    authorization_endpoint, **auth_params)
            except Exception as e:
                raise ApiError("authorization_url_error",
                               "Failed to create authorization URL", e)

            return auth_url

    async def complete_interactive_login(
        self,
        url: str,
        store_options: dict = None
    ) -> dict[str, Any]:
        """
        Completes the login process after user is redirected back.

        Args:
            url: The full callback URL including query parameters
            store_options: Options to pass to the state store

        Returns:
            Dictionary containing session data and app state
        """
        # Parse the URL to get query parameters
        parsed_url = urlparse(url)
        query_params = parse_qs(parsed_url.query)

        # Get state parameter from the URL
        state = query_params.get("state", [""])[0]
        if not state:
            raise MissingRequiredArgumentError("state")

        # Retrieve the transaction data using the state
        transaction_identifier = f"{self._transaction_identifier}:{state}"
        transaction_data = await self._transaction_store.get(transaction_identifier, options=store_options)

        if not transaction_data:
            raise MissingTransactionError()

        # Check for error response from Auth0
        if "error" in query_params:
            error = query_params.get("error", [""])[0]
            error_description = query_params.get("error_description", [""])[0]
            raise ApiError(error, error_description)

        # Get the authorization code from the URL
        code = query_params.get("code", [""])[0]
        if not code:
            raise MissingRequiredArgumentError("code")

        # Get domain from transaction, or resolve domain if not present (resolver mode)
        origin_domain = transaction_data.domain or await self._resolve_current_domain(store_options)

        # Fetch metadata and derive issuer from the origin domain
        metadata = await self._get_oidc_metadata_cached(origin_domain)
        origin_issuer = metadata.get('issuer')
        self._oauth.metadata = metadata

        # Exchange the code for tokens
        # Use redirect_uri from transaction if available, otherwise fall back to default
        token_redirect_uri = transaction_data.redirect_uri or self._redirect_uri
        try:
            token_endpoint = self._oauth.metadata["token_endpoint"]
            token_response = await self._oauth.fetch_token(
                token_endpoint,
                code=code,
                code_verifier=transaction_data.code_verifier,
                redirect_uri=token_redirect_uri,
            )
        except OAuthError as e:
            # Raise a custom error (or handle it as appropriate)
            raise ApiError(
                "token_error", f"Token exchange failed: {str(e)}", e)

        # Use the userinfo field from the token_response for user claims
        user_info = token_response.get("userinfo")
        user_claims = None
        id_token = token_response.get("id_token")
        expected_org = transaction_data.organization

        if not user_info and not id_token and expected_org:
            raise OrganizationTokenValidationError(
                "Organization was requested but the token response included neither an ID token nor userinfo; "
                "cannot verify organization membership"
            )

        if user_info:
            if not isinstance(user_info, dict):
                if expected_org:
                    raise OrganizationTokenValidationError(
                        "Userinfo response is not a valid claims dictionary; cannot verify organization membership"
                    )
                raise ApiError(
                    "invalid_response",
                    "Userinfo response is not a valid claims dictionary"
                )
            if expected_org:
                validate_org_claims(user_info, expected_org)
            user_claims = UserClaims.parse_obj(user_info)
        elif id_token:
            # Fetch JWKS for signature verification
            jwks = await self._get_jwks_cached(origin_domain, metadata)

            # Decode and verify ID token with signature verification enabled
            try:
                claims = await self._verify_and_decode_jwt(
                    id_token, jwks, audience=self._client_id
                )

                # Custom normalized issuer validation
                token_issuer = claims.get("iss", "")
                if self._normalize_url(token_issuer) != self._normalize_url(origin_issuer):
                    raise IssuerValidationError("ID token issuer mismatch. Ensure your Auth0 domain is configured correctly.")

                # Organization claim validation — mandatory when org was requested.
                if expected_org:
                    validate_org_claims(claims, expected_org)

                user_claims = UserClaims.parse_obj(claims)
            except ValueError as e:
                raise ApiError("jwks_key_not_found", str(e))
            except jwt.InvalidSignatureError as e:
                raise ApiError(
                    "invalid_signature",
                    f"ID token signature verification failed. The token may have been tampered with or is from an untrusted source: {str(e)}",
                    e
                )
            except jwt.InvalidAudienceError as e:
                raise ApiError(
                    "invalid_audience",
                    f"ID token audience mismatch. Expected: {self._client_id}. Ensure your client_id is configured correctly: {str(e)}",
                    e
                )
            except jwt.ExpiredSignatureError as e:
                raise ApiError(
                    "token_expired",
                    f"ID token has expired: {str(e)}",
                    e
                )
            except jwt.InvalidTokenError as e:
                raise ApiError(
                    "invalid_token",
                    f"ID token verification failed: {str(e)}",
                    e
                )


        # Build a token set using the token response data
        token_set = TokenSet(
            audience=transaction_data.audience or self.DEFAULT_AUDIENCE_STATE_KEY,
            access_token=token_response.get("access_token", ""),
            scope=token_response.get("scope", ""),
            expires_at=int(time.time()) +
            token_response.get("expires_in", 3600)
        )

        # Generate a session id (sid) from token_response or transaction data, or create a new one
        sid = user_info.get(
            "sid") if user_info and "sid" in user_info else PKCE.generate_random_string(32)

        # Construct state data to represent the session
        state_data = StateData(
            user=user_claims,
            id_token=token_response.get("id_token"),
            # might be None if not provided
            refresh_token=token_response.get("refresh_token"),
            token_sets=[token_set],
            domain=origin_domain,
            internal={
                "sid": sid,
                "created_at": int(time.time())
            }
        )

        # Store the state data in the state store using store_options (Response required)
        await self._state_store.set(self._state_identifier, state_data, options=store_options)

        # Clean up transaction data after successful login
        await self._transaction_store.delete(transaction_identifier, options=store_options)

        result = {"state_data": state_data.dict()}
        if transaction_data.app_state:
            result["app_state"] = transaction_data.app_state

        # For RAR
        authorization_details = token_response.get("authorization_details")
        if authorization_details:
            result["authorization_details"] = authorization_details

        return result

    # ============================================================================
    # USER SESSION MANAGEMENT
    # Methods for retrieving user information, session data, and logout operations.
    # ============================================================================

    async def get_user(self, store_options: Optional[dict[str, Any]] = None) -> Optional[dict[str, Any]]:
        """
        Retrieves the user from the store, or None if no user found.

        Args:
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            The user, or None if no user found in the store.
        """
        state_data = await self._state_store.get(self._state_identifier, store_options)

        if state_data:
            # Domain check should work for both Pydantic models and plain dicts
            if hasattr(state_data, "dict") and callable(state_data.dict):
                state_data = state_data.dict()

            # In resolver mode, reject sessions with mismatched domain
            if self._domain_resolver:
                session_domain = self._get_session_domain(state_data)
                if not session_domain:
                    return None
                current_domain = await self._resolve_current_domain(store_options)
                if self._normalize_url(session_domain) != self._normalize_url(current_domain):
                    return None

            return state_data.get("user")
        return None

    async def get_session(self, store_options: Optional[dict[str, Any]] = None) -> Optional[dict[str, Any]]:
        """
        Retrieve the user session from the store, or None if no session found.

        Args:
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            The session, or None if no session found in the store.
        """
        state_data = await self._state_store.get(self._state_identifier, store_options)

        if state_data:
            # Domain check should work for both Pydantic models and plain dicts
            if hasattr(state_data, "dict") and callable(state_data.dict):
                state_data = state_data.dict()

            # In resolver mode, reject sessions with mismatched domain
            if self._domain_resolver:
                session_domain = self._get_session_domain(state_data)
                if not session_domain:
                    return None
                current_domain = await self._resolve_current_domain(store_options)
                if self._normalize_url(session_domain) != self._normalize_url(current_domain):
                    return None

            session_data = {k: v for k, v in state_data.items()
                            if k != "internal"}
            return session_data
        return None

    async def logout(
        self,
        options: Optional[LogoutOptions] = None,
        store_options: Optional[dict[str, Any]] = None
    ) -> str:
        options = options or LogoutOptions()

        if not self._domain_resolver:
            await self._state_store.delete(self._state_identifier, store_options)
            domain = self._domain
        else:
            # Resolver mode: delete session if domains match
            domain = await self._resolve_current_domain(store_options)
            state_data = await self._state_store.get(self._state_identifier, store_options)

            if state_data:
                if hasattr(state_data, "dict") and callable(state_data.dict):
                    state_data = state_data.dict()
                session_domain = self._get_session_domain(state_data)
                if session_domain and self._normalize_url(session_domain) == self._normalize_url(domain):
                    await self._state_store.delete(self._state_identifier, store_options)

        # Return logout URL for the current resolved domain
        logout_url = URL.create_logout_url(
            domain, self._client_id, options.return_to)

        return logout_url

    async def handle_backchannel_logout(
        self,
        logout_token: str,
        store_options: Optional[dict[str, Any]] = None
    ) -> None:
        """
        Handles backchannel logout requests.

        Args:
            logout_token: The logout token sent by Auth0
            store_options: Options to pass to the state store
        """
        if not logout_token:
            raise BackchannelLogoutError("Missing logout token")

        try:
            # Determine domain for JWKS fetch and issuer validation
            if self._domain_resolver:
                # Resolve domain from request context
                resolved_domain = await self._resolve_current_domain(store_options)

                # Read iss from unverified token for comparison
                try:
                    unverified = jwt.decode(
                        logout_token, algorithms=["RS256"],
                        options={"verify_signature": False}
                    )
                    token_issuer = unverified.get("iss", "")
                except Exception as e:
                    raise BackchannelLogoutError(
                        f"Failed to extract issuer from logout token: {str(e)}"
                    )

                if not token_issuer:
                    raise BackchannelLogoutError(
                        "Cannot determine domain: logout token has no valid issuer"
                    )

                # Validate token's iss matches the resolved domain
                normalized_iss = self._normalize_url(token_issuer)
                normalized_resolved = self._normalize_url(resolved_domain)
                if normalized_iss != normalized_resolved:
                    raise BackchannelLogoutError(
                        "Logout token issuer does not match the resolved domain"
                    )

                domain = resolved_domain
            else:
                domain = self._domain

            # Fetch JWKS and verify logout token
            jwks = await self._get_jwks_cached(domain)

            try:
                claims = await self._verify_and_decode_jwt(logout_token, jwks, audience=self._client_id)

                # Normalized issuer validation
                token_issuer = claims.get("iss", "")
                expected_issuer = self._normalize_url(domain)
                if self._normalize_url(token_issuer) != self._normalize_url(expected_issuer):
                    raise IssuerValidationError("Logout token issuer mismatch.Ensure your Auth0 domain is configured correctly."
                    )
            except ValueError as e:
                raise BackchannelLogoutError(str(e))
            except jwt.InvalidSignatureError as e:
                raise BackchannelLogoutError(
                    f"Logout token signature verification failed: {str(e)}"
                )
            except jwt.InvalidTokenError as e:
                raise BackchannelLogoutError(
                    f"Logout token verification failed: {str(e)}"
                )

            # Validate the token is a logout token
            events = claims.get("events", {})
            if "http://schemas.openid.net/event/backchannel-logout" not in events:
                raise BackchannelLogoutError(
                    "Invalid logout token: not a backchannel logout event")

            # Delete sessions associated with this token
            logout_claims = LogoutTokenClaims(
                sub=claims.get("sub"),
                sid=claims.get("sid"),
                iss=claims.get("iss")
            )

            await self._state_store.delete_by_logout_token(
                logout_claims.dict(), store_options
            )

        except (jwt.PyJWTError, ValidationError) as e:
            raise BackchannelLogoutError(
                f"Error processing logout token: {str(e)}")

    # ============================================================================
    # ACCESS TOKEN MANAGEMENT
    # Retrieves, validates, and refreshes access tokens for API calls.
    # ============================================================================

    async def get_access_token(
        self,
        store_options: Optional[dict[str, Any]] = None,
        audience: Optional[str] = None,
        scope: Optional[str] = None,
    ) -> str:
        """
        Retrieves the access token from the store, or calls Auth0 when the access token
        is expired and a refresh token is available in the store.
        Also updates the store when a new token was retrieved from Auth0.

        Args:
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            The access token, retrieved from the store or Auth0.

        Raises:
            AccessTokenError: If the token is expired and no refresh token is available.
        """
        state_data = await self._state_store.get(self._state_identifier, store_options)

        # Domain check should work for both Pydantic models and plain dicts
        if state_data and hasattr(state_data, "dict") and callable(state_data.dict):
            state_data_dict = state_data.dict()
        else:
            state_data_dict = state_data or {}

        # In resolver mode, reject sessions with mismatched domain
        if state_data and self._domain_resolver:
            session_domain = self._get_session_domain(state_data_dict)
            if not session_domain:
                raise AccessTokenError(
                    AccessTokenErrorCode.MISSING_SESSION_DOMAIN,
                    "Session domain does not match the current domain."
                )
            current_domain = await self._resolve_current_domain(store_options)
            if self._normalize_url(session_domain) != self._normalize_url(current_domain):
                raise AccessTokenError(
                    AccessTokenErrorCode.DOMAIN_MISMATCH,
                    "Session domain does not match the current domain."
                )

        auth_params = self._default_authorization_params or {}

        # Get audience passed in on options or use defaults
        if not audience:
            audience = auth_params.get("audience", None)

        merged_scope = self._merge_scope_with_defaults(scope, audience)

        # Find matching token set
        token_set = None
        if state_data_dict and "token_sets" in state_data_dict:
            token_set = self._find_matching_token_set(state_data_dict["token_sets"], audience, merged_scope)

        # If token is valid, return it
        if token_set and token_set.get("expires_at", 0) > time.time():
            return token_set["access_token"]

        # Check for refresh token
        if not state_data_dict or not state_data_dict.get("refresh_token"):
            raise AccessTokenError(
                AccessTokenErrorCode.MISSING_REFRESH_TOKEN,
                "The access token has expired and a refresh token was not provided. The user needs to re-authenticate."
            )

        # Get new token with refresh token
        try:
            # Use session's domain for token refresh
            session_domain = state_data_dict.get("domain") or self._domain
            get_refresh_token_options = {
                "refresh_token": state_data_dict["refresh_token"],
                "domain": session_domain
            }
            if audience:
                get_refresh_token_options["audience"] = audience

            if merged_scope:
                get_refresh_token_options["scope"] = merged_scope

            token_endpoint_response = await self.get_token_by_refresh_token(get_refresh_token_options)

            # Update state data with new token
            existing_state_data = await self._state_store.get(self._state_identifier, store_options)
            updated_state_data = State.update_state_data(
                audience, existing_state_data, token_endpoint_response)

            # Store updated state
            await self._state_store.set(self._state_identifier, updated_state_data, options=store_options)

            return token_endpoint_response["access_token"]
        except Exception as e:
            # Check for mfa_required error from token refresh
            if isinstance(e, ApiError) and e.code == "mfa_required":
                raw_mfa_token = getattr(e, "mfa_token", None)
                mfa_requirements = getattr(e, "mfa_requirements", None)

                if raw_mfa_token:
                    encrypted_token = self._mfa_client._encrypt_mfa_token(
                        raw_mfa_token=raw_mfa_token,
                        audience=audience or self.DEFAULT_AUDIENCE_STATE_KEY,
                        scope=merged_scope or "",
                        mfa_requirements=mfa_requirements
                    )
                    raise MfaRequiredError(
                        "Multifactor authentication required",
                        mfa_token=encrypted_token,
                        mfa_requirements=mfa_requirements
                    )

            if isinstance(e, AccessTokenError):
                raise
            raise AccessTokenError(
                AccessTokenErrorCode.REFRESH_TOKEN_ERROR,
                f"Failed to get token with refresh token: {str(e)}"
            )


    async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str, Any]:
        """
        Retrieves a token by exchanging a refresh token.

        Args:
            options: Dictionary containing the refresh token and any additional options.
                Must include a 'refresh_token' key.

        Raises:
            AccessTokenError: If there was an issue requesting the access token.

        Returns:
            A dictionary containing the token response from Auth0.
        """
        refresh_token = options.get("refresh_token")
        if not refresh_token:
            raise MissingRequiredArgumentError("refresh_token")

        try:
            # Use session domain if provided, otherwise fallback to static domain
            domain = options.get("domain") or self._domain

            # Fetch OIDC metadata from the correct domain
            metadata = await self._get_oidc_metadata_cached(domain)

            token_endpoint = metadata.get("token_endpoint")
            if not token_endpoint:
                raise ApiError("configuration_error",
                               "Token endpoint missing in OIDC metadata")

            # Prepare the token request parameters
            token_params = {
                "grant_type": "refresh_token",
                "refresh_token": refresh_token,
                "client_id": self._client_id,
            }

            audience = options.get("audience")
            if audience:
                token_params["audience"] = audience

            # Merge scope if present in options with any in the original authorization params
            merged_scope = self._merge_scope_with_defaults(
                request_scope=options.get("scope"),
                audience=audience
            )

            if merged_scope:
                token_params["scope"] = merged_scope

            # Exchange the refresh token for an access token
            async with self._get_http_client() as client:
                response = await client.post(
                    token_endpoint,
                    data=token_params,
                    auth=(self._client_id, self._client_secret)
                )

                if response.status_code != 200:
                    error_data = response.json()
                    error_code = error_data.get("error", "refresh_token_error")

                    # Preserve mfa_required details for upstream handling
                    if error_code == "mfa_required":
                        error = ApiError(
                            error_code,
                            error_data.get("error_description", "MFA required")
                        )
                        error.mfa_token = error_data.get("mfa_token")
                        mfa_requirements_data = error_data.get("mfa_requirements")
                        error.mfa_requirements = None
                        if mfa_requirements_data:
                            error.mfa_requirements = MfaRequirements(**mfa_requirements_data)
                        raise error

                    raise ApiError(
                        error_code,
                        error_data.get("error_description",
                                       "Failed to exchange refresh token")
                    )

                token_response = response.json()

                # Add required fields if they are missing
                if "expires_in" in token_response and "expires_at" not in token_response:
                    token_response["expires_at"] = int(
                        time.time()) + token_response["expires_in"]

                return token_response

        except Exception as e:
            if isinstance(e, ApiError):
                raise
            raise AccessTokenError(
                AccessTokenErrorCode.REFRESH_TOKEN_ERROR,
                "The access token has expired and there was an error while trying to refresh it.",
                e
            )

    def _merge_scope_with_defaults(
        self,
        request_scope: Optional[str],
        audience: Optional[str]
    ) -> Optional[str]:
        """Helper: Merges requested scopes with default authorization params."""
        audience = audience or self.DEFAULT_AUDIENCE_STATE_KEY
        default_scopes = ""
        if self._default_authorization_params and "scope" in self._default_authorization_params:
            auth_param_scope = self._default_authorization_params.get("scope")
            # For backwards compatibility, allow scope to be a single string
            # or dictionary by audience for MRRT
            if isinstance(auth_param_scope, dict) and audience in auth_param_scope:
                default_scopes = auth_param_scope[audience]
            elif isinstance(auth_param_scope, str):
                default_scopes = auth_param_scope

        default_scopes_list = default_scopes.split()
        request_scopes_list = (request_scope or "").split()

        merged_scopes = list(dict.fromkeys(default_scopes_list + request_scopes_list))
        return " ".join(merged_scopes) if merged_scopes else None

    def _find_matching_token_set(
        self,
        token_sets: list[dict[str, Any]],
        audience: Optional[str],
        scope: Optional[str]
    ) -> Optional[dict[str, Any]]:
        """Helper: Finds a token set matching the requested audience and scopes."""
        audience = audience or self.DEFAULT_AUDIENCE_STATE_KEY
        requested_scopes = set(scope.split()) if scope else set()
        matches: list[tuple[int, dict]] = []
        for token_set in token_sets:
            token_set_audience = token_set.get("audience")
            token_set_scopes = set(token_set.get("scope", "").split())
            if token_set_audience == audience and token_set_scopes == requested_scopes:
                # short-circuit if exact match
                return token_set
            if token_set_audience == audience and token_set_scopes.issuperset(requested_scopes):
                # consider stored tokens with more scopes than requested by number of scopes
                matches.append((len(token_set_scopes), token_set))

        # Return the token set with the smallest superset of scopes that matches the requested audience and scopes
        return min(matches, key=lambda t: t[0])[1] if matches else None

    # ============================================================================
    # BACKCHANNEL AUTHENTICATION (CIBA)
    # Client-Initiated Backchannel Authentication for decoupled authentication
    # flows where users authenticate on a separate device.
    # ============================================================================

    async def login_backchannel(
        self,
        options: dict[str, Any],
        store_options: Optional[dict[str, Any]] = None
    ) -> dict[str, Any]:
        """
        Logs in using Client-Initiated Backchannel Authentication.

        Note:
            Using Client-Initiated Backchannel Authentication requires the feature
            to be enabled in the Auth0 dashboard.

        See:
            https://auth0.com/docs/get-started/authentication-and-authorization-flow/client-initiated-backchannel-authentication-flow

        Args:
            options: Options used to configure the backchannel login process.
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            A dictionary containing the authorizationDetails (when RAR was used).
        """
        token_endpoint_response = await self.backchannel_authentication({
            "binding_message": options.get("binding_message"),
            "login_hint": options.get("login_hint"),
            "authorization_params": options.get("authorization_params"),
        }, store_options=store_options)

        existing_state_data = await self._state_store.get(self._state_identifier, store_options)

        audience = self._default_authorization_params.get(
            "audience", self.DEFAULT_AUDIENCE_STATE_KEY)

        state_data = State.update_state_data(
            audience,
            existing_state_data,
            token_endpoint_response
        )

        # Store domain for MCD session
        domain = await self._resolve_current_domain(store_options)
        state_data["domain"] = domain

        await self._state_store.set(self._state_identifier, state_data, store_options)

        result = {
            "authorization_details": token_endpoint_response.get("authorization_details")
        }
        return result

    async def backchannel_authentication(
        self,
        options: dict[str, Any],
        store_options: Optional[dict[str, Any]] = None
    ) -> dict[str, Any]:
        """
        Performs backchannel authentication with Auth0.

        This method starts a Client-Initiated Backchannel Authentication (CIBA) flow,
        which allows an application to request authentication from a user via a separate
        device or channel.

        Then polls the token endpoint until the user has authenticated or the request times out.

        Args:
            options (dict): Configuration options for backchannel authentication
                - login_hint (dict): Must contain a 'sub' field (e.g., {'sub': 'user_id'}).
                - binding_message (str, optional): Message to display to the user.
                - authorization_params (dict, optional): Additional authorization parameters.
                    - requested_expiry (int, optional): Requested expiry time in seconds, default is 30 secs.
                    - authorization_details (str, optional): JSON string for RAR.

        Returns:
            Token response data from the backchannel authentication

        Raises:
            ApiError: If the backchannel authentication fails
        """
        backchannel_data = await self.initiate_backchannel_authentication(options, store_options=store_options)
        auth_req_id = backchannel_data.get("auth_req_id")
        expires_in = backchannel_data.get(
            "expires_in", 120)  # Default to 2 minutes
        interval = backchannel_data.get(
            "interval", 5)  # Default to 5 seconds

        # Calculate when to stop polling
        end_time = time.time() + expires_in

        # Poll until we get a response or timeout
        while time.time() < end_time:
            # Make token request
            try:
                token_response = await self.backchannel_authentication_grant(
                    auth_req_id,
                    store_options=store_options,
                )
                return token_response

            except Exception as e:
                if isinstance(e, PollingApiError):
                    if e.code == "authorization_pending":
                        # Wait for the specified interval before polling again
                        await asyncio.sleep(interval)
                        continue
                    if e.code == "slow_down":
                        # Wait for the specified interval before polling again
                        await asyncio.sleep(e.interval or interval)
                        continue
                raise ApiError(
                    "backchannel_error",
                    f"Backchannel authentication failed: {str(e) or 'Unknown error'}",
                    e
                )

        # If we get here, we've timed out
        raise ApiError(
            "timeout", "Backchannel authentication timed out")

    async def initiate_backchannel_authentication(
            self,
            options: dict[str, Any],
            store_options: Optional[dict[str, Any]] = None
    ) -> dict[str, Any]:
        """
        Start backchannel authentication with Auth0.

        This method starts a Client-Initiated Backchannel Authentication (CIBA) flow,
        which allows an application to request authentication from a user via a separate
        device or channel.

        Args:
            options (dict): Configuration options for backchannel authentication
                - login_hint (dict): Must contain a 'sub' field (e.g., {'sub': 'user_id'}).
                - binding_message (str, optional): Message to display to the user.
                - authorization_params (dict, optional): Additional authorization parameters.
                    - requested_expiry (int, optional): Requested expiry time in seconds, default is 30 secs.
                    - authorization_details (str, optional): JSON string for RAR.

        Returns:
            dict: Response data from the bc-authorize backchannel authentication
                - auth_req_id (str): The authentication request ID.
                - expires_in (int): Time in seconds until the request expires.
                - interval (int, optional): Polling interval in seconds.

        Raises:
            ApiError: If the backchannel authentication fails

        See:
            https://auth0.com/docs/get-started/authentication-and-authorization-flow/client-initiated-backchannel-authentication-flow
        """

        sub = options.get('login_hint', {}).get("sub")
        if not sub:
            raise MissingRequiredArgumentError(
                "login_hint.sub"
            )

        authorization_params = options.get('authorization_params')
        if authorization_params is not None and not isinstance(authorization_params, dict):
            raise ApiError(
                "invalid_argument",
                "authorization_params must be a dict"
            )

        if authorization_params:
            requested_expiry = authorization_params.get("requested_expiry")
            if requested_expiry is not None:
                if not isinstance(requested_expiry, int) or requested_expiry <= 0:
                    raise ApiError(
                        "invalid_argument",
                        "authorization_params.requested_expiry must be a positive integer"
                    )

        try:
            # Resolve domain
            domain = await self._resolve_current_domain(store_options)
            metadata = await self._get_oidc_metadata_cached(domain)

            # Get the issuer from metadata
            issuer = metadata.get(
                "issuer") or f"https://{domain}/"

            # Get backchannel authentication endpoint
            backchannel_endpoint = metadata.get(
                "backchannel_authentication_endpoint")
            if not backchannel_endpoint:
                raise ApiError(
                    "configuration_error",
                    "Backchannel authentication is not supported by the authorization server"
                )

            # Prepare login hint in the required format
            login_hint = json.dumps({
                "format": "iss_sub",
                "iss": issuer,
                "sub": sub
            })

            # The Request Parameters
            params = {
                "client_id": self._client_id,
                "scope": "openid profile email",  # DEFAULT_SCOPES
                "login_hint": login_hint,
            }

            # Add binding message if provided
            if options.get('binding_message'):
                params["binding_message"] = options.get('binding_message')

            # Add any additional authorization parameters
            if self._default_authorization_params:
                params.update(self._default_authorization_params)

            if authorization_params:
                params.update(authorization_params)

            # Make the backchannel authentication request
            async with self._get_http_client() as client:
                backchannel_response = await client.post(
                    backchannel_endpoint,
                    data=params,
                    auth=(self._client_id, self._client_secret)
                )

                if backchannel_response.status_code != 200:
                    error_data = backchannel_response.json()
                    raise ApiError(
                        error_data.get("error", "backchannel_error"),
                        error_data.get(
                            "error_description", "Backchannel authentication request failed")
                    )

                backchannel_data = backchannel_response.json()
                auth_req_id = backchannel_data.get("auth_req_id")

                if not auth_req_id:
                    raise ApiError(
                        "invalid_response",
                        "Missing auth_req_id in backchannel authentication response"
                    )

                return backchannel_data

        except Exception as e:
            if isinstance(e, ApiError):
                raise
            raise ApiError(
                "backchannel_error",
                f"Backchannel authentication failed: {str(e) or 'Unknown error'}",
                e
            )

    async def backchannel_authentication_grant(
        self,
        auth_req_id: str,
        store_options: Optional[dict[str, Any]] = None
    ) -> dict[str, Any]:
        """
        Retrieves a token by exchanging an auth_req_id.

        Args:
            auth_req_id (str): The authentication request ID obtained from bc-authorize
            store_options: Optional options used to pass to the Transaction and State Store.

        Raises:
            AccessTokenError: If there was an issue requesting the access token.

        Returns:
            A dictionary containing the token response from Auth0.
        """
        if not auth_req_id:
            raise MissingRequiredArgumentError("auth_req_id")

        try:
            # Resolve domain (supports MCD mode)
            domain = await self._resolve_current_domain(store_options)
            metadata = await self._get_oidc_metadata_cached(domain)

            token_endpoint = metadata.get("token_endpoint")
            if not token_endpoint:
                raise ApiError("configuration_error",
                               "Token endpoint missing in OIDC metadata")

            # Prepare the token request parameters
            token_params = {
                "grant_type": "urn:openid:params:grant-type:ciba",
                "auth_req_id": auth_req_id,
                "client_id": self._client_id,
                "client_secret": self._client_secret
            }

            # Exchange the auth_req_id for an access token
            async with self._get_http_client() as client:
                response = await client.post(
                    token_endpoint,
                    data=token_params,
                    auth=(self._client_id, self._client_secret)
                )

                if response.status_code != 200:
                    error_data = response.json()
                    retry_after = response.headers.get("Retry-After")
                    interval = int(retry_after) if retry_after is not None else None
                    raise PollingApiError(
                        error_data.get("error", "auth_req_id_error"),
                        error_data.get("error_description",
                                       "Failed to exchange auth_req_id"),
                        interval
                    )

                try:
                    token_response = response.json()
                except json.JSONDecodeError:
                    raise ApiError(
                        "invalid_response",
                        "Failed to parse token response as JSON"
                    )

                # Add required fields if they are missing
                if "expires_in" in token_response and "expires_at" not in token_response:
                    token_response["expires_at"] = int(
                        time.time()) + token_response["expires_in"]

                return token_response

        except Exception as e:
            if isinstance(e, (ApiError, PollingApiError)):
                raise
            raise AccessTokenError(
                AccessTokenErrorCode.AUTH_REQ_ID_ERROR,
                "There was an error while trying to exchange the auth_req_id for an access token.",
                e
            )

    # ============================================================================
    # USER LINKING / UNLINKING
    # Methods for linking and unlinking external identity provider accounts
    # to a user's Auth0 profile.
    # ============================================================================

    async def start_link_user(
        self,
        options,
        store_options: Optional[dict[str, Any]] = None
    ):
        """
        Starts the user linking process, and returns a URL to redirect the user-agent to.

        Args:
            options: Options used to configure the user linking process.
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            URL to redirect the user to for authentication.
        """
        state_data = await self._state_store.get(self._state_identifier, store_options)

        if not state_data or not state_data.get("id_token"):
            raise StartLinkUserError(
                "Unable to start the user linking process without a logged in user. Ensure to login using the SDK before starting the user linking process."
            )

        # Resolve domain for MCD
        origin_domain = await self._resolve_current_domain(store_options)

        # In resolver mode, reject sessions with mismatched domain
        if self._domain_resolver:
            if hasattr(state_data, "dict") and callable(state_data.dict):
                state_data = state_data.dict()
            session_domain = self._get_session_domain(state_data)
            if not session_domain:
                raise StartLinkUserError(
                    "Session domain does not match the current domain."
                )
            if self._normalize_url(session_domain) != self._normalize_url(origin_domain):
                raise StartLinkUserError(
                    "Session domain does not match the current domain."
                )

        # Generate PKCE and state for security
        code_verifier = PKCE.generate_code_verifier()
        state = PKCE.generate_random_string(32)

        # Build the URL for user linking
        link_user_url = await self._build_link_user_url(
            connection=options.get("connection"),
            connection_scope=options.get("connectionScope"),
            id_token=state_data["id_token"],
            code_verifier=code_verifier,
            state=state,
            authorization_params=options.get("authorization_params"),
            domain=origin_domain
        )

        # Store transaction data
        transaction_data = TransactionData(
            code_verifier=code_verifier,
            app_state=options.get("app_state"),
            domain=origin_domain,
        )

        await self._transaction_store.set(
            f"{self._transaction_identifier}:{state}",
            transaction_data,
            options=store_options
        )

        return link_user_url

    async def complete_link_user(
        self,
        url: str,
        store_options: Optional[dict[str, Any]] = None
    ) -> dict[str, Any]:
        """
        Completes the user linking process.

        Args:
            url: The URL from which the query params should be extracted
            store_options: Optional options for the stores

        Returns:
            Dictionary containing the original app state
        """

        # We can reuse the interactive login completion since the flow is similar
        result = await self.complete_interactive_login(url, store_options)

        # Return just the app state as specified
        return {
            "app_state": result.get("app_state")
        }

    async def start_unlink_user(
        self,
        options,
        store_options: Optional[dict[str, Any]] = None
    ):
        """
        Starts the user unlinking process, and returns a URL to redirect the user-agent to.

        Args:
            options: Options used to configure the user unlinking process.
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            URL to redirect the user to for authentication.
        """
        state_data = await self._state_store.get(self._state_identifier, store_options)

        if not state_data or not state_data.get("id_token"):
            raise StartLinkUserError(
                "Unable to start the user linking process without a logged in user. Ensure to login using the SDK before starting the user linking process."
            )

        # Resolve domain for MCD
        origin_domain = await self._resolve_current_domain(store_options)

        # In resolver mode, reject sessions with mismatched domain
        if self._domain_resolver:
            if hasattr(state_data, "dict") and callable(state_data.dict):
                state_data = state_data.dict()
            session_domain = self._get_session_domain(state_data)
            if not session_domain:
                raise StartLinkUserError(
                    "Session domain does not match the current domain."
                )
            if self._normalize_url(session_domain) != self._normalize_url(origin_domain):
                raise StartLinkUserError(
                    "Session domain does not match the current domain."
                )

        # Generate PKCE and state for security
        code_verifier = PKCE.generate_code_verifier()
        state = PKCE.generate_random_string(32)

        # Build the URL for user linking
        link_user_url = await self._build_unlink_user_url(
            connection=options.get("connection"),
            id_token=state_data["id_token"],
            code_verifier=code_verifier,
            state=state,
            authorization_params=options.get("authorization_params"),
            domain=origin_domain
        )

        # Store transaction data
        transaction_data = TransactionData(
            code_verifier=code_verifier,
            app_state=options.get("app_state"),
            domain=origin_domain,
        )

        await self._transaction_store.set(
            f"{self._transaction_identifier}:{state}",
            transaction_data,
            options=store_options
        )

        return link_user_url

    async def complete_unlink_user(
        self,
        url: str,
        store_options: Optional[dict[str, Any]] = None
    ) -> dict[str, Any]:
        """
        Completes the user unlinking process.

        Args:
            url: The URL from which the query params should be extracted
            store_options: Optional options for the stores

        Returns:
            Dictionary containing the original app state
        """

        # We can reuse the interactive login completion since the flow is similar
        result = await self.complete_interactive_login(url, store_options)

        # Return just the app state as specified
        return {
            "app_state": result.get("app_state")
        }

    async def _build_link_user_url(
        self,
        connection: str,
        id_token: str,
        code_verifier: str,
        state: str,
        connection_scope: Optional[str] = None,
        authorization_params: Optional[dict[str, Any]] = None,
        domain: Optional[str] = None
    ) -> str:
        """Build a URL for linking user accounts"""
        # Generate code challenge from verifier
        code_challenge = PKCE.generate_code_challenge(code_verifier)

        # Use provided domain or fall back to static domain
        resolved_domain = domain or self._domain
        metadata = await self._get_oidc_metadata_cached(resolved_domain)

        # Get authorization endpoint
        auth_endpoint = metadata.get("authorization_endpoint",
                                     f"https://{resolved_domain}/authorize")

        # Build params
        params = {
            "client_id": self._client_id,
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
            "state": state,
            "requested_connection": connection,
            "requested_connection_scope": connection_scope,
            "response_type": "code",
            "id_token_hint": id_token,
            "scope": "openid link_account",
            "audience": "my-account"
        }

        # Add connection scope if provided
        if connection_scope:
            params["requested_connection_scope"] = connection_scope

        # Add any additional parameters
        if authorization_params:
            params.update(authorization_params)
        return URL.build_url(auth_endpoint, params)

    async def _build_unlink_user_url(
        self,
        connection: str,
        id_token: str,
        code_verifier: str,
        state: str,
        authorization_params: Optional[dict[str, Any]] = None,
        domain: Optional[str] = None
    ) -> str:
        """Build a URL for unlinking user accounts"""
        # Generate code challenge from verifier
        code_challenge = PKCE.generate_code_challenge(code_verifier)

        # Use provided domain or fall back to static domain
        resolved_domain = domain or self._domain
        metadata = await self._get_oidc_metadata_cached(resolved_domain)

        # Get authorization endpoint
        auth_endpoint = metadata.get("authorization_endpoint",
                                     f"https://{resolved_domain}/authorize")

        # Build params
        params = {
            "client_id": self._client_id,
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
            "state": state,
            "requested_connection": connection,
            "response_type": "code",
            "id_token_hint": id_token,
            "scope": "openid unlink_account",
            "audience": "my-account"
        }
        # Add any additional parameters
        if authorization_params:
            params.update(authorization_params)

        return URL.build_url(auth_endpoint, params)

    # ============================================================================
    # FEDERATED CONNECTION TOKENS
    # Retrieves access tokens for federated identity provider connections
    # (e.g., Google, GitHub) using token exchange.
    # ============================================================================

    async def get_access_token_for_connection(
        self,
        options: dict[str, Any],
        store_options: Optional[dict[str, Any]] = None
    ) -> str:
        """
        Retrieves an access token for a connection.

        This method attempts to obtain an access token for a specified connection.
        It first checks if a refresh token exists in the store.
        If no refresh token is found, it throws an `AccessTokenForConnectionError` indicating
        that the refresh token was not found.

        Args:
            options: Options for retrieving an access token for a connection.
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            The access token for the connection

        Raises:
            AccessTokenForConnectionError: If the access token was not found or
                there was an issue requesting the access token.
        """
        state_data = await self._state_store.get(self._state_identifier, store_options)

        if state_data and hasattr(state_data, "dict") and callable(state_data.dict):
            state_data_dict = state_data.dict()
        else:
            state_data_dict = state_data or {}

        # In resolver mode, reject sessions with mismatched domain
        if self._domain_resolver:
            session_domain = self._get_session_domain(state_data_dict)
            if not session_domain:
                raise AccessTokenForConnectionError(
                    AccessTokenForConnectionErrorCode.MISSING_SESSION_DOMAIN,
                    "Session domain does not match the current domain."
                )
            current_domain = await self._resolve_current_domain(store_options)
            if self._normalize_url(session_domain) != self._normalize_url(current_domain):
                raise AccessTokenForConnectionError(
                    AccessTokenForConnectionErrorCode.DOMAIN_MISMATCH,
                    "Session domain does not match the current domain."
                )

        # Find existing connection token
        connection_token_set = None
        if state_data_dict and len(state_data_dict["connection_token_sets"]) > 0:
            for ts in state_data_dict.get("connection_token_sets"):
                if ts.get("connection") == options["connection"]:
                    connection_token_set = ts
                    break

        # If token is valid, return it
        if connection_token_set and connection_token_set.get("expires_at", 0) > time.time():
            return connection_token_set["access_token"]

        # Check for refresh token
        if not state_data_dict or not state_data_dict.get("refresh_token"):
            raise AccessTokenForConnectionError(
                AccessTokenForConnectionErrorCode.MISSING_REFRESH_TOKEN,
                "A refresh token was not found but is required to be able to retrieve an access token for a connection."
            )
        # Get new token for connection
        # Use session's domain for token exchange
        session_domain = state_data_dict.get("domain") or self._domain
        token_endpoint_response = await self.get_token_for_connection({
            "connection": options.get("connection"),
            "login_hint": options.get("login_hint"),
            "refresh_token": state_data_dict["refresh_token"],
            "domain": session_domain
        })

        # Update state data with new token
        updated_state_data = State.update_state_data_for_connection_token_set(
            options, state_data_dict, token_endpoint_response)

        # Store updated state
        await self._state_store.set(self._state_identifier, updated_state_data, store_options)

        return token_endpoint_response["access_token"]

    async def get_token_for_connection(self, options: dict[str, Any]) -> dict[str, Any]:
        """
        Retrieves a token for a connection.

        Args:
            options: Options for retrieving an access token for a connection.
                Must include 'connection' and 'refresh_token' keys.
                May optionally include 'login_hint'.

        Raises:
            AccessTokenForConnectionError: If there was an issue requesting the access token.

        Returns:
            Dictionary containing the token response with accessToken, expiresAt, and scope.
        """
        # Constants
        SUBJECT_TYPE_REFRESH_TOKEN = "urn:ietf:params:oauth:token-type:refresh_token"
        REQUESTED_TOKEN_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN = "http://auth0.com/oauth/token-type/federated-connection-access-token"
        GRANT_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN = "urn:auth0:params:oauth:grant-type:token-exchange:federated-connection-access-token"
        try:
            # Use session domain if provided, otherwise fallback to static domain
            domain = options.get("domain") or self._domain

            # Fetch OIDC metadata from the correct domain
            metadata = await self._get_oidc_metadata_cached(domain)

            token_endpoint = metadata.get("token_endpoint")
            if not token_endpoint:
                raise ApiError("configuration_error",
                               "Token endpoint missing in OIDC metadata")

            # Prepare parameters
            params = {
                "connection": options["connection"],
                "subject_token_type": SUBJECT_TYPE_REFRESH_TOKEN,
                "subject_token": options["refresh_token"],
                "requested_token_type": REQUESTED_TOKEN_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN,
                "grant_type": GRANT_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN,
                "client_id": self._client_id
            }

            # Add login_hint if provided
            if "login_hint" in options and options["login_hint"]:
                params["login_hint"] = options["login_hint"]

            # Make the request
            async with self._get_http_client() as client:
                response = await client.post(
                    token_endpoint,
                    data=params,
                    auth=(self._client_id, self._client_secret)
                )

                if response.status_code != 200:
                    error_data = response.json() if response.headers.get(
                        "content-type") == "application/json" else {}
                    raise ApiError(
                        error_data.get("error", "connection_token_error"),
                        error_data.get(
                            "error_description", f"Failed to get token for connection: {response.status_code}")
                    )

                token_endpoint_response = response.json()

                return {
                    "access_token": token_endpoint_response.get("access_token"),
                    "expires_at": int(time.time()) + int(token_endpoint_response.get("expires_in", 3600)),
                    "scope": token_endpoint_response.get("scope", "")
                }

        except Exception as e:
            if isinstance(e, ApiError):
                raise AccessTokenForConnectionError(
                    AccessTokenForConnectionErrorCode.API_ERROR,
                    str(e)
                )
            raise AccessTokenForConnectionError(
                AccessTokenForConnectionErrorCode.FETCH_ERROR,
                "There was an error while trying to retrieve an access token for a connection.",
                e
            )

    # ============================================================================
    # CONNECTED ACCOUNTS
    # Methods for managing third-party account connections via the My Account API.
    # Includes initiating connections, completing flows, and CRUD operations.
    # ============================================================================

    async def start_connect_account(
        self,
        options: ConnectAccountOptions,
        store_options: dict = None
    ) -> str:
        """
        Initiates the connect account flow for linking a third-party account to the user's profile.

        This method generates PKCE parameters, creates a transaction and calls the My Account API
        to create a connect account request, returning /connect url containing a ticket.

        Args:
            options: Options for retrieving an access token for a connection.
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            The a connect URL containing a ticket to redirect the user to.
        """
        # Use the default redirect_uri if none is specified
        redirect_uri = options.redirect_uri or self._redirect_uri
        # Ensure we have a redirect_uri
        if not redirect_uri:
            raise MissingRequiredArgumentError("redirect_uri")

        # Generate PKCE code verifier and challenge
        code_verifier = PKCE.generate_code_verifier()
        code_challenge = PKCE.generate_code_challenge(code_verifier)

        state= PKCE.generate_random_string(32)

        connect_request = ConnectAccountRequest(
            connection=options.connection,
            scopes=options.scopes,
            redirect_uri = redirect_uri,
            code_challenge=code_challenge,
            code_challenge_method="S256",
            state=state,
            authorization_params=options.authorization_params
        )

        access_token = await self.get_access_token(
            audience=self._my_account_client.audience,
            scope="create:me:connected_accounts",
            store_options=store_options
        )
        connect_response = await self._my_account_client.connect_account(
            access_token=access_token,
            request=connect_request
        )

        # Build the transaction data to store
        transaction_data = TransactionData(
            code_verifier=code_verifier,
            app_state=options.app_state,
            auth_session=connect_response.auth_session,
            redirect_uri=redirect_uri
        )

        # Store the transaction data
        await self._transaction_store.set(
            f"{self._transaction_identifier}:{state}",
            transaction_data,
            options=store_options
        )

        parsedUrl = urlparse(connect_response.connect_uri)
        query = urlencode({"ticket": connect_response.connect_params.ticket})
        return urlunparse((parsedUrl.scheme, parsedUrl.netloc, parsedUrl.path, parsedUrl.params, query, parsedUrl.fragment))

    async def complete_connect_account(
        self,
        url: str,
        store_options: dict = None
    ) -> CompleteConnectAccountResponse:
        """
        Handles the redirect callback to complete the connect account flow for linking a third-party
        account to the user's profile.

        This works similar to the redirect from the login flow except it verifies the `connect_code`
        with the My Account API rather than the `code` with the Authorization Server.

        Args:
            url: The full callback URL including query parameters
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            A response from the connect account flow.
        """
        # Parse the URL to get query parameters
        parsed_url = urlparse(url)
        query_params = parse_qs(parsed_url.query)

        # Get state parameter from the URL
        state = query_params.get("state", [""])[0]
        if not state:
            raise MissingRequiredArgumentError("state")

        # Get the authorization code from the URL
        connect_code = query_params.get("connect_code", [""])[0]
        if not connect_code:
            raise MissingRequiredArgumentError("connect_code")

        # Retrieve the transaction data using the state
        transaction_identifier = f"{self._transaction_identifier}:{state}"
        transaction_data = await self._transaction_store.get(transaction_identifier, options=store_options)

        if not transaction_data:
            raise MissingTransactionError()

        access_token = await self.get_access_token(
            audience=self._my_account_client.audience,
            scope="create:me:connected_accounts",
            store_options=store_options
        )

        request = CompleteConnectAccountRequest(
            auth_session=transaction_data.auth_session,
            connect_code=connect_code,
            redirect_uri=transaction_data.redirect_uri,
            code_verifier=transaction_data.code_verifier
        )
        try:
            response = await self._my_account_client.complete_connect_account(
                access_token=access_token, request=request)
            if transaction_data.app_state is not None:
                response.app_state = transaction_data.app_state
        finally:
            # Clean up transaction data
            await self._transaction_store.delete(transaction_identifier, options=store_options)

        return response

    async def list_connected_accounts(
        self,
        connection: Optional[str] = None,
        from_param: Optional[str] = None,
        take: Optional[int] = None,
        store_options: dict = None
    ) -> ListConnectedAccountsResponse:
        """
        Retrieves a list of connected accounts for the authenticated user.

        Args:
            connection (Optional[str], optional): Filter results to a specific connection. Defaults to None.
            from_param (Optional[str], optional): A pagination token indicating where to start retrieving results, obtained from a prior request. Defaults to None.
            take (Optional[int], optional): The maximum number of connections to retrieve. Defaults to None.
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            ListConnectedAccountsResponse: The response object containing the list of connected accounts.

        Raises:
            Auth0Error: If there is an error retrieving the access token.
            MyAccountApiError: If the My Account API returns an error response.
        """
        if take is not None and (not isinstance(take, int) or take < 1):
            raise InvalidArgumentError("take", "The 'take' parameter must be a positive integer.")

        access_token = await self.get_access_token(
            audience=self._my_account_client.audience,
            scope="read:me:connected_accounts",
            store_options=store_options
        )
        return await self._my_account_client.list_connected_accounts(
            access_token=access_token, connection=connection, from_param=from_param, take=take)

    async def delete_connected_account(
        self,
        connected_account_id: str,
        store_options: dict = None
    ) -> None:
        """
        Deletes a connected account.

        Args:
            connected_account_id (str): The ID of the connected account to delete.
            store_options: Optional options used to pass to the Transaction and State Store.

        Raises:
            Auth0Error: If there is an error retrieving the access token.
            MyAccountApiError: If the My Account API returns an error response.
        """
        if not connected_account_id:
            raise MissingRequiredArgumentError("connected_account_id")

        access_token = await self.get_access_token(
            audience=self._my_account_client.audience,
            scope="delete:me:connected_accounts",
            store_options=store_options
        )
        await self._my_account_client.delete_connected_account(
            access_token=access_token, connected_account_id=connected_account_id)

    async def list_connected_account_connections(
        self,
        from_param: Optional[str] = None,
        take: Optional[int] = None,
        store_options: dict = None
    ) -> ListConnectedAccountConnectionsResponse:
        """
        Retrieves a list of available connections that can be used connected accounts for the authenticated user.

        Args:
            from_param (Optional[str], optional): A pagination token indicating where to start retrieving results, obtained from a prior request. Defaults to None.
            take (Optional[int], optional): The maximum number of connections to retrieve. Defaults to None.
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            ListConnectedAccountConnectionsResponse: The response object containing the list of connected account connections.

        Raises:
            Auth0Error: If there is an error retrieving the access token.
            MyAccountApiError: If the My Account API returns an error response.
        """
        if take is not None and (not isinstance(take, int) or take < 1):
            raise InvalidArgumentError("take", "The 'take' parameter must be a positive integer.")

        access_token = await self.get_access_token(
            audience=self._my_account_client.audience,
            scope="read:me:connected_accounts",
            store_options=store_options
        )
        return await self._my_account_client.list_connected_account_connections(
            access_token=access_token, from_param=from_param, take=take)

    # ============================================================================
    # CUSTOM TOKEN EXCHANGE (RFC 8693)
    # Exchanges external custom tokens for Auth0 tokens.
    # ============================================================================

    async def custom_token_exchange(
        self,
        options: CustomTokenExchangeOptions,
        store_options: Optional[dict[str, Any]] = None
    ) -> TokenExchangeResponse:
        """
        Exchanges a custom token for Auth0 tokens using RFC 8693.

        This method implements the OAuth 2.0 Token Exchange specification,
        allowing you to exchange external custom tokens for Auth0 access tokens.

        Args:
            options: Configuration for the token exchange
            store_options: Optional options used to pass to the Transaction and State Store.

        Returns:
            TokenExchangeResponse containing access_token and metadata

        Raises:
            CustomTokenExchangeError: If token exchange fails
            MissingRequiredArgumentError: If required parameters are missing

        Example:
            ```python
            response = await client.custom_token_exchange(
                CustomTokenExchangeOptions(
                    subject_token="custom-token-value",
                    subject_token_type="urn:acme:mcp-token",
                    audience="https://api.example.com",
                    scope="read:data write:data"
                )
            )
            print(response.access_token)
            ```

        See:
            https://datatracker.ietf.org/doc/html/rfc8693
        """
        try:
            if not isinstance(options, CustomTokenExchangeOptions):
                options = CustomTokenExchangeOptions(**options)

            if not options.subject_token.strip():
                raise CustomTokenExchangeError(
                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
                    "subject_token cannot be empty or whitespace-only"
                )
            if not options.subject_token_type.strip():
                raise CustomTokenExchangeError(
                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
                    "subject_token_type cannot be empty or whitespace-only"
                )
            if options.subject_token.strip().startswith("Bearer "):
                raise CustomTokenExchangeError(
                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
                    "subject_token should not include 'Bearer ' prefix"
                )
            if options.actor_token is not None and not options.actor_token.strip():
                raise CustomTokenExchangeError(
                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
                    "actor_token cannot be empty or whitespace-only"
                )
            if options.actor_token and options.actor_token.strip().startswith("Bearer "):
                raise CustomTokenExchangeError(
                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
                    "actor_token should not include 'Bearer ' prefix"
                )
            if options.actor_token and not options.actor_token_type:
                raise CustomTokenExchangeError(
                    CustomTokenExchangeErrorCode.MISSING_ACTOR_TOKEN_TYPE,
                    "actor_token_type is required when actor_token is provided"
                )
            if options.actor_token_type and not options.actor_token:
                raise CustomTokenExchangeError(
                    CustomTokenExchangeErrorCode.MISSING_ACTOR_TOKEN,
                    "actor_token is required when actor_token_type is provided"
                )

            # Resolve domain
            domain = await self._resolve_current_domain(store_options)
            metadata = await self._get_oidc_metadata_cached(domain)

            token_endpoint = metadata.get("token_endpoint")
            if not token_endpoint:
                raise ApiError("configuration_error", "Token endpoint missing in OIDC metadata")

            # Prepare token exchange parameters per RFC 8693
            params = {
                "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
                "client_id": self._client_id,
                "subject_token": options.subject_token,
                "subject_token_type": options.subject_token_type,
            }

            # Add optional parameters
            if options.audience:
                params["audience"] = options.audience

            if options.scope:
                params["scope"] = options.scope

            if options.actor_token:
                params["actor_token"] = options.actor_token
                params["actor_token_type"] = options.actor_token_type

            if options.organization:
                params["organization"] = options.organization

            # Merge additional authorization params
            if options.authorization_params:
                # Prevent override of critical parameters
                forbidden_params = {"grant_type", "client_id", "subject_token", "subject_token_type"}
                for key, value in options.authorization_params.items():
                    if key not in forbidden_params:
                        params[key] = value

            # Make the token exchange request
            async with self._get_http_client() as client:
                response = await client.post(
                    token_endpoint,
                    data=params,
                    auth=(self._client_id, self._client_secret)
                )

                if response.status_code != 200:
                    error_data = response.json() if response.headers.get(
                        "content-type", "").startswith("application/json") else {}
                    raise CustomTokenExchangeError(
                        error_data.get("error", CustomTokenExchangeErrorCode.TOKEN_EXCHANGE_FAILED),
                        error_data.get("error_description", f"Token exchange failed: {response.status_code}")
                    )

                try:
                    token_data = response.json()
                except json.JSONDecodeError:
                    raise CustomTokenExchangeError(
                        CustomTokenExchangeErrorCode.INVALID_RESPONSE,
                        "Failed to parse token response as JSON"
                    )

                token_response = TokenExchangeResponse(**token_data)

                # Surface the actor claim for delegation exchanges. Best-effort:
                # a decode/verify hiccup must not fail an exchange the token
                # endpoint already accepted, so act stays None on any failure.
                if options.actor_token and token_response.id_token:
                    try:
                        jwks = await self._get_jwks_cached(domain, metadata)
                        claims = await self._verify_and_decode_jwt(
                            token_response.id_token, jwks, audience=self._client_id
                        )
                        # Apply the same normalized issuer check the login path uses
                        # before trusting any claim from the token.
                        if self._normalize_url(claims.get("iss", "")) == self._normalize_url(metadata.get("issuer")):
                            token_response.act = claims.get("act")
                    except Exception:
                        token_response.act = None

                return token_response

        except ValidationError as e:
            raise CustomTokenExchangeError(
                CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
                f"Token validation failed: {str(e)}"
            )
        except Exception as e:
            if isinstance(e, (CustomTokenExchangeError, ApiError)):
                raise
            raise CustomTokenExchangeError(
                CustomTokenExchangeErrorCode.TOKEN_EXCHANGE_FAILED,
                f"Token exchange failed: {str(e)}",
                e
            )

    async def login_with_custom_token_exchange(
        self,
        options: LoginWithCustomTokenExchangeOptions,
        store_options: Optional[dict[str, Any]] = None
    ) -> LoginWithCustomTokenExchangeResult:
        """
        Performs token exchange and establishes a user session.

        This method combines custom_token_exchange() with session management,
        exchanging a custom token for Auth0 tokens and storing the session state.

        Args:
            options: Configuration for token exchange and login
            store_options: Optional options for state store (e.g., request/response for cookies)

        Returns:
            LoginWithCustomTokenExchangeResult containing session state

        Raises:
            CustomTokenExchangeError: If token exchange fails
            ApiError: If session management fails

        Example:
            ```python
            result = await client.login_with_custom_token_exchange(
                LoginWithCustomTokenExchangeOptions(
                    subject_token="custom-token-value",
                    subject_token_type="urn:acme:mcp-token",
                    audience="https://api.example.com"
                ),
                store_options={"request": request, "response": response}
            )
            print(result.state_data["user"])
            ```

        See:
            https://datatracker.ietf.org/doc/html/rfc8693
        """
        try:
            # Perform token exchange
            exchange_options = CustomTokenExchangeOptions(
                subject_token=options.subject_token,
                subject_token_type=options.subject_token_type,
                audience=options.audience,
                scope=options.scope,
                actor_token=options.actor_token,
                actor_token_type=options.actor_token_type,
                organization=options.organization,
                authorization_params=options.authorization_params
            )

            token_response = await self.custom_token_exchange(exchange_options, store_options=store_options)

            # Resolve domain and fetch metadata for verification
            domain = await self._resolve_current_domain(store_options)
            metadata = await self._get_oidc_metadata_cached(domain)

            # Extract user claims from ID token if present
            user_claims = None
            sid = PKCE.generate_random_string(32)  # Default sid

            if token_response.id_token:
                # Fetch JWKS and verify ID token signature
                jwks = await self._get_jwks_cached(domain, metadata)
                try:
                    claims = await self._verify_and_decode_jwt(
                        token_response.id_token, jwks, audience=self._client_id
                    )

                    # Validate issuer against resolved domain
                    origin_issuer = metadata.get("issuer")
                    token_issuer = claims.get("iss", "")
                    if self._normalize_url(token_issuer) != self._normalize_url(origin_issuer):
                        raise IssuerValidationError(
                            "ID token issuer mismatch. Ensure your Auth0 domain is configured correctly."
                        )

                    # UserClaims allows extra fields, so any act claim in the
                    # verified id_token is carried onto the session user here.
                    user_claims = UserClaims.parse_obj(claims)
                    # Extract sid from token if available
                    sid = claims.get("sid", sid)
                except ValueError as e:
                    raise ApiError("jwks_key_not_found", str(e))
                except jwt.InvalidSignatureError as e:
                    raise ApiError(
                        "invalid_signature",
                        f"ID token signature verification failed: {str(e)}",
                        e
                    )
                except jwt.InvalidAudienceError as e:
                    raise ApiError(
                        "invalid_audience",
                        f"ID token audience mismatch. Expected: {self._client_id}: {str(e)}",
                        e
                    )
                except jwt.ExpiredSignatureError as e:
                    raise ApiError(
                        "token_expired",
                        f"ID token has expired: {str(e)}",
                        e
                    )
                except jwt.InvalidTokenError as e:
                    raise ApiError(
                        "invalid_token",
                        f"ID token verification failed: {str(e)}",
                        e
                    )

            # Determine audience for token set
            audience = options.audience or self.DEFAULT_AUDIENCE_STATE_KEY

            # Build token set
            token_set = TokenSet(
                audience=audience,
                access_token=token_response.access_token,
                scope=token_response.scope or options.scope or "",
                expires_at=int(time.time()) + token_response.expires_in
            )

            # Construct state data
            state_data = StateData(
                user=user_claims,
                id_token=token_response.id_token,
                refresh_token=token_response.refresh_token,
                token_sets=[token_set],
                domain=domain,
                internal={
                    "sid": sid,
                    "created_at": int(time.time())
                }
            )

            # Store session
            await self._state_store.set(self._state_identifier, state_data, options=store_options)

            # Build result
            result = LoginWithCustomTokenExchangeResult(
                state_data=state_data.dict()
            )

            return result

        except Exception as e:
            if isinstance(e, (CustomTokenExchangeError, ApiError, IssuerValidationError)):
                raise
            raise CustomTokenExchangeError(
                CustomTokenExchangeErrorCode.TOKEN_EXCHANGE_FAILED,
                f"Login with custom token exchange failed: {str(e)}",
                e
            )

    # ============================================================================
    # MFA (Multi-Factor Authentication)
    # ============================================================================

    @property
    def mfa(self) -> MfaClient:
        """Access the MFA client for multi-factor authentication operations."""
        return self._mfa_client