feat: add MCP tools caching to list_mcp_tools

pyproject.toml

@@ -1,7 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-
-version = "0.49.0"
+version = "0.49.1"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"

src/sap_cloud_sdk/agentgateway/__init__.py

@@ -54,6 +54,7 @@
 
 from sap_cloud_sdk.agentgateway._models import (
     AuthResult,
+    CacheOptions,
     MCPTool,
     MCPToolFilter,
     Agent,
@@ -78,6 +79,7 @@
     "ClientConfig",
     # Data models
     "AuthResult",
+    "CacheOptions",
     "MCPTool",
     "MCPToolFilter",
     "Agent",

src/sap_cloud_sdk/agentgateway/_models.py

@@ -1,8 +1,18 @@
 """Data models for Agent Gateway MCP tools."""
 
+from __future__ import annotations
+
 import json
 from dataclasses import dataclass, field
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+from sap_cloud_sdk.agentgateway.config import (
+    DEFAULT_MAX_MCP_TOOLS_CACHE_SIZE,
+    DEFAULT_MCP_TOOLS_CACHE_TTL_SECONDS,
+)
+
+if TYPE_CHECKING:
+    from sap_cloud_sdk.agentgateway._tools_cache import MCPToolsCache
 
 
 @dataclass
@@ -213,3 +223,50 @@ class MCPToolFilter:
 
     names: list[str] = field(default_factory=list)
     ord_ids: list[str] = field(default_factory=list)
+
+
+class CacheOptions:
+    """Options for caching the result of list_mcp_tools.
+
+    Pass an instance to list_mcp_tools(cache=...) to enable result caching.
+    The same instance can be reused across calls — cache state is stored on
+    it. Call evict() to force a fresh fetch on the next call.
+
+    Args:
+        ttl: Cache lifetime in seconds. Defaults to 600 s.
+        max_size: Maximum number of distinct cached entries (keyed by filter
+            combo + auth type). Oldest entry is evicted when the limit is
+            exceeded. Defaults to 32.
+
+    Example:
+        ```python
+        from sap_cloud_sdk.agentgateway import CacheOptions
+
+        cache = CacheOptions(ttl=300)
+        tools = await agw_client.list_mcp_tools(cache=cache)
+
+        # Later — force a fresh fetch (e.g. after a tool was added):
+        cache.evict()
+        tools = await agw_client.list_mcp_tools(cache=cache)
+        ```
+
+    Note:
+        Cache is in-process only. It is not shared across client instances,
+        processes, or Kubernetes pods. Two concurrent calls that both miss
+        the cache will both fetch independently — the last writer wins, no
+        data corruption occurs.
+    """
+
+    def __init__(
+        self,
+        ttl: float = DEFAULT_MCP_TOOLS_CACHE_TTL_SECONDS,
+        max_size: int = DEFAULT_MAX_MCP_TOOLS_CACHE_SIZE,
+    ) -> None:
+        self.ttl = ttl
+        self.max_size = max_size
+        self._cache: MCPToolsCache | None = None
+
+    def evict(self) -> None:
+        """Clear all cached tool list entries. Forces a fresh fetch on the next call."""
+        if self._cache is not None:
+            self._cache.evict()

src/sap_cloud_sdk/agentgateway/_tools_cache.py

@@ -0,0 +1,89 @@
+"""Result cache for MCP tool lists.
+
+Caches list[MCPTool] per (filter, auth-type) key to avoid redundant MCP
+session round-trips during agentic loops. Bounded by max_size with LRU
+eviction; each entry has a monotonic TTL.
+
+Thread safety:
+CPython GIL makes individual OrderedDict operations atomic, but compound
+check-then-set is not. Two concurrent coroutines for the same key may both
+miss and both fetch; the race produces redundant tool-list requests, not
+data corruption. This matches the accepted behaviour in _token_cache.py.
+"""
+
+import logging
+import time
+from collections import OrderedDict
+from dataclasses import dataclass
+
+from sap_cloud_sdk.agentgateway._models import CacheOptions, MCPTool, MCPToolFilter
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class _CachedToolList:
+    tools: list[MCPTool]
+    expires_at: float  # time.monotonic() value
+
+    def is_valid(self) -> bool:
+        return time.monotonic() < self.expires_at
+
+
+def _make_cache_key(filter: MCPToolFilter | None, user_scoped: bool) -> str:
+    """Build a stable string key from filter options and auth type."""
+    ord_ids = "|".join(sorted(filter.ord_ids)) if filter and filter.ord_ids else ""
+    names = "|".join(sorted(filter.names)) if filter and filter.names else ""
+    auth = "user" if user_scoped else "system"
+    return f"{auth}:ord={ord_ids}:names={names}"
+
+
+class MCPToolsCache:
+    """TTL + LRU cache for MCP tool list results.
+
+    Keyed by (filter combo, auth type). Entries expire after `options.ttl`
+    seconds. When the number of entries exceeds `options.max_size`, the
+    least-recently-used entry is evicted.
+
+    Callers hold a reference to their CacheOptions instance and call
+    evict() to invalidate all entries.
+    """
+
+    def __init__(self) -> None:
+        self._entries: OrderedDict[str, _CachedToolList] = OrderedDict()
+
+    def get(
+        self,
+        filter: MCPToolFilter | None,
+        user_scoped: bool,
+    ) -> list[MCPTool] | None:
+        """Return cached tools for the given filter/auth combo, or None if miss/expired."""
+        key = _make_cache_key(filter, user_scoped)
+        entry = self._entries.get(key)
+        if entry and entry.is_valid():
+            self._entries.move_to_end(key)
+            return entry.tools
+        if entry:
+            del self._entries[key]
+        return None
+
+    def set(
+        self,
+        tools: list[MCPTool],
+        filter: MCPToolFilter | None,
+        user_scoped: bool,
+        options: CacheOptions,
+    ) -> None:
+        """Store tools under the given filter/auth key, evicting LRU if at capacity."""
+        key = _make_cache_key(filter, user_scoped)
+        expires_at = time.monotonic() + options.ttl
+        self._entries[key] = _CachedToolList(tools=tools, expires_at=expires_at)
+        self._entries.move_to_end(key)
+        while len(self._entries) > options.max_size:
+            evicted, _ = self._entries.popitem(last=False)
+            logger.debug("MCP tools cache full — evicted key '%s'", evicted)
+
+    def evict(self) -> None:
+        """Clear all cached entries. Forces a fresh fetch on the next call."""
+        self._entries.clear()
+        logger.debug("MCP tools cache evicted")

src/sap_cloud_sdk/agentgateway/agw_client.py

@@ -34,11 +34,13 @@
     Agent,
     AgentCardFilter,
     AuthResult,
+    CacheOptions,
     MCPTool,
     MCPToolFilter,
 )
 from sap_cloud_sdk.core._tenant import _validate_tenant_subdomain
 from sap_cloud_sdk.agentgateway._token_cache import _GatewayUrlCache, _TokenCache
