Add token vault subject_token_type access_token to api sdk

packages/auth0_api_python/README.md

@@ -85,6 +85,34 @@ asyncio.run(main())
 
 In this example, the returned dictionary contains the decoded claims (like `sub`, `scope`, etc.) from the verified token.
 
+### 4. Get an access token for a connection
+
+If you need to get an access token for an upstream idp via a connection, you can use the `get_access_token_for_connection` method:
+
+```python
+import asyncio
+
+from auth0_api_python import ApiClient, ApiClientOptions
+
+async def main():
+    api_client = ApiClient(ApiClientOptions(
+        domain="<AUTH0_DOMAIN>",
+        audience="<AUTH0_AUDIENCE>",
+        client_id="<AUTH0_CLIENT_ID>",
+        client_secret="<AUTH0_CLIENT_SECRET>",
+    ))
+    connection = "my-connection" # The Auth0 connection to the upstream idp
+    access_token = "..." # The Auth0 access token to exchange
+
+    connection_access_token = await api_client.get_access_token_for_connection({"connection": connection, "access_token": access_token})
+    # The returned token is the access token for the upstream idp
+    print(connection_access_token)
+
+asyncio.run(main())
+```
+
+More info https://auth0.com/docs/secure/tokens/token-vault
+
 #### Requiring Additional Claims
 
 If your application demands extra claims, specify them with `required_claims`:

packages/auth0_api_python/src/auth0_api_python/api_client.py

@@ -1,11 +1,14 @@
 import time
 from typing import Any, Optional
 
