poc: add get_token_on_behalf_of() to auth0-api-python

EXAMPLES.md

@@ -2,6 +2,64 @@
 
 This document provides examples for using the `auth0-api-python` package to validate Auth0 tokens in your API.
 
+## On Behalf Of Token Exchange
+
+Use `get_token_on_behalf_of()` when your API receives an `Auth0` access token for itself and needs
+to exchange it for another `Auth0` access token targeting a downstream API while preserving the same
+user identity. This is especially useful for `MCP` servers and other intermediary APIs that need to
+call downstream APIs on behalf of the user.
+
+The following example verifies the incoming access token for your API, exchanges it for a token for the downstream API, and then calls the downstream API with the exchanged token.
+
+```python
+import asyncio
+import httpx
+
+from auth0_api_python import ApiClient, ApiClientOptions
+
+async def exchange_on_behalf_of():
+    api_client = ApiClient(ApiClientOptions(
+        domain="your-tenant.auth0.com",
+        audience="https://mcp-server.example.com",
+        client_id="<AUTH0_CLIENT_ID>",
+        client_secret="<AUTH0_CLIENT_SECRET>"
+    ))
+
+    incoming_access_token = "incoming-auth0-access-token"
+
+    claims = await api_client.verify_access_token(access_token=incoming_access_token)
+
+    result = await api_client.get_token_on_behalf_of(
+        access_token=incoming_access_token,
+        audience="https://calendar-api.example.com",
+        scope="calendar:read calendar:write"
+    )
+
+    async with httpx.AsyncClient() as client:
+        downstream_response = await client.get(
+            "https://calendar-api.example.com/events",
+            headers={"Authorization": f"Bearer {result.access_token}"}
+        )
+
+    downstream_response.raise_for_status()
+
+    return {
+        "user": claims["sub"],
+        "data": downstream_response.json(),
+    }
+
+asyncio.run(exchange_on_behalf_of())
+```
+
+> [!TIP] Production notes:
+> - Pass the raw access token to `get_token_on_behalf_of()`. Do not pass the full `Authorization` header or include the `Bearer ` prefix.
+> - Verify the incoming token for your API before exchanging it so your application rejects invalid or mis-targeted tokens early.
+> - The downstream `audience` must match an API identifier configured in your Auth0 tenant.
+> - `get_token_on_behalf_of()` only returns access-token-oriented fields. It does not expose `id_token` or `refresh_token`.
+
+In the current implementation, `get_token_on_behalf_of()` forwards the incoming access token as
+the [RFC 8693](https://datatracker.ietf.org/doc/html/rfc8693#section-2.1) `subject_token` and relies on Auth0 to handle any DPoP-specific behavior for that token.
+
 ## Bearer Authentication
 
 Bearer authentication is the standard OAuth 2.0 token authentication method.
@@ -157,4 +215,4 @@ async def verify_dpop_token(access_token, dpop_proof, http_method, http_url):
         "token_claims": token_claims,
         "proof_claims": proof_claims
     }
-```
+```

README.md

@@ -212,6 +212,44 @@ except ApiError as e:
 
 More info: https://auth0.com/docs/authenticate/custom-token-exchange
 
+#### On Behalf Of Token Exchange
+
+Use `get_token_on_behalf_of()` when your API receives an `Auth0` access token for itself and needs
+to exchange it for another `Auth0` access token targeting a downstream API while preserving the
+same user identity. This is especially useful for `MCP` servers and other intermediary APIs that
+need to call downstream APIs on behalf of the user.
+
+The following example verifies the incoming access token for your API, exchanges it for a token for the downstream API, and then calls the downstream API with the exchanged token.
+
+```python
+import httpx
+
+async def handle_calendar_request(incoming_access_token: str):
+    await api_client.verify_access_token(access_token=incoming_access_token)
+
+    result = await api_client.get_token_on_behalf_of(
+        access_token=incoming_access_token,
+        audience="https://calendar-api.example.com",
+        scope="calendar:read calendar:write"
+    )
+
+    async with httpx.AsyncClient() as client:
+        downstream_response = await client.get(
+            "https://calendar-api.example.com/events",
+            headers={"Authorization": f"Bearer {result.access_token}"}
+        )
+
+    downstream_response.raise_for_status()
+
+    return downstream_response.json()
+```
+
+The OBO wrapper reuses the existing RFC 8693 exchange support and fixes both token-type parameters
+to Auth0 access-token exchange. In the current implementation, the SDK forwards the incoming access
+token as the `subject_token` and relies on Auth0 to handle any DPoP-specific behavior for that token.
+The OBO result only includes access-token-oriented fields. It does not expose `id_token` or
+`refresh_token`.
+
 #### Requiring Additional Claims
 
 If your application demands extra claims, specify them with `required_claims`:
@@ -353,4 +391,4 @@ Please do not report security vulnerabilities on the public GitHub issue tracker
 </p>
 <p align="center">
   This project is licensed under the MIT license. See the <a href="https://github.com/auth0/auth0-api-python/LICENSE"> LICENSE</a> file for more info.
-</p>
+</p>

src/auth0_api_python/__init__.py

@@ -14,7 +14,7 @@
     DomainsResolverError,
     GetTokenByExchangeProfileError,
 )
