Tighten response cache: session guard, identity and meta handling, refresh purge

Author: maxisbey

docs/advanced/caching.md

@@ -49,9 +49,13 @@ Four calls, three fetches. The second call found a fresh entry and never reached
 * `"refresh"` never serves: it fetches and stores the result, replacing whatever was cached.
 * `"bypass"` makes the round trip without touching the cache at all — no read, no write.
 
+One rule sits above `"use"`: **calls carrying `meta` always reach the server.** A request with `meta` set (a progress token, tracing fields) expects a wire request, so under `cache_mode="use"` it is treated as `"refresh"` — the cache read is skipped, and the fetched result still replaces the cached entry. `"bypass"` and an explicit `"refresh"` behave as they always do.
+
 To turn caching off entirely, construct with `Client(server, cache=False)`: every call is a round trip again, and `cache_mode`, while still accepted, does nothing.
 
-Scope is honored automatically too — `"private"` entries are keyed to the cache's *partition* (below); `"public"` ones may opt into wider sharing — and **notifications beat TTL**: a `list_changed` notification evicts the matching cached listing, and `resources/updated` evicts the cached read for its URI, however fresh they were.
+Scope is honored automatically too — `"private"` entries are keyed to the cache's *partition* (below); `"public"` ones may opt into wider sharing — and **notifications beat TTL** for the exact entries they name: a `list_changed` notification evicts the matching cached listing, and `resources/updated` evicts the cached read stored under exactly its URI, however fresh they were.
+
+One caveat on `resources/updated`: eviction is exact-URI only. The store contract has no enumerate or scan operation (same as the reference TypeScript implementation), so a notification carrying a *sub*-resource URI does not evict a cached read of its parent. If your server signals sub-resources this way, refetch the parent with `cache_mode="refresh"`.
 
 ### Configuring it: `CacheConfig`
 
@@ -61,7 +65,7 @@ from mcp.client import CacheConfig
 client = Client("https://api.example.com/mcp", cache=CacheConfig(default_ttl_ms=5_000))
 ```
 
-* `store` — where entries live. The default is a fresh in-memory store per client; pass your own `ResponseCacheStore` implementation (Redis-backed, say) to share a cache across clients or processes. A custom store **requires** an explicit `partition`.
+* `store` — where entries live. The default is a fresh in-memory store per client; pass your own `ResponseCacheStore` implementation (Redis-backed, say) to share a cache across clients or processes — the contract types (`ResponseCacheStore`, `CacheKey`, `CacheEntry`, and the default `InMemoryResponseCacheStore`) are importable from `mcp.client`. A lookup may issue up to two sequential store `get`s (the private arm, then the public one), so size a remote store's latency expectations accordingly. A custom store **requires** an explicit `partition`.
 * `partition` — the authorization-context label that keeps one principal's `"private"` entries from being served to another within a shared store.
 * `target_id` — explicit server identity, for custom transports and in-process servers (below).
 * `default_ttl_ms` — TTL applied to results that carry no `ttlMs` hint. The default `0` leaves hint-less results uncached.

src/mcp/client/__init__.py

@@ -2,17 +2,28 @@
 
 from mcp.client._input_required import InputRequiredRoundsExceededError
 from mcp.client._transport import Transport
-from mcp.client.caching import CacheConfig, CacheMode
+from mcp.client.caching import (
+    CacheConfig,
+    CacheEntry,
+    CacheKey,
+    CacheMode,
+    InMemoryResponseCacheStore,
+    ResponseCacheStore,
+)
 from mcp.client.client import Client
 from mcp.client.context import ClientRequestContext
 from mcp.client.session import ClientSession
 
 __all__ = [
     "CacheConfig",
+    "CacheEntry",
+    "CacheKey",
     "CacheMode",
     "Client",
     "ClientRequestContext",
     "ClientSession",
+    "InMemoryResponseCacheStore",
     "InputRequiredRoundsExceededError",
+    "ResponseCacheStore",
     "Transport",
 ]

src/mcp/client/caching.py

@@ -103,6 +103,10 @@ class ResponseCacheStore(Protocol):
     no rehydration hook to rebuild it from serialized data. An entry that
     comes back in the wrong shape (e.g. with a plain-dict value) degrades to
     a cache miss, never an error.
+
+    A cache lookup may issue up to two sequential `get` calls - the private
+    arm, then the public one - so remote-store implementers should size
+    latency expectations accordingly.
     """
 
     async def get(self, key: CacheKey) -> CacheEntry | None: ...
@@ -119,8 +123,8 @@ class CacheConfig:
     """Configuration for a `Client`'s response cache.
 
     Raises:
-        ValueError: If a custom `store` is given without a `partition`, or if
-            `default_ttl_ms` is negative.
+        ValueError: If a custom `store` is given without a `partition`, if
+            `target_id` is an empty string, or if `default_ttl_ms` is negative.
     """
 
     store: ResponseCacheStore | None = None
@@ -146,7 +150,9 @@ class CacheConfig:
 
     target_id: str | None = None
     """Explicit server-identity override, for custom transports and proxies
-    where the SDK cannot derive an identity from a server URL."""
+    where the SDK cannot derive an identity from a server URL. Must be
+    non-empty when provided - an empty string would collapse distinct servers
+    onto one identity."""
 
     default_ttl_ms: int = 0
     """Time-to-live, in milliseconds, applied to results that carry no `ttlMs`
@@ -172,6 +178,8 @@ class CacheConfig:
     def __post_init__(self) -> None:
         if self.store is not None and not self.partition:
             raise ValueError("a custom store requires an explicit partition")
+        if self.target_id == "":
+            raise ValueError("target_id must be a non-empty string or omitted")
         if self.default_ttl_ms < 0:
             raise ValueError(f"default_ttl_ms must be >= 0, got {self.default_ttl_ms}")
 
@@ -200,7 +208,14 @@ async def get(self, key: CacheKey) -> CacheEntry | None:
         return self._entries.get(key)
 
     async def set(self, key: CacheKey, entry: CacheEntry) -> None:
-        if self._max_read_entries and key.method == "resources/read" and key not in self._entries:
+        if (
+            self._max_read_entries
+            and key.method == "resources/read"
+            and key not in self._entries
+            # Strictly below the cap the read-key subset cannot be at the cap,
+            # so the scan only runs when an eviction is actually possible.
+            and len(self._entries) >= self._max_read_entries
+        ):
             # dict preserves insertion order and replacement keeps position, so
             # the dict itself is the FIFO ledger - no parallel structure to drift.
             read_keys = [k for k in self._entries if k.method == "resources/read"]
@@ -337,6 +352,12 @@ async def write(
         # Opposite arm first: a failed (or cancelled) delete aborts before the
         # set, leaving a miss - never two arms answering for one key.
         if not await self._delete(opposite):
+            # The fetch superseded whatever the own arm holds, so it must not
+            # keep serving either: best-effort delete it too (shielded like the
+            # other cleanup deletes), degrading the key to a full miss - no
+            # stale pair AND no superseded entry.
+            with anyio.CancelScope(shield=True):
+                await self._delete(own)
             return
         entry = CacheEntry(value=result.model_copy(deep=True), scope=scope, expires_at=self._clock() + ttl_ms / 1000)
         try:

src/mcp/client/client.py

@@ -464,6 +464,7 @@ async def _cached_fetch(
         method: str,
         *,
         cursor: str | None,
+        meta: RequestParamsMeta | None,
         cache_mode: CacheMode,
         send: Callable[[], Awaitable[_CacheableT]],
         absorb: Callable[[_CacheableT], _CacheableT] | None = None,
@@ -476,6 +477,15 @@ async def _cached_fetch(
         cache = self._response_cache
         if cache is None or cache_mode == "bypass":
             return await send()  # no read, no write, no eviction side-effects
+        # Cache participation requires a live session: a closed (or never-entered)
+        # client raises the no-context RuntimeError on every verb, exactly as the
+        # verbs did before the cache existed - never serving stale entries.
+        _ = self.session
+        if meta is not None and cache_mode == "use":
+            # A call carrying meta (a progress token, tracing fields) expects a
+            # wire request, so it is never served from the cache; the fetched
+            # result still replaces the entry, the same as an explicit refresh.
+            cache_mode = "refresh"
         if cursor is not None:
             # Continuation pages never read or write the (cursor-less) entry, but an
             # expired-cursor rejection signals the listing changed since the entry was
@@ -506,11 +516,13 @@ async def list_resources(
     ) -> ListResourcesResult:
         """List available resources from the server.
 
-        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`).
+        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`);
+        calls carrying `meta` always reach the server.
         """
         return await self._cached_fetch(
             "resources/list",
             cursor=cursor,
+            meta=meta,
             cache_mode=cache_mode,
             send=lambda: self.session.list_resources(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
         )
@@ -524,11 +536,13 @@ async def list_resource_templates(
     ) -> ListResourceTemplatesResult:
         """List available resource templates from the server.
 
-        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`).
+        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`);
+        calls carrying `meta` always reach the server.
         """
         return await self._cached_fetch(
             "resources/templates/list",
             cursor=cursor,
+            meta=meta,
             cache_mode=cache_mode,
             send=lambda: self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
         )
@@ -554,7 +568,8 @@ async def read_resource(
             input_responses: Responses to seed the first call with (e.g. when
                 resuming from a persisted `InputRequiredResult`).
             request_state: Opaque state to seed the first call with.
-            meta: Additional metadata for the request.
+            meta: Additional metadata for the request. Calls carrying `meta`
+                always reach the server.
             cache_mode: Adjusts the response cache's behavior for this call
                 (see `CacheMode`). Seeded calls (either `input_responses` or
                 `request_state` set) are resumptions of a multi-round-trip
@@ -581,6 +596,12 @@ async def retry(r: InputResponses | None, s: str | None) -> ReadResourceResult |
         cache = None if seeded else self._response_cache
         if cache is None or cache_mode == "bypass":
             return await self._drive_input_required(await retry(input_responses, request_state), retry)
+        # Cache participation requires a live session: a closed (or never-entered)
+        # client raises the no-context RuntimeError here, never serving stale entries.
+        _ = self.session
+        if meta is not None and cache_mode == "use":
+            # Calls carrying meta always reach the server (mirrors `_cached_fetch`).
+            cache_mode = "refresh"
         if cache_mode == "use" and (hit := await cache.read("resources/read", uri)) is not None:
             # InputRequiredResult is never stored (only terminal first-round results
             # are written below), so a hit is always terminal and legitimately skips
@@ -669,11 +690,13 @@ async def list_prompts(
     ) -> ListPromptsResult:
         """List available prompts from the server.
 
-        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`).
+        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`);
+        calls carrying `meta` always reach the server.
         """
         return await self._cached_fetch(
             "prompts/list",
             cursor=cursor,
+            meta=meta,
             cache_mode=cache_mode,
             send=lambda: self.session.list_prompts(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
         )
@@ -767,11 +790,13 @@ async def list_tools(
     ) -> ListToolsResult:
         """List available tools from the server.
 
-        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`).
+        `cache_mode` adjusts the response cache's behavior for this call (see `CacheMode`);
+        calls carrying `meta` always reach the server.
         """
         return await self._cached_fetch(
             "tools/list",
             cursor=cursor,
+            meta=meta,
             cache_mode=cache_mode,
             send=lambda: self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
             # A cache hit skips session.list_tools, so the session re-absorbs the

tests/client/test_caching.py

@@ -229,6 +229,15 @@ def test_a_custom_store_with_an_explicit_partition_constructs() -> None:
     assert config.partition == "token-subject-1"
 
 
+def test_an_empty_target_id_is_rejected_at_construction() -> None:
+    """SDK-defined guard: an explicit empty `target_id` would hash to the one
+    shared `sha256("")` identity, collapsing distinct servers onto it -
+    rejected at construction; omit the field (None) to derive an identity."""
+    with pytest.raises(ValueError) as exc:
+        CacheConfig(target_id="")
+    assert str(exc.value) == snapshot("target_id must be a non-empty string or omitted")
+
+
 def test_a_negative_default_ttl_is_rejected_at_construction() -> None:
     """SDK-defined guard: a negative configured TTL is a programming error,
     rejected at construction (negative `ttlMs` from the wire is tolerated as 0
@@ -457,6 +466,30 @@ async def clear(self) -> None:
         raise NotImplementedError
 
 
+class _ArmDeleteFailingStore:
+    """In-memory store whose `delete` raises only for keys on the given arm,
+    modelling a write whose opposite-arm cleanup fails while everything else
+    works. A write hitting that failure never reaches `set`."""
+
+    def __init__(self, failing_arm: str) -> None:
+        self.inner = InMemoryResponseCacheStore()
+        self.failing_arm = failing_arm
+
+    async def get(self, key: CacheKey) -> CacheEntry | None:
+        return await self.inner.get(key)
+
+    async def set(self, key: CacheKey, entry: CacheEntry) -> None:
+        raise NotImplementedError
+
+    async def delete(self, key: CacheKey) -> None:
+        if key.partition == self.failing_arm:
+            raise RuntimeError("store delete failed")
+        await self.inner.delete(key)
+
+    async def clear(self) -> None:
+        raise NotImplementedError
+
+
 class _RehydratingStore:
     """Models a persistent store whose `get` returns what its deserializer
     produced - possibly not the shape `set` received."""
@@ -832,6 +865,25 @@ async def test_a_raising_opposite_arm_delete_aborts_the_write() -> None:
     assert await store.inner.get(CacheKey("tools/list", "", _public_arm())) is None
 
 
+async def test_a_failed_opposite_arm_delete_degrades_the_key_to_a_full_miss() -> None:
+    """SDK error discipline: when only the opposite-arm delete fails, the write
+    cannot set its own arm (two arms might answer) - but the warm own-arm
+    entry was superseded by the fetch, so it is best-effort deleted too: both
+    arms read as misses, and the write itself never raises."""
+    store = _ArmDeleteFailingStore(failing_arm=_public_arm())
+    cache = _coordinator(store)
+    await store.inner.set(
+        CacheKey("tools/list", "", _private_arm()),
+        CacheEntry(value=_wire_result(), scope="private", expires_at=2_000_000.0),
+    )
+    assert await cache.read("tools/list", "") is not None  # the warm own-arm entry
+    gen = cache.capture("tools/list", "")
+    await cache.write("tools/list", "", _wire_result(ttl_ms=60_000), gen, "use")
+    assert await store.inner.get(CacheKey("tools/list", "", _private_arm())) is None
+    assert await store.inner.get(CacheKey("tools/list", "", _public_arm())) is None
+    assert await cache.read("tools/list", "") is None
+
+
 async def test_a_raising_store_set_caches_nothing_and_does_not_raise() -> None:
     """SDK error discipline: a `set` raise is logged and swallowed - the fetch
     already succeeded, the result just is not cached."""