+import httpx
 from authlib.jose import JsonWebKey, JsonWebToken
 
 from .config import ApiClientOptions
 from .errors import (
+    ApiError,
     BaseAuthError,
+    GetAccessTokenForConnectionError,
     InvalidAuthSchemeError,
     InvalidDpopProofError,
     MissingAuthorizationError,
@@ -390,6 +393,84 @@ async def verify_dpop_proof(
 
         return claims
 
+    async def get_access_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 'access_token' keys.
+                May optionally include 'login_hint'.
+
+        Raises:
+            GetAccessTokenForConnectionError: If there was an issue requesting the access token.
+            ApiError: If the token exchange endpoint returns an error.
+
+        Returns:
+            Dictionary containing the token response with access_token, expires_in, and scope.
+        """
+        # Constants
+        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")
+
+        if not connection:
+            raise MissingRequiredArgumentError("connection")
+
+        if not access_token:
+            raise MissingRequiredArgumentError("access_token")
+
+        client_id = self.options.client_id
+        client_secret = self.options.client_secret
+        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.")
+
+        metadata = await self._discover()
+
+        token_endpoint = metadata.get("token_endpoint")
+        if not token_endpoint:
+            raise GetAccessTokenForConnectionError("Token endpoint missing in OIDC metadata")
+
+        # Prepare parameters
+        params = {
+            "connection": connection,
+            "requested_token_type": REQUESTED_TOKEN_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN,
+            "grant_type": GRANT_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN,
+            "client_id": client_id,
+            "subject_token": access_token,
+            "subject_token_type": SUBJECT_TYPE_ACCESS_TOKEN,
+        }
+
+        # Add login_hint if provided
+        if "login_hint" in options and options["login_hint"]:
+            params["login_hint"] = options["login_hint"]
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                token_endpoint,
+                data=params,
+                auth=(client_id, client_secret)
+            )
+
+            if response.status_code != 200:
+                error_data = response.json() if 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", "")
+            }
+
     # ===== Private Methods =====
 
     async def _discover(self) -> dict[str, Any]:

packages/auth0_api_python/src/auth0_api_python/config.py

@@ -17,6 +17,8 @@ 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: Optional required if you want to use get_access_token_for_connection.
+        client_secret: Optional required if you want to use get_access_token_for_connection.
     """
     def __init__(
         self,
@@ -27,6 +29,8 @@ def __init__(
         dpop_required: bool = False,
         dpop_iat_leeway: int = 30,
         dpop_iat_offset: int = 300,
+        client_id: Optional[str] = None,
+        client_secret: Optional[str] = None,
     ):
         self.domain = domain
         self.audience = audience
@@ -35,3 +39,5 @@ def __init__(
         self.dpop_required = dpop_required
         self.dpop_iat_leeway = dpop_iat_leeway
         self.dpop_iat_offset = dpop_iat_offset
+        self.client_id = client_id
+        self.client_secret = client_secret

packages/auth0_api_python/src/auth0_api_python/errors.py

@@ -94,3 +94,32 @@ def get_status_code(self) -> int:
 
     def get_error_code(self) -> str:
         return "invalid_request"
+
+
+class GetAccessTokenForConnectionError(Exception):
+    """Error raised when getting a token for a connection fails."""
+    code = "get_access_token_for_connection_error"
+
+    def __init__(self, message: str):
+        super().__init__(message)
+        self.name = self.__class__.__name__
+
+
+class ApiError(Exception):
+    """
+    Error raised when an API request to Auth0 fails.
+    Contains details about the original error from Auth0.
+    """
+
+    def __init__(self, code: str, message: str, cause=None):
+        super().__init__(message)
+        self.code = code
+        self.cause = cause
+
+        # Extract additional error details if available
+        if cause:
+            self.error = getattr(cause, "error", None)
+            self.error_description = getattr(cause, "error_description", None)
+        else:
+            self.error = None
+            self.error_description = None

packages/auth0_api_python/tests/test_api_client.py

@@ -1,11 +1,14 @@
 import base64
 import json
 import time
+import urllib
 
 import pytest
 from auth0_api_python.api_client import ApiClient
 from auth0_api_python.config import ApiClientOptions
 from auth0_api_python.errors import (
+    ApiError,
+    GetAccessTokenForConnectionError,
     InvalidAuthSchemeError,
     InvalidDpopProofError,
     MissingAuthorizationError,
@@ -1589,3 +1592,141 @@ async def test_verify_request_fail_multiple_dpop_proofs():
     assert "multiple" in str(err.value).lower()
 
 
+@pytest.mark.asyncio
+async def test_get_access_token_for_connection_success(httpx_mock: HTTPXMock):
+    httpx_mock.add_response(
+        method="GET",
+        url="https://auth0.local/.well-known/openid-configuration",
+        json={
+            "token_endpoint": "https://auth0.local/oauth/token"
+        }
+    )
+    httpx_mock.add_response(
+        method="POST",
+        url="https://auth0.local/oauth/token",
+        json={"access_token": "abc123", "expires_in": 3600, "scope": "openid"}
+    )
+    options = ApiClientOptions(
+        domain="auth0.local",
+        audience="my-audience",
+        client_id="cid",
+        client_secret="csecret",
+    )
+    api_client = ApiClient(options)
+    result = await api_client.get_access_token_for_connection({
+        "connection": "test-conn",
+        "access_token": "user-token"
+    })
+    assert result["access_token"] == "abc123"
+    assert result["scope"] == "openid"
+    assert isinstance(result["expires_at"], int)
+
+@pytest.mark.asyncio
+async def test_get_access_token_for_connection_with_login_hint(httpx_mock: HTTPXMock):
+    httpx_mock.add_response(
+        method="GET",
+        url="https://auth0.local/.well-known/openid-configuration",
+        json={
+            "token_endpoint": "https://auth0.local/oauth/token"
+        }
+    )
+    httpx_mock.add_response(
+        method="POST",
+        url="https://auth0.local/oauth/token",
+        json={"access_token": "abc123", "expires_in": 3600, "scope": "openid"}
+    )
+    options = ApiClientOptions(
+        domain="auth0.local",
+        audience="my-audience",
+        client_id="cid",
+        client_secret="csecret",
+    )
+    api_client = ApiClient(options)
+    result = await api_client.get_access_token_for_connection({
+        "connection": "test-conn",
+        "access_token": "user-token",
+        "login_hint": "user@example.com"
+    })
+    assert result["access_token"] == "abc123"
+    request = httpx_mock.get_requests()[-1]
+    form_data = urllib.parse.parse_qs(request.content.decode())
+    assert form_data["login_hint"] == ["user@example.com"]
+
+
+
+
+@pytest.mark.asyncio
+async def test_get_access_token_for_connection_missing_connection():
+    options = ApiClientOptions(
+        domain="auth0.local",
+        audience="my-audience",
+        client_id="cid",
+        client_secret="csecret",
+    )
+    api_client = ApiClient(options)
+    with pytest.raises(MissingRequiredArgumentError):
+        await api_client.get_access_token_for_connection({
+            "access_token": "user-token"
+        })
+
+
+@pytest.mark.asyncio
+async def test_get_access_token_for_connection_missing_access_token():
+    options = ApiClientOptions(
+        domain="auth0.local",
+        audience="my-audience",
+        client_id="cid",
+        client_secret="csecret",
+    )
+    api_client = ApiClient(options)
+    with pytest.raises(MissingRequiredArgumentError):
+        await api_client.get_access_token_for_connection({
+            "connection": "test-conn"
+        })
+
+
+@pytest.mark.asyncio
+async def test_get_access_token_for_connection_no_client_id():
+    options = ApiClientOptions(
+        domain="auth0.local",
+        audience="my-audience"
+        # client_id missing
+    )
+    api_client = ApiClient(options)
+    with pytest.raises(GetAccessTokenForConnectionError) as err:
+        await api_client.get_access_token_for_connection({
+            "connection": "test-conn",
+            "access_token": "user-token"
+        })
+
+    assert "You must configure the SDK with a client_id and client_secret to use get_access_token_for_connection." == str(err.value)
+
+
+@pytest.mark.asyncio
+async def test_get_access_token_for_connection_token_endpoint_error(httpx_mock: HTTPXMock):
+    httpx_mock.add_response(
+        method="GET",
+        url="https://auth0.local/.well-known/openid-configuration",
+        json={
+            "token_endpoint": "https://auth0.local/oauth/token"
+        }
+    )
+    httpx_mock.add_response(
+        method="POST",
+        url="https://auth0.local/oauth/token",
+        status_code=400,
+        json={"error": "invalid_request", "error_description": "Bad request"}
+    )
+    options = ApiClientOptions(
+        domain="auth0.local",
+        audience="my-audience",
+        client_id="cid",
+        client_secret="csecret",
+    )
+    api_client = ApiClient(options)
+    with pytest.raises(ApiError) as err:
+        await api_client.get_access_token_for_connection({
+            "connection": "test-conn",
+            "access_token": "user-token"
+        })
+    assert err.value.code == "invalid_request"