-from .types import DomainsResolver, DomainsResolverContext
+from .types import DomainsResolver, DomainsResolverContext, OnBehalfOfTokenResult
 
 __all__ = [
     "ApiClient",
@@ -27,4 +27,5 @@
     "DomainsResolverError",
     "GetTokenByExchangeProfileError",
     "InMemoryCache",
+    "OnBehalfOfTokenResult",
 ]

src/auth0_api_python/api_client.py

@@ -21,6 +21,7 @@
     MissingRequiredArgumentError,
     VerifyAccessTokenError,
 )
+from .types import OnBehalfOfTokenResult
 from .utils import (
     calculate_jwk_thumbprint,
     fetch_jwks,
@@ -34,6 +35,7 @@
 
 # Token Exchange constants
 TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"  # noqa: S105
+OBO_ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"  # noqa: S105
 MAX_ARRAY_VALUES_PER_KEY = 20  # DoS protection for extra parameter arrays
 
 # OAuth parameter denylist - parameters that cannot be overridden via extras
@@ -962,6 +964,64 @@ async def example():
                 exc
             )
 
+    async def get_token_on_behalf_of(
+        self,
+        access_token: str,
+        audience: str,
+        scope: Optional[str] = None,
+    ) -> OnBehalfOfTokenResult:
+        """
+        Exchange an Auth0 access token for another Auth0 access token targeting a downstream API
+        while acting on behalf of the same end user (OBO).
+
+        This is a convenience wrapper around get_token_by_exchange_profile() that fixes the
+        RFC 8693 token types for Auth0 access-token-to-access-token exchange.
+
+        Args:
+            access_token: The Auth0 access token to exchange
+            audience: Target API identifier for the exchanged access token
+            scope: Optional space-separated OAuth 2.0 scopes to request
+
+        Returns:
+            Dictionary containing:
+            - access_token (str): The exchanged Auth0 access token
+            - expires_in (int): Token lifetime in seconds
+            - expires_at (int): Unix timestamp when token expires
+            - scope (str, optional): Granted scopes
+            - token_type (str, optional): Token type (typically "Bearer")
+            - issued_token_type (str, optional): RFC 8693 issued token type identifier
+
+        Raises:
+            MissingRequiredArgumentError: If required parameters are missing
+            GetTokenByExchangeProfileError: If client credentials are not configured or validation fails
+            ApiError: If the token endpoint returns an error
+        """
+        if not audience:
+            raise MissingRequiredArgumentError("audience")
+
+        result = await self.get_token_by_exchange_profile(
+            subject_token=access_token,
+            subject_token_type=OBO_ACCESS_TOKEN_TYPE,
+            audience=audience,
+            scope=scope,
+            requested_token_type=OBO_ACCESS_TOKEN_TYPE,
+        )
+
+        obo_result: OnBehalfOfTokenResult = {
+            "access_token": result["access_token"],
+            "expires_in": result["expires_in"],
+            "expires_at": result["expires_at"],
+        }
+
+        if "scope" in result:
+            obo_result["scope"] = result["scope"]
+        if "token_type" in result:
+            obo_result["token_type"] = result["token_type"]
+        if "issued_token_type" in result:
+            obo_result["issued_token_type"] = result["issued_token_type"]
+
+        return obo_result
+
     # ===== Private Methods =====
 
     def _apply_extra(

src/auth0_api_python/config.py

@@ -27,8 +27,10 @@ class ApiClientOptions:
         dpop_required: Whether DPoP is required (default: False, allows both Bearer and DPoP).
         dpop_iat_leeway: Leeway in seconds for DPoP proof iat claim (default: 30).
         dpop_iat_offset: Maximum age in seconds for DPoP proof iat claim (default: 300).
-        client_id: Required for get_access_token_for_connection and get_token_by_exchange_profile.
-        client_secret: Required for get_access_token_for_connection and get_token_by_exchange_profile.
+        client_id: Required for get_access_token_for_connection, get_token_by_exchange_profile,
+                   and get_token_on_behalf_of.
+        client_secret: Required for get_access_token_for_connection, get_token_by_exchange_profile,
+                       and get_token_on_behalf_of.
         timeout: HTTP timeout in seconds for token endpoint requests (default: 10.0).
     """
     def __init__(

src/auth0_api_python/types.py

@@ -19,6 +19,27 @@ class DomainsResolverContext(TypedDict, total=False):
     request_headers: Optional[dict]
     unverified_iss: str
 
+
+class OnBehalfOfTokenResult(TypedDict, total=False):
+    """
+    Result returned from an On Behalf Of token exchange.
+
+    Attributes:
+        access_token: The access token issued for the downstream API.
+        expires_in: Token lifetime in seconds.
+        expires_at: Unix timestamp when the token expires.
+        scope: Granted scopes, if returned by Auth0.
+        token_type: Token type, if returned by Auth0.
+        issued_token_type: RFC 8693 issued token type, if returned by Auth0.
+    """
+
+    access_token: str
+    expires_in: int
+    expires_at: int
+    scope: str
+    token_type: str
+    issued_token_type: str
+
 DomainsResolver = Callable[
     [DomainsResolverContext], Union[list[str], Awaitable[list[str]]]
 ]