SAP · cassiofariasmachado · Jun 2, 2026 · Jun 1, 2026 · Jun 1, 2026 · Jun 1, 2026
@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.22.0"
+version = "0.23.0"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"

@@ -24,6 +24,7 @@
     IntegrationDependency,
     MCPTool,
 )
+from sap_cloud_sdk.agentgateway._token_cache import _TokenCache
 from sap_cloud_sdk.agentgateway.exceptions import AgentGatewaySDKError
 
 logger = logging.getLogger(__name__)
@@ -42,6 +43,11 @@
 _GRANT_TYPE_JWT_BEARER = "urn:ietf:params:oauth:grant-type:jwt-bearer"
 
 
+def _cache_scope_key(credentials: CustomerCredentials, app_tid: str | None) -> str:
+    """Build a cache scope key for customer-flow tokens."""
+    return f"customer::{credentials.client_id}::{app_tid or ''}"
+
+
 class _CredentialFields:
     """Field names in the credentials JSON file."""
 
@@ -212,17 +218,18 @@ def _request_token_mtls(
     timeout: float,
     app_tid: str | None = None,
     extra_data: dict | None = None,
-) -> str:
+) -> dict:
     """Make mTLS token request to IAS.
 
     Args:
         credentials: Customer credentials with certificate and private key.
         grant_type: OAuth2 grant type.
+        timeout: HTTP timeout in seconds.
         app_tid: BTP Application Tenant ID of subscriber (optional).
         extra_data: Additional form data for the token request.
 
     Returns:
-        Access token string.
+        Token response payload.
 
     Raises:
         AgentGatewaySDKError: If token request fails.
@@ -282,7 +289,7 @@ def _request_token_mtls(
             )
 
         logger.debug("Token acquired successfully (length: %d)", len(access_token))
-        return access_token
+        return token_data
 
     except httpx.RequestError as e:
         raise AgentGatewaySDKError(f"Token request failed: {e}")
@@ -292,6 +299,7 @@ def get_system_token_mtls(
     credentials: CustomerCredentials,
     timeout: float,
     app_tid: str | None = None,
+    token_cache: _TokenCache | None = None,
 ) -> str:
     """Get system-scoped token using mTLS client credentials flow.
 
