feat: add On-Behalf-Of token exchange support

EXAMPLES.md

@@ -17,6 +17,11 @@
   - [DPoP with MCD](#dpop-with-mcd)
 - [Discovery Cache Configuration](#discovery-cache-configuration)
 - [Protecting API Routes](#protecting-api-routes)
+- [On-Behalf-Of Token Exchange](#on-behalf-of-token-exchange)
+  - [Performing the Exchange](#performing-the-exchange)
+  - [`get_token_on_behalf_of()` Return Value](#get_token_on_behalf_of-return-value)
+  - [Error Handling](#error-handling)
+  - [Inspecting Delegation After Token Verification](#inspecting-delegation-after-token-verification)
 
 ## Configuration
 
@@ -356,3 +361,140 @@ async def protected_route(claims=Depends(auth0.require_auth())):
 
 > [!IMPORTANT]
 > The above is to protect API routes by the means of a bearer token, and not server-side rendering routes using a session.
+
+## On-Behalf-Of Token Exchange
+
+Use `get_token_on_behalf_of()` on the underlying `api_client` 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 flow has three steps:
+
+1. **Verify** the incoming access token so your API rejects invalid or mis-targeted tokens before exchanging. `require_auth()` does this for the route.
+2. **Exchange** the verified token for a new access token scoped to the downstream API.
+3. **Call** the downstream API using the exchanged token.
+
+`get_token_on_behalf_of()` requires a confidential client. Configure the plugin with `client_id` and `client_secret`. Calling it without client credentials raises `GetTokenByExchangeProfileError`.
+
+### Performing the Exchange
+
+Inside a protected route, extract the raw incoming token, exchange it for a downstream audience, then call the downstream API:
+
+```python
+import httpx
+from fastapi import Depends, FastAPI, Request
+from fastapi_plugin import Auth0FastAPI
+
+app = FastAPI()
+
+auth0 = Auth0FastAPI(
+    domain="<AUTH0_DOMAIN>",              # your MCP server's Auth0 tenant domain
+    audience="<AUTH0_AUDIENCE>",          # your MCP server's API audience
+    client_id="<AUTH0_CLIENT_ID>",        # required for OBO
+    client_secret="<AUTH0_CLIENT_SECRET>",  # required for OBO
+)
+
+@app.post("/schedule-meeting")
+async def schedule_meeting(request: Request, claims=Depends(auth0.require_auth())):
+    # require_auth() already verified the incoming token. Pass the raw token
+    # to the exchange, without the "Bearer " prefix.
+    incoming_access_token = request.headers["authorization"].split(" ", 1)[1]
+
+    obo = await auth0.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:
+        response = await client.post(
+            "https://calendar-api.example.com/meetings",
+            headers={"Authorization": f"Bearer {obo['access_token']}"},
+            json=await request.json(),
+        )
+    response.raise_for_status()
+
+    return {"user": claims["sub"], "meeting": response.json()}
+```
+
+> [!TIP]
+> **Production notes:**
+> - `require_auth()` verifies the incoming token before your handler runs. Always protect routes with `require_auth()` before calling `get_token_on_behalf_of()`.
+> - Pass the raw JWT to `get_token_on_behalf_of()`. Do not include the `Bearer ` prefix or the full `Authorization` header.
+> - The downstream `audience` must match an API identifier configured in your Auth0 tenant, and your client must be authorized to access it.
+> - `get_token_on_behalf_of()` only returns access-token-oriented fields. It does not expose `id_token` or `refresh_token`.
+> - OBO requires a **confidential client**. Calling it without client credentials raises `GetTokenByExchangeProfileError`.
+
+> [!NOTE]
+> **DPoP:** `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 does not carry a DPoP proof or request a DPoP-bound result. The exchanged token comes back as a plain bearer token, so the downstream call uses `Authorization: Bearer`. If the incoming token was DPoP-bound, that binding is not preserved on the exchanged token.
+
+### `get_token_on_behalf_of()` Return Value
+
+On success, the method returns a dict containing:
+
+- `access_token`: The exchanged access token issued for the downstream API.
+- `expires_in`: Token lifetime in seconds.
+- `expires_at`: The access token expiration time, in seconds since the Unix epoch.
+- `scope`: The scope granted for the exchanged token, if returned.
+- `token_type`: The returned token type, if returned.
+- `issued_token_type`: The returned RFC 8693 issued token type, if returned.
+
+### Error Handling
+
+These error types cover the failure scenarios you will encounter. All are re-exported from `fastapi_plugin`, and all subclass `BaseAuthError`, so you can catch that single type if you want one handler for everything:
+
+- `MissingRequiredArgumentError`: Raised when `audience` or `access_token` is empty.
+- `GetTokenByExchangeProfileError`: Raised for a missing confidential client (no `client_id` or `client_secret`), and also at request time when the incoming token is malformed. This includes a token carrying the `Bearer ` prefix, a blank or whitespace-only token, a token with leading or trailing whitespace, and a `token_endpoint` missing from OIDC discovery. Pass the raw JWT with no `Bearer ` prefix to avoid the token-format cases.
+- `ApiError`: Raised when Auth0 rejects the exchange. The error preserves the OAuth error code and description from Auth0 (for example, `invalid_target` when the client is not authorized to access the downstream API).
+- `VerifyAccessTokenError`: Raised by `get_current_actor()` and `get_delegation_chain()` when the token's `act` claim is present but malformed.
+
+```python
+from fastapi_plugin import ApiError, GetTokenByExchangeProfileError, MissingRequiredArgumentError
+
+try:
+    obo = await auth0.api_client.get_token_on_behalf_of(
+        access_token=incoming_access_token,
+        audience="https://calendar-api.example.com",
+    )
+except MissingRequiredArgumentError:
+    # audience or access_token was empty.
+    raise
+except GetTokenByExchangeProfileError:
+    # Missing client credentials, or a malformed incoming token (for example the "Bearer " prefix).
+    raise
+except ApiError as err:
+    # Auth0 rejected the exchange. err.get_error_code() carries the OAuth error code.
+    raise
+```
+
+### Inspecting Delegation After Token Verification
+
+When a downstream API receives an exchanged token, it can verify the token first and then inspect the `act` claim to identify the current actor for authorization and the full delegation chain for audit or attribution. The plugin re-exports `get_current_actor` and `get_delegation_chain` for this.
+
+```python
+from fastapi import Depends, FastAPI, HTTPException
+from fastapi_plugin import Auth0FastAPI, get_current_actor, get_delegation_chain
+
+app = FastAPI()
+
+auth0 = Auth0FastAPI(
+    domain="<AUTH0_DOMAIN>",
+    audience="https://calendar-api.example.com",
+)
+
+ALLOWED_ACTORS = ["<MCP_SERVER_CLIENT_ID>"]
+
+@app.get("/meetings")
+async def list_meetings(claims=Depends(auth0.require_auth())):
+    current_actor = get_current_actor(claims)
+    delegation_chain = get_delegation_chain(claims)
+
+    if current_actor not in ALLOWED_ACTORS:
+        raise HTTPException(status_code=403, detail={"error": "insufficient_permissions"})
+
+    return {
+        "user_sub": claims["sub"],
+        "current_actor": current_actor,
+        "delegation_chain": delegation_chain,
+    }
+```
+
+Only the outermost `act.sub` represents the current actor and should be used for authorization decisions. Nested `act` values represent prior actors and are better suited for logging, audit, or attribution. See [RFC 8693, section 4.1](https://datatracker.ietf.org/doc/html/rfc8693#section-4.1) for details.

README.md

@@ -327,6 +327,38 @@ When both `domain` and `domains` are configured, the SDK uses `domains` exclusiv
 
 For detailed examples including dynamic resolvers, cache configuration, security requirements, and DPoP integration, see the [Multiple Custom Domains section in EXAMPLES.md](EXAMPLES.md#multiple-custom-domains-mcd).
 
+### 8. On-Behalf-Of Token Exchange
+
+If your API receives an Auth0 access token for itself and needs to call a downstream API on behalf of the same user (for example, an MCP server), use `get_token_on_behalf_of` on the underlying `api_client` to exchange the incoming token for one scoped to the downstream API. This requires a confidential client (`client_id` and `client_secret`).
+
+```python
+import asyncio
+
+from fastapi_plugin import Auth0FastAPI
+
+async def main():
+    auth0 = Auth0FastAPI(
+        domain="<AUTH0_DOMAIN>",
+        audience="<AUTH0_AUDIENCE>",
+        client_id="<AUTH0_CLIENT_ID>",
+        client_secret="<AUTH0_CLIENT_SECRET>",
+    )
+    incoming_access_token = "..."  # the verified Auth0 access token to exchange
+
+    result = await auth0.api_client.get_token_on_behalf_of(
+        access_token=incoming_access_token,
+        audience="https://calendar-api.example.com",
+        scope="calendar:read calendar:write",
+    )
+    print(result["access_token"])  # short-lived token for the downstream API
+
+asyncio.run(main())
+```
+
+A downstream API can inspect the delegation on a verified token with the re-exported `get_current_actor` and `get_delegation_chain` helpers. Only the outermost `act.sub` should be used for authorization decisions.
+
+For the full flow, production notes, and delegation inspection, see the [On-Behalf-Of Token Exchange section in EXAMPLES.md](EXAMPLES.md#on-behalf-of-token-exchange).
+
 ## Feedback
 
 ### Contributing

fastapi_plugin/__init__.py

@@ -1,20 +1,38 @@
 from auth0_api_python import (
+    ApiError,
     CacheAdapter,
     ConfigurationError,
     DomainsResolver,
     DomainsResolverContext,
     DomainsResolverError,
+    GetTokenByExchangeProfileError,
     InMemoryCache,
+    OnBehalfOfTokenResult,
+    get_current_actor,
+    get_delegation_chain,
+)
+from auth0_api_python.errors import (
+    BaseAuthError,
+    MissingRequiredArgumentError,
+    VerifyAccessTokenError,
 )
 
 from .fast_api_client import Auth0FastAPI
 
 __all__ = [
+    "ApiError",
     "Auth0FastAPI",
+    "BaseAuthError",
     "CacheAdapter",
     "ConfigurationError",
     "DomainsResolver",
     "DomainsResolverContext",
     "DomainsResolverError",
+    "GetTokenByExchangeProfileError",
     "InMemoryCache",
+    "MissingRequiredArgumentError",
+    "OnBehalfOfTokenResult",
+    "VerifyAccessTokenError",
+    "get_current_actor",
+    "get_delegation_chain",
 ]

tests/test_on_behalf_of.py

@@ -0,0 +1,139 @@
+"""
+Tests for the On-Behalf-Of (OBO) surface this plugin owns: the require_auth() ->
+pull the verified token -> exchange flow through a real FastAPI route, and the
+re-exports the plugin adds on top of auth0-api-python.
+
+The exchange, the form well-formedness, the Basic auth encoding, and the act-claim
+parsing are exercised in auth0-api-python's own suite (test_api_client.py, test_act.py),
+so they are not repeated here.
+"""
+import base64
+import urllib.parse
+
+import pytest
+from fastapi import Depends, FastAPI, Request
+from fastapi.testclient import TestClient
+from pytest_httpx import HTTPXMock
+
+from fastapi_plugin import (
+    ApiError,
+    Auth0FastAPI,
+    BaseAuthError,
+    GetTokenByExchangeProfileError,
+    MissingRequiredArgumentError,
+    OnBehalfOfTokenResult,
+    VerifyAccessTokenError,
+    get_current_actor,
+    get_delegation_chain,
+)
+
+from .test_utils import generate_token
+
+TOKEN_ENDPOINT = "https://auth0.local/oauth/token"
+
+
+def _setup_obo_mocks(httpx_mock: HTTPXMock):
+    """OIDC discovery (with a token_endpoint), JWKS, and the token-exchange response."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://auth0.local/.well-known/openid-configuration",
+        json={
+            "issuer": "https://auth0.local/",
+            "jwks_uri": "https://auth0.local/.well-known/jwks.json",
+            "token_endpoint": TOKEN_ENDPOINT,
+        },
+    )
+    from .conftest import PUBLIC_DPOP_JWK, RSA_PUBLIC_KEY
+    httpx_mock.add_response(
+        method="GET",
+        url="https://auth0.local/.well-known/jwks.json",
+        json={"keys": [RSA_PUBLIC_KEY, PUBLIC_DPOP_JWK]},
+    )
+    httpx_mock.add_response(
+        method="POST",
+        url=TOKEN_ENDPOINT,
+        json={
+            "access_token": "obo-access-token",
+            "expires_in": 3600,
+            "scope": "calendar:read",
+            "token_type": "Bearer",
+            "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        },
+    )
+
+
+# =============================================================================
+# The owned flow: require_auth() -> pull verified token -> exchange, via a route
+# =============================================================================
+
+@pytest.mark.asyncio
+async def test_documented_route_verifies_then_exchanges(httpx_mock: HTTPXMock):
+    """A protected route pulls the verified token and exchanges it on behalf of the user."""
+    _setup_obo_mocks(httpx_mock)
+
+    access_token = await generate_token(
+        domain="auth0.local",
+        user_id="user_123",
+        audience="my-audience",
+        issuer="https://auth0.local/",
+        iat=True,
+        exp=True,
+    )
+
+    app = FastAPI()
+    auth0 = Auth0FastAPI(
+        domain="auth0.local",
+        audience="my-audience",
+        client_id="cid",
+        client_secret="csecret",
+    )
+
+    @app.post("/schedule-meeting")
+    async def schedule_meeting(request: Request, claims=Depends(auth0.require_auth())):
+        incoming = request.headers["authorization"].split(" ", 1)[1]
+        obo = await auth0.api_client.get_token_on_behalf_of(
+            access_token=incoming,
+            audience="https://calendar-api.example.com",
+            scope="calendar:read",
+        )
+        return {"user": claims["sub"], "downstream_token": obo["access_token"]}
+
+    client = TestClient(app)
+    response = client.post(
+        "/schedule-meeting",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"user": "user_123", "downstream_token": "obo-access-token"}
+
+    exchange_req = httpx_mock.get_requests(method="POST", url=TOKEN_ENDPOINT)[-1]
+    form = urllib.parse.parse_qs(exchange_req.content.decode())
+    assert form["subject_token"] == [access_token]
+    assert form["audience"] == ["https://calendar-api.example.com"]
+    assert "client_secret" not in form
+    auth_header = exchange_req.headers.get("authorization")
+    assert auth_header.startswith("Basic ")
+    assert base64.b64decode(auth_header.split(" ")[1]).decode() == "cid:csecret"
+
+
+# =============================================================================
+# Re-exports the plugin adds on top of auth0-api-python
+# =============================================================================
+
+def test_obo_surface_reexported():
+    """The OBO method's helpers, result type, and error types are re-exported from the plugin."""
+    import auth0_api_python as dep
+    import auth0_api_python.errors as errors
+
+    assert get_current_actor is dep.get_current_actor
+    assert get_delegation_chain is dep.get_delegation_chain
+    assert OnBehalfOfTokenResult is dep.OnBehalfOfTokenResult
+    assert GetTokenByExchangeProfileError is dep.GetTokenByExchangeProfileError
+    assert ApiError is dep.ApiError
+    # These three are only under auth0_api_python.errors upstream, not at its top level.
+    assert MissingRequiredArgumentError is errors.MissingRequiredArgumentError
+    assert VerifyAccessTokenError is errors.VerifyAccessTokenError
+    assert BaseAuthError is errors.BaseAuthError
+    for err in (GetTokenByExchangeProfileError, ApiError, MissingRequiredArgumentError, VerifyAccessTokenError):
+        assert issubclass(err, BaseAuthError)