+from sap_cloud_sdk.agentgateway._tools_cache import MCPToolsCache
 from sap_cloud_sdk.agentgateway.exceptions import AgentGatewaySDKError
 from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
 
@@ -375,6 +377,7 @@ async def list_mcp_tools(
         self,
         user_token: str | Callable[[], str] | None = None,
         filter: MCPToolFilter | None = None,
+        cache: CacheOptions | None = None,
     ) -> list[MCPTool]:
         """List all MCP tools from MCP servers.
 
@@ -395,6 +398,11 @@ async def list_mcp_tools(
                 If provided, uses user-scoped auth instead of system auth.
             filter: Optional filter to narrow results by tool name or ORD ID.
                 If None or empty, all tools are included.
+            cache: Optional caching options. When provided, tool lists are cached
+                in-process for ``cache.ttl`` seconds (default 600 s). Distinct filter
+                and auth-type combinations are cached independently, up to
+                ``cache.max_size`` entries (LRU eviction). Call ``cache.evict()`` to
+                clear all entries and force a fresh fetch on the next call.
 
         Returns:
             List of MCPTool objects from all MCP servers.
@@ -419,9 +427,32 @@ async def list_mcp_tools(
                     ord_ids=["sap.s4:apiAccess:salesOrder:v1"],
                 )
             )
+
+            # With caching — avoids redundant MCP round-trips:
+            from sap_cloud_sdk.agentgateway import CacheOptions
+            cache = CacheOptions(ttl=300)
+            tools = await agw_client.list_mcp_tools(cache=cache)
+
+            # Force a fresh fetch (e.g. after a tool was added on the server):
+            cache.evict()
+            tools = await agw_client.list_mcp_tools(cache=cache)
             ```
         """
         try:
+            user_scoped = bool(user_token)
+
+            if cache is not None:
+                if cache._cache is None:
+                    cache._cache = MCPToolsCache()
+                tools_cache: MCPToolsCache | None = cache._cache
+                cache_opts: CacheOptions | None = cache
+                cached = tools_cache.get(filter, user_scoped)
+                if cached is not None:
+                    return cached
+            else:
+                tools_cache = None
+                cache_opts = None
+
             if user_token:
                 auth = await self.get_user_auth(user_token)
             else:
@@ -434,32 +465,41 @@ async def list_mcp_tools(
                     "Customer agent credentials detected at '%s'", credentials_path
                 )
                 credentials = load_customer_credentials(credentials_path)
-                return await get_mcp_tools_customer(
+                tools = await get_mcp_tools_customer(
                     credentials,
                     auth.access_token,
                     self._config.timeout,
                     filter=filter,
                 )
+                if tools_cache is not None and cache_opts is not None:
+                    tools_cache.set(tools, filter, user_scoped, cache_opts)
+                return tools
 
             # Check for transparent mode
             if detect_transparent_credentials():
                 logger.info(_LOG_TRANSPARENT_MODE)
                 credentials = load_customer_credentials_from_env()
-                return await get_mcp_tools_customer(
+                tools = await get_mcp_tools_customer(
                     credentials,
                     auth.access_token,
                     self._config.timeout,
                     filter=filter,
                 )
+                if tools_cache is not None and cache_opts is not None:
+                    tools_cache.set(tools, filter, user_scoped, cache_opts)
+                return tools
 
             # LoB flow - requires tenant_subdomain
             tenant = self._resolve_tenant_subdomain()
-            return await get_mcp_tools_lob(
+            tools = await get_mcp_tools_lob(
                 tenant,
                 auth.access_token,
                 self._config.timeout,
                 filter=filter,
             )
+            if tools_cache is not None and cache_opts is not None:
+                tools_cache.set(tools, filter, user_scoped, cache_opts)
+            return tools
 
         except AgentGatewaySDKError:
             raise

src/sap_cloud_sdk/agentgateway/config.py

@@ -7,6 +7,8 @@
 DEFAULT_TOKEN_EXPIRY_BUFFER_SECONDS = 30.0
 DEFAULT_MAX_SYSTEM_TOKEN_CACHE_SIZE = 32
 DEFAULT_MAX_USER_TOKEN_CACHE_SIZE = 256
+DEFAULT_MCP_TOOLS_CACHE_TTL_SECONDS = 600.0
+DEFAULT_MAX_MCP_TOOLS_CACHE_SIZE = 32
 
 
 @dataclass

src/sap_cloud_sdk/agentgateway/user-guide.md

@@ -95,6 +95,36 @@ agents = await agw_client.list_agent_cards(
 )
 ```
 
+### Caching Tool Lists
+
+In agentic loops, `list_mcp_tools()` can be called repeatedly. By default every call opens fresh MCP sessions — expensive for a tool list that rarely changes. Pass a `CacheOptions` instance to cache results in-process.
+
+```python
+from sap_cloud_sdk.agentgateway import CacheOptions, create_client
+
+agw_client = create_client(tenant_subdomain="my-tenant")
+cache = CacheOptions(ttl=300)  # cache for 5 minutes
+
+# First call fetches from network and stores in cache
+tools = await agw_client.list_mcp_tools(cache=cache)
+
+# Subsequent calls within TTL return immediately — no network round-trip
+tools = await agw_client.list_mcp_tools(cache=cache)
+
+# Force a fresh fetch (e.g. after a tool was added on the server):
+cache.evict()
+tools = await agw_client.list_mcp_tools(cache=cache)
+```
+
+The cache is scoped to the `CacheOptions` instance — different instances don't share state. Distinct filter and auth-type combinations are cached as independent entries, up to `max_size` entries total (LRU eviction when the limit is hit).
+
+```python
+# Custom TTL and size cap
+cache = CacheOptions(ttl=600, max_size=10)
+```
+
+The cache is **in-process only** — not shared across client instances, processes, or Kubernetes pods.
+
 ### LangChain Integration
 
 Convert MCP tools to LangChain `StructuredTool` objects for use with LangChain agents:
@@ -221,6 +251,7 @@ class AgentGatewayClient:
         self,
         user_token: str | Callable[[], str] | None = None,
         filter: MCPToolFilter | None = None,
+        cache: CacheOptions | None = None,
     ) -> list[MCPTool]
 
     async def call_mcp_tool(
@@ -281,6 +312,21 @@ Both fields default to empty lists. `names` is applied after fetching; `ord_ids`
 
 > Both filter classes use AND semantics: if both fields are set, a result must match all of them to be included.
 
+### CacheOptions
+
+```python
+from sap_cloud_sdk.agentgateway import CacheOptions
+
+CacheOptions(
+    ttl=600.0,  # cache lifetime in seconds; default 600
+    max_size=32,  # max distinct cached entries (LRU eviction); default 32
+)
+```
+
+- `ttl`: How long a cached tool list is considered valid. After expiry the next call fetches fresh from the network.
+- `max_size`: Cap on how many distinct entries (filter + auth-type combinations) are held in memory. When exceeded, the least-recently-used entry is evicted.
+- `.evict()`: Clears all entries immediately, forcing a fresh fetch on the next call.
+
 ### Data Models
 
 ```python