@@ -301,25 +309,44 @@ def get_system_token_mtls(
         credentials: Customer credentials.
         timeout: HTTP timeout in seconds.
         app_tid: BTP Application Tenant ID of subscriber (optional).
+        token_cache: Optional token cache used to reuse still-valid tokens.
 
     Returns:
-        System-scoped access token.
+        System-scoped access token, fetched or served from cache.
     """
+    scope_key = _cache_scope_key(credentials, app_tid)
+    if token_cache:
+        cached_token = token_cache.get_system_token(scope_key)
+        if cached_token:
+            logger.debug("Using cached system token for scope '%s'", scope_key)
+            return cached_token
+
     logger.info("Acquiring system token via mTLS client credentials")
-    return _request_token_mtls(
+    token_data = _request_token_mtls(
         credentials,
         grant_type=_GRANT_TYPE_CLIENT_CREDENTIALS,
         timeout=timeout,
         app_tid=app_tid,
         extra_data={"response_type": "token"},
     )
+    access_token = token_data["access_token"]
+
+    if token_cache:
+        token_cache.set_system_token(
+            access_token,
+            token_cache.compute_expires_at(token_data),
+            scope_key,
+        )
+
+    return access_token
 
 
 def exchange_user_token(
     credentials: CustomerCredentials,
     user_token: str,
     timeout: float,
     app_tid: str | None = None,
+    token_cache: _TokenCache | None = None,
 ) -> str:
     """Exchange user token for AGW-scoped token using jwt-bearer grant.
 
@@ -331,12 +358,21 @@ def exchange_user_token(
         user_token: User's JWT token to exchange.
         timeout: HTTP timeout in seconds.
         app_tid: BTP Application Tenant ID of subscriber (optional).
+        token_cache: Optional token cache used to reuse still-valid exchanged
+            tokens.
 
     Returns:
-        AGW-scoped access token with user identity.
+        AGW-scoped access token with user identity, fetched or served from cache.
     """
+    scope_key = _cache_scope_key(credentials, app_tid)
+    if token_cache:
+        cached_token = token_cache.get_user_token(user_token, scope_key)
+        if cached_token:
+            logger.debug("Using cached exchanged user token for scope '%s'", scope_key)
+            return cached_token
+
     logger.info("Exchanging user token for AGW-scoped token via jwt-bearer grant")
-    return _request_token_mtls(
+    token_data = _request_token_mtls(
         credentials,
         grant_type=_GRANT_TYPE_JWT_BEARER,
         timeout=timeout,
@@ -346,6 +382,17 @@ def exchange_user_token(
             "token_format": "jwt",
         },
     )
+    access_token = token_data["access_token"]
+
+    if token_cache:
+        token_cache.set_user_token(
+            user_token,
+            access_token,
+            token_cache.compute_expires_at(token_data),
+            scope_key,
+        )
+
+    return access_token
 
 
 def _build_mcp_url(gateway_url: str, ord_id: str, gt_id: str) -> str:

@@ -23,6 +23,7 @@
 )
 
 from sap_cloud_sdk.agentgateway._models import MCPTool
+from sap_cloud_sdk.agentgateway._token_cache import _GatewayUrlCache, _TokenCache
 from sap_cloud_sdk.agentgateway.exceptions import MCPServerNotFoundError
 
 logger = logging.getLogger(__name__)
@@ -38,6 +39,16 @@
 _DESTINATION_INSTANCE = "default"
 
 
+def _system_scope_key(tenant_subdomain: str) -> str:
+    """Build the cache scope key for tenant-scoped system auth."""
+    return f"lob-system::{tenant_subdomain}"
+
+
+def _user_scope_key(tenant_subdomain: str) -> str:
+    """Build the cache scope key for tenant-scoped user auth."""
+    return f"lob-user::{tenant_subdomain}"
+
+
 def _ias_dest_name() -> str:
     """Get IAS destination name based on landscape.
 
@@ -184,6 +195,8 @@ def get_ias_user_fragment_name(tenant_subdomain: str) -> str:
 
 async def fetch_system_auth(
     tenant_subdomain: str,
+    token_cache: _TokenCache | None = None,
+    gateway_url_cache: _GatewayUrlCache | None = None,
 ) -> tuple[str, str]:
     """Fetch system-scoped auth (Phase 1 - client credentials).
 
@@ -192,13 +205,29 @@ async def fetch_system_auth(
 
     Args:
         tenant_subdomain: Tenant subdomain for multi-tenant lookup.
+        token_cache: Optional token cache used to reuse still-valid system
+            tokens.
+        gateway_url_cache: Optional cache for gateway URLs associated with the
+            cached system-token scope.
 
     Returns:
-        Tuple of (raw_access_token, gateway_url).
+        Tuple of `(raw_access_token, gateway_url)`, fetched or served from cache.
 
     Raises:
         MCPServerNotFoundError: If no IAS fragment or auth token is found.
     """
+    scope_key = _system_scope_key(tenant_subdomain)
+    if (token_cache is None) != (gateway_url_cache is None):
+        raise ValueError(
+            "token_cache and gateway_url_cache must both be provided or both be None"
+        )
+    if token_cache and gateway_url_cache is not None:
+        cached_token = token_cache.get_system_token(scope_key)
+        cached_gateway_url = gateway_url_cache.get(scope_key)
+        if cached_token and cached_gateway_url:
+            logger.debug("Using cached system auth for tenant '%s'", tenant_subdomain)
+            return cached_token, cached_gateway_url
+
     loop = asyncio.get_running_loop()
 
     def _fetch_system_auth_sync():
@@ -218,12 +247,25 @@ def _fetch_system_auth_sync():
 
         return _fetch_auth_token(dest_name, tenant_subdomain, options)
 
-    return await loop.run_in_executor(None, _fetch_system_auth_sync)
+    token, gateway_url = await loop.run_in_executor(None, _fetch_system_auth_sync)
+
+    if token_cache:
+        token_cache.set_system_token(
+            token,
+            token_cache.compute_expires_at_from_bearer(token),
+            scope_key,
+        )
+    if gateway_url_cache is not None:
+        gateway_url_cache[scope_key] = gateway_url
+
+    return token, gateway_url
 
 
 async def fetch_user_auth(
     user_token: str,
     tenant_subdomain: str,
+    token_cache: _TokenCache | None = None,
+    gateway_url_cache: _GatewayUrlCache | None = None,
 ) -> tuple[str, str]:
     """Fetch user-scoped auth (Phase 2 - token exchange).
 
@@ -234,13 +276,29 @@ async def fetch_user_auth(
     Args:
         user_token: User's JWT for principal propagation.
         tenant_subdomain: Tenant subdomain for multi-tenant lookup.
+        token_cache: Optional token cache used to reuse still-valid exchanged
+            user tokens.
+        gateway_url_cache: Optional cache for gateway URLs associated with the
+            cached user-token scope.
 
     Returns:
-        Tuple of (raw_access_token, gateway_url).
+        Tuple of `(raw_access_token, gateway_url)`, fetched or served from cache.
 
     Raises:
         MCPServerNotFoundError: If no IAS user fragment or auth token is found.
     """
+    scope_key = _user_scope_key(tenant_subdomain)
+    if (token_cache is None) != (gateway_url_cache is None):
+        raise ValueError(
+            "token_cache and gateway_url_cache must both be provided or both be None"
+        )
+    if token_cache and gateway_url_cache is not None:
+        cached_token = token_cache.get_user_token(user_token, scope_key)
+        cached_gateway_url = gateway_url_cache.get(scope_key)
+        if cached_token and cached_gateway_url:
+            logger.debug("Using cached user auth for tenant '%s'", tenant_subdomain)
+            return cached_token, cached_gateway_url
+
     loop = asyncio.get_running_loop()
 
     def _fetch_user_auth_sync():
@@ -262,7 +320,19 @@ def _fetch_user_auth_sync():
 
         return _fetch_auth_token(dest_name, tenant_subdomain, options)
 
-    return await loop.run_in_executor(None, _fetch_user_auth_sync)
+    token, gateway_url = await loop.run_in_executor(None, _fetch_user_auth_sync)
+
+    if token_cache:
+        token_cache.set_user_token(
+            user_token,
+            token,
+            token_cache.compute_expires_at_from_bearer(token),
+            scope_key,
+        )
+    if gateway_url_cache is not None:
+        gateway_url_cache[scope_key] = gateway_url
+
+    return token, gateway_url
 
 
 async def list_server_tools(