feat: add client credentials token support for M2M

EXAMPLES.md

@@ -89,6 +89,41 @@ result = await api_client.get_token_on_behalf_of(
 See the **[Token Storage Guide](docs/TokenStorage.md)** for a full working example, how to
 implement a Redis-backed store, and the built-in encryption helpers.
 
+## Client Credentials for Server-to-Server Calls
+
+Use `get_client_credentials_token()` to obtain an M2M access token for server-to-server calls
+using the OAuth 2.0 client credentials grant. The SDK authenticates via HTTP Basic and caches
+the result in the configured `token_store` (keyed by audience and scope set) so subsequent calls
+within the token's lifetime skip the network round-trip.
+
+```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>",
+))
+
+result = await api_client.get_client_credentials_token(
+    audience="https://downstream-api.example.com",
+    scope="read:data",
+)
+
+# result always has "access_token", "expires_in", "expires_at".
+# "scope" is not guaranteed in the response and may be absent.
+async with httpx.AsyncClient() as client:
+    response = await client.get(
+        "https://downstream-api.example.com/data",
+        headers={"Authorization": f"Bearer {result['access_token']}"},
+    )
+```
+
+More info: [Client Credentials Flow](https://auth0.com/docs/get-started/authentication-and-authorization-flow/client-credentials-flow)
+
 ## Inspecting Delegation After Token Verification
 
 When a downstream API or `MCP` server receives an access token that may have been issued through

README.md

@@ -255,6 +255,21 @@ Configuring a `token_store` on `ApiClientOptions` caches the exchanged token, so
 the same caller, audience, organization, scopes, and session reuses it instead of exchanging again.
 With no `token_store`, every call performs a fresh exchange, matching the existing behavior above.
 
+#### Client Credentials for Server-to-Server Calls
+
+Use `get_client_credentials_token()` to obtain an M2M access token for server-to-server calls
+using the OAuth 2.0 client credentials grant.
+
+```python
+result = await api_client.get_client_credentials_token(
+    audience="https://downstream-api.example.com",
+    scope="read:data",
+)
+# call downstream API with result["access_token"]
+```
+
+See the **[Client Credentials example](EXAMPLES.md#client-credentials-for-server-to-server-calls)** for a full example that calls a downstream API, plus the caching behavior.
+
 #### Inspecting Delegation After Token Verification
 
 When a downstream API or `MCP` server receives an access token that may have been issued through

src/auth0_api_python/__init__.py

@@ -13,6 +13,7 @@
     ApiError,
     ConfigurationError,
     DomainsResolverError,
+    GetClientCredentialsTokenError,
     GetTokenByExchangeProfileError,
     MissingOrganizationError,
     OrganizationNotAllowedError,
@@ -26,6 +27,7 @@
     VerifiedToken,
 )
 from .types import (
+    ClientCredentialsTokenResult,
     DomainsResolver,
     DomainsResolverContext,
     OnBehalfOfTokenResult,
@@ -36,10 +38,12 @@
     "ApiClientOptions",
     "ApiError",
     "CacheAdapter",
+    "ClientCredentialsTokenResult",
     "ConfigurationError",
     "DomainsResolver",
     "DomainsResolverContext",
     "DomainsResolverError",
+    "GetClientCredentialsTokenError",
     "GetTokenByExchangeProfileError",
     "TokenStoreError",
     "get_current_actor",

src/auth0_api_python/api_client.py

@@ -7,6 +7,7 @@
 import httpx
 from authlib.jose import JsonWebKey, JsonWebToken
 
+from ._internal.cache_keys import m2m_cache_key
 from ._internal.obo_cache import OboCache
 from .cache import InMemoryCache
 from .config import ApiClientOptions
@@ -16,20 +17,22 @@
     ConfigurationError,
     DomainsResolverError,
     GetAccessTokenForConnectionError,
+    GetClientCredentialsTokenError,
     GetTokenByExchangeProfileError,
     InvalidAuthSchemeError,
     InvalidDpopProofError,
     MissingAuthorizationError,
     MissingOrganizationError,
     MissingRequiredArgumentError,
     OrganizationNotAllowedError,
+    TokenStoreError,
     VerifyAccessTokenError,
 )
 from .token_store import (
     IndexedTokenStore,
     VerifiedToken,
 )
-from .types import OnBehalfOfTokenResult
+from .types import ClientCredentialsTokenResult, OnBehalfOfTokenResult
 from .utils import (
     calculate_jwk_thumbprint,
     fetch_jwks,
@@ -1123,6 +1126,151 @@ async def get_token_on_behalf_of(
 
         return obo_result
 
+    async def get_client_credentials_token(
+        self,
+        audience: str,
+        scope: Optional[str] = None,
+    ) -> ClientCredentialsTokenResult:
+        """
+        Obtain a client credentials (M2M) access token for a server-to-server call.
+
+        Args:
+            audience: Target API identifier for the access token
+            scope: Optional space-separated OAuth 2.0 scopes to request
+
+        Returns:
+            Dictionary containing:
+            - access_token (str): The access token
+            - expires_in (int): Token lifetime in seconds
+            - expires_at (int): Absolute expiration time as a Unix timestamp in seconds, calculated by the SDK from expires_in
+            - scope (str, optional): Granted scopes, if returned by Auth0
+            - token_type (str, optional): Token type, if returned by Auth0
+
+        Raises:
+            MissingRequiredArgumentError: If audience is not provided
+            GetClientCredentialsTokenError: If client credentials are not configured or token endpoint is missing
+            ApiError: If the token endpoint returns an error
+        """
+        if not audience:
+            raise MissingRequiredArgumentError("audience")
+
+        client_id = self.options.client_id
+        client_secret = self.options.client_secret
+        if not client_id or not client_secret:
+            raise GetClientCredentialsTokenError(
+                "Client credentials are required to use get_client_credentials_token. "
+                "Configure client_id and client_secret in ApiClientOptions to use this feature"
+            )
+
+        # Check cache before making a network call
+        cache_key = None
+        if self._token_store is not None:
+            cache_key = m2m_cache_key(tenant=self.options.domain, client_id=client_id, audience=audience, scopes=scope)
+
+            cached = None
+            try:
+                cached = await self._token_store.get(cache_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)
+
+            if cached is not 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")
+                elif cached["expires_at"] > int(time.time()):
+                    hit: ClientCredentialsTokenResult = {
+                        "access_token": cached["access_token"],
+                        "expires_in": cached["expires_at"] - int(time.time()),
+                        "expires_at": cached["expires_at"],
+                    }
+                    if cached.get("granted_scopes"):
+                        hit["scope"] = cached["granted_scopes"]
+                    return hit
+
+        metadata = await self._discover()
+        token_endpoint = metadata.get("token_endpoint")
+        if not token_endpoint:
+            raise GetClientCredentialsTokenError(
+                "Token endpoint missing in OIDC metadata. "
+                "Verify your domain configuration and that the OIDC discovery endpoint is accessible"
+            )
+
+        params: dict[str, str] = {
+            "grant_type": "client_credentials",
+            "audience": audience,
+        }
+
+        if scope:
+            params["scope"] = scope
+
+        try:
+            async with httpx.AsyncClient(timeout=httpx.Timeout(self.options.timeout)) as client:
+                response = await client.post(
+                    token_endpoint,
+                    data=params,
+                    auth=(client_id, client_secret)
+                )
+
+                if response.status_code != 200:
+                    error_data = {}
+                    try:
+                        content_type = response.headers.get("content-type", "").lower()
+                        if "json" in content_type:
+                            error_data = response.json()
+                    except ValueError:
+                        pass  # Ignore JSON parse errors, use generic error message below
+
+                    raise ApiError(
+                        error_data.get("error", "client_credentials_error"),
+                        error_data.get("error_description", "Failed to get client credentials token."),
+                        response.status_code
+                    )
+
+                token_response = response.json()
+
+                expires_in = token_response["expires_in"]
+                cc_result = {
+                    "access_token": token_response["access_token"],
+                    "expires_in": expires_in,
+                    "expires_at": int(time.time()) + expires_in,
+                }
+
+                # scope is not guaranteed in the client credentials response
+                if "scope" in token_response:
+                    cc_result["scope"] = token_response["scope"]
+                if "token_type" in token_response:
+                    cc_result["token_type"] = token_response["token_type"]
+
+        except httpx.TimeoutException as exc:
+            raise ApiError(
+                "timeout_error",
+                f"Request to token endpoint timed out: {str(exc)}",
+                504,
+                exc
+            )
+        except httpx.HTTPError as exc:
+            raise ApiError(
+                "network_error",
+                f"Network error occurred: {str(exc)}",
+                502,
+                exc
+            )
+
+        if cache_key is not None:
+            try:
+                entry: dict[str, Any] = {
+                    "access_token": cc_result["access_token"],
+                    "expires_at": cc_result["expires_at"],
+                }
+                if "scope" in cc_result:
+                    entry["granted_scopes"] = cc_result["scope"]
+                await self._token_store.set(cache_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)
+
+        return cc_result
+
     # ===== Private Methods =====
 
     def _apply_extra(

src/auth0_api_python/errors.py

@@ -124,6 +124,16 @@ def get_error_code(self) -> str:
         return "get_token_by_exchange_profile_error"
 
 
+class GetClientCredentialsTokenError(BaseAuthError):
+    """Error raised when a client credentials token request fails before the network call."""
+
+    def get_status_code(self) -> int:
+        return 400
+
+    def get_error_code(self) -> str:
+        return "get_client_credentials_token_error"
+
+
 class ApiError(BaseAuthError):
     """
     Error raised when an API request to Auth0 fails.

src/auth0_api_python/types.py

@@ -53,6 +53,25 @@ class OnBehalfOfTokenResult(TypedDict, total=False):
     token_type: str
     issued_token_type: str
 
+
+class ClientCredentialsTokenResult(TypedDict, total=False):
+    """
+    Result returned from a client credentials (M2M) token request.
+
+    Attributes:
+        access_token: The access token issued for the target API.
+        expires_in: Token lifetime in seconds.
+        expires_at: Absolute expiration time as a Unix timestamp in seconds, calculated by the SDK from expires_in.
+        scope: Granted scopes, if returned by Auth0.
+        token_type: Token type, if returned by Auth0.
+    """
+
+    access_token: str
+    expires_in: int
+    expires_at: int
+    scope: str
+    token_type: str
+
 DomainsResolver = Callable[
     [DomainsResolverContext], Union[list[str], Awaitable[list[str]]]
 ]