feat(token-vault): add caching for federated connection token exchanges

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

Author: kishore7snehil

EXAMPLES.md

@@ -124,6 +124,68 @@ async with httpx.AsyncClient() as client:
 
 More info: [Client Credentials Flow](https://auth0.com/docs/get-started/authentication-and-authorization-flow/client-credentials-flow)
 
+## Access Token for a Connection (Token Vault)
+
+`get_access_token_for_connection()` exchanges the user's Auth0 access token for a token from a federated identity provider (such as Google) via Auth0's Token Vault.
+
+Three things must exist on your Auth0 tenant before this works. The Token Vault grant must be enabled on your application, which must be a confidential (resource-server) client. The connection (for example `google-oauth2`) must be configured and enabled on that application. The user must have linked their account through Auth0's Connected Accounts flow. See the [Token Vault setup guide](https://auth0.com/docs/secure/tokens/token-vault/configure-token-vault) for the application setup and the [Connected Accounts guide](https://auth0.com/docs/secure/tokens/token-vault/connected-accounts-for-token-vault) for the user-linking step.
+
+### Basic call
+
+```python
+import httpx
+
+from auth0_api_python import ApiClient, ApiClientOptions
+
+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"
+
+result = await api_client.get_access_token_for_connection({
+    "connection": "google-oauth2",
+    "access_token": incoming_access_token,
+})
+
+async with httpx.AsyncClient() as client:
+    response = await client.get(
+        "https://www.googleapis.com/calendar/v3/calendars/primary/events",
+        headers={"Authorization": f"Bearer {result['access_token']}"},
+    )
+```
+
+### With caching
+
+Pass a `token_store` to cache the connection token. The SDK caches by `sub` (caller) and `connection`, skipping the exchange on repeat calls for the same user and provider. Pass `verified=` with the result of `verify_access_token()` to reuse that verification rather than running it again inside the exchange.
+
+```python
+from auth0_api_python import ApiClient, ApiClientOptions
+
+# token_store is your AbstractTokenStore implementation.
+# See docs/TokenStorage.md for how to build one and for encryption details.
+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>",
+    token_store=your_token_store,
+))
+
+incoming_access_token = "incoming-auth0-access-token"
+
+# Verify once, then pass the result to avoid a second verification inside the exchange.
+verified = await api_client.verify_access_token(access_token=incoming_access_token)
+
+result = await api_client.get_access_token_for_connection(
+    {"connection": "google-oauth2", "access_token": incoming_access_token},
+    verified=verified,
+)
+```
+
 ## Inspecting Delegation After Token Verification
 
 When a downstream API or `MCP` server receives an access token that may have been issued through

README.md

@@ -122,6 +122,8 @@ async def main():
 asyncio.run(main())
 ```
 
+When a `token_store` is configured on `ApiClientOptions`, the connection token is cached by caller and connection, skipping the exchange on repeat calls. Pass `verified=` to reuse an already-verified token rather than verifying a second time. See [Access Token for a Connection](EXAMPLES.md#access-token-for-a-connection-token-vault) for a full example and setup requirements.
+
 More info https://auth0.com/docs/secure/tokens/token-vault
 
 ### 5. Custom Token Exchange (Early Access)

docs/TokenStorage.md

@@ -1,9 +1,6 @@
 # Token Storage
 
-The SDK can cache access tokens it mints on the caller's behalf. Currently this covers tokens
-returned by `get_token_on_behalf_of()`. This is separate from the `CacheAdapter` described in the
-[Caching Guide](Caching.md), which only caches OIDC discovery metadata and JWKS keys, never a live
-bearer token.
+The SDK can cache access tokens it mints on the caller's behalf. The same store covers tokens returned by `get_token_on_behalf_of()`, M2M tokens from `get_client_credentials_token()`, and federated-connection tokens from `get_access_token_for_connection()`. This is separate from the `CacheAdapter` described in the [Caching Guide](Caching.md), which only caches OIDC discovery metadata and JWKS keys, never a live bearer token.
 
 ## Default Behavior
 
@@ -14,6 +11,10 @@ To enable caching, pass a `token_store` to `ApiClientOptions`. Once a store is c
 automatically builds a cache key from the incoming token and no additional argument is needed per
 call.
 
+## Cache Key Shapes
+
+The store uses a different cache key for each exchange method. OBO tokens from `get_token_on_behalf_of()` use the most detailed key, incorporating verified issuer, incoming client, exchange tenant and client, audience, organization, session, and granted scopes (see [On Behalf Of Exchange with Caching](#on-behalf-of-exchange-with-caching) for the full design). M2M tokens from `get_client_credentials_token()` are keyed on tenant, client, audience, and the requested scope set because the server is the identity and there is no user `sub`. Connection tokens from `get_access_token_for_connection()` are keyed on tenant, client, `sub` (the caller), and `connection` with no scope dimension, since the federated provider determines what scopes apply.
+
 ## On Behalf Of Exchange with Caching
 
 The following example verifies an incoming token and exchanges for a downstream token. The result

src/auth0_api_python/__init__.py

@@ -13,9 +13,11 @@
     ApiError,
     ConfigurationError,
     DomainsResolverError,
+    GetAccessTokenForConnectionError,
     GetClientCredentialsTokenError,
     GetTokenByExchangeProfileError,
     TokenStoreError,
+    VerifyAccessTokenError,
 )
 from .token_store import (
     AbstractTokenStore,
@@ -41,9 +43,11 @@
     "DomainsResolver",
     "DomainsResolverContext",
     "DomainsResolverError",
+    "GetAccessTokenForConnectionError",
     "GetClientCredentialsTokenError",
     "GetTokenByExchangeProfileError",
     "TokenStoreError",
+    "VerifyAccessTokenError",
     "get_current_actor",
     "get_delegation_chain",
     "InMemoryCache",

src/auth0_api_python/_internal/cache_keys.py

@@ -140,7 +140,7 @@ def m2m_cache_key(*, tenant: str, client_id: str, audience: str, scopes: Optiona
     return hashlib.sha256(fields.encode()).hexdigest()
 
 
-def token_vault_cache_key(sub: str, connection: str) -> str:
-    """Cache key for a Token Vault exchange, scoped to caller and connection."""
-    fields = _FIELD_SEPARATOR.join(["token_vault", sub, connection])
+def token_vault_cache_key(*, sub: str, connection: str, tenant: str, client_id: str) -> str:
+    """Cache key for a Token Vault exchange, scoped to tenant, client, caller, and connection."""
+    fields = _FIELD_SEPARATOR.join(["token_vault", tenant, client_id, sub, connection])
     return hashlib.sha256(fields.encode()).hexdigest()

src/auth0_api_python/_internal/token_vault_cache.py

@@ -0,0 +1,68 @@
+"""Caching for Token Vault (federated connection) exchanges: lookup and write."""
+
+import logging
+import time
+from typing import Optional
+
+from ..errors import TokenStoreError
+from ..token_store import AbstractTokenStore, TokenSet
+from .cache_keys import _normalized_scopes, token_vault_cache_key
+
+
+class TokenVaultCache:
+    """Caches Token Vault exchange results in a token store, keyed by caller and connection.
+
+    ApiClient builds one TokenVaultCache when a token_store is configured and delegates
+    every cache decision to it.
+    """
+
+    def __init__(self, token_store: AbstractTokenStore) -> None:
+        self._store = token_store
+
+    def cache_key(self, tenant: str, client_id: str, sub: Optional[str], connection: str) -> Optional[str]:
+        """Return the cache key, or None when sub is absent (cache skipped with a warning)."""
+        if not sub:
+            logging.warning("Access token has no usable sub claim, skipping Token Vault cache")
+            return None
+        return token_vault_cache_key(sub=sub, connection=connection, tenant=tenant, client_id=client_id)
+
+    async def lookup(self, key: str) -> Optional[dict]:
+        """Return a live cached token or None. Miss cases: absent, malformed, expired."""
+        cached: Optional[TokenSet] = None
+        try:
+            cached = await self._store.get(key)
+        except Exception as exc:
+            store_err = TokenStoreError("Token store read failed", cause=exc)
+            logging.warning("Token store read failed, treating as cache miss: %s", store_err.cause)
+            return None
+
+        if cached is None:
+            return None
+        if "access_token" not in cached or "expires_at" not in cached:
+            logging.warning("Token store returned a malformed entry, treating as cache miss")
+            return None
+        if cached["expires_at"] <= int(time.time()):
+            return None
+
+        result = {
+            "access_token": cached["access_token"],
+            "expires_in": cached["expires_at"] - int(time.time()),
+            "expires_at": cached["expires_at"],
+        }
+        granted = cached.get("granted_scopes")
+        if granted:
+            result["scope"] = granted
+        return result
+
+    async def write(self, key: str, result: dict) -> None:
+        """Store a freshly exchanged connection token. Failures are logged and swallowed."""
+        entry: TokenSet = {
+            "access_token": result["access_token"],
+            "expires_at": result["expires_at"],
+            "granted_scopes": _normalized_scopes(result.get("scope")),
+        }
+        try:
+            await self._store.set(key, entry)
+        except Exception as exc:
+            store_err = TokenStoreError("Token store write failed", cause=exc)
+            logging.warning("Token store write failed, token still returned: %s", store_err.cause)

src/auth0_api_python/api_client.py

@@ -9,6 +9,7 @@
 
 from ._internal.cache_keys import m2m_cache_key
 from ._internal.obo_cache import OboCache
+from ._internal.token_vault_cache import TokenVaultCache
 from .cache import InMemoryCache
 from .config import ApiClientOptions
 from .errors import (
@@ -156,6 +157,11 @@ def __init__(self, options: ApiClientOptions):
             if options.token_store is not None
             else None
         )
+        self._token_vault_cache = (
+            TokenVaultCache(options.token_store)
+            if options.token_store is not None
+            else None
+        )
 
         self._cache_ttl = options.cache_ttl_seconds
 
@@ -717,18 +723,27 @@ async def verify_dpop_proof(
 
         return claims
 
-    async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict[str, Any]:
+    async def get_access_token_for_connection(
+        self,
+        options: dict[str, Any],
+        *,
+        verified: Optional[VerifiedToken] = None,
+    ) -> dict[str, Any]:
         """
         Retrieves a token for a connection.
 
         Args:
             options: Options for retrieving an access token for a connection.
                 Must include 'connection' and 'access_token' keys.
                 May optionally include 'login_hint'.
+            verified: The already-verified token, supplied by a caller that has verified it (for
+                      example an MCP server). When omitted and a token_store is configured, the
+                      token is verified here before any cache lookup.
 
         Raises:
             GetAccessTokenForConnectionError: If there was an issue requesting the access token.
             ApiError: If the token exchange endpoint returns an error.
+            VerifyAccessTokenError: If a store is configured and either verified is omitted and the token fails verification, or verified is supplied but does not match the access token being exchanged.
 
         Returns:
             Dictionary containing the token response with access_token, expires_in, and scope.
@@ -737,8 +752,8 @@ async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict
         SUBJECT_TYPE_ACCESS_TOKEN = "urn:ietf:params:oauth:token-type:access_token"  # noqa S105
         REQUESTED_TOKEN_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN = "http://auth0.com/oauth/token-type/federated-connection-access-token"  # noqa S105
         GRANT_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN = "urn:auth0:params:oauth:grant-type:token-exchange:federated-connection-access-token"  # noqa S105
-        connection = options.get("connection")
-        access_token = options.get("access_token")
+        connection = options.get("connection", "")
+        access_token = options.get("access_token", "")
 
         if not connection:
             raise MissingRequiredArgumentError("connection")
@@ -751,6 +766,28 @@ async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict
         if not client_id or not client_secret:
             raise GetAccessTokenForConnectionError("You must configure the SDK with a client_id and client_secret to use get_access_token_for_connection.")
 
+        cache_key = None
+        if self._token_vault_cache is not None:
+            if verified is None:
+                claims = await self.verify_access_token(access_token)
+                verified = VerifiedToken(access_token=access_token, claims=claims)
+            elif verified.access_token != access_token:
+                # Claims must belong to the token being exchanged or a caller could read another token's entry.
+                raise VerifyAccessTokenError(
+                    "verified token does not match the access token being exchanged"
+                )
+            cache_key = self._token_vault_cache.cache_key(
+                tenant=self.options.domain or "",
+                client_id=self.options.client_id or "",
+                sub=verified.claims.get("sub"),
+                connection=connection,
+            )
+
+        if cache_key is not None:
+            hit = await self._token_vault_cache.lookup(cache_key)
+            if hit is not None:
+                return hit
+
         metadata = await self._discover()
 
         token_endpoint = metadata.get("token_endpoint")
@@ -809,12 +846,18 @@ async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict
                 except (TypeError, ValueError):
                     raise ApiError("invalid_response", "expires_in is not an integer.", 502)
 
-                return {
+                result = {
                     "access_token": access_token,
+                    "expires_in": expires_in,
                     "expires_at": int(time.time()) + expires_in,
                     "scope": token_endpoint_response.get("scope", "")
                 }
 
+                if cache_key is not None:
+                    await self._token_vault_cache.write(cache_key, result)
+
+                return result
+
         except httpx.TimeoutException as exc:
             raise ApiError(
                 "timeout_error",