diff --git a/docs/clients/client.mdx b/docs/clients/client.mdx index 30e09e613..a4141b71d 100644 --- a/docs/clients/client.mdx +++ b/docs/clients/client.mdx @@ -234,6 +234,27 @@ async with client: fresh = await client.list_tools_mcp(cache_mode="refresh") ``` +### Sharing a cache across clients + +The default cache lives in each client's process. To share cached responses across a fleet — a set of proxy replicas backed by one Redis, for example — pass a `KeyValueResponseCacheStore`, FastMCP's adapter over the same `AsyncKeyValue` key-value abstraction the event store and OAuth proxy use. It accepts any compatible backend (memory, Redis, and more). + +A shared store mingles responses from different principals, so it requires an explicit `partition` that isolates them. Derive the partition from a verified credential — never from request data or the server URL — and construct a new client when the principal changes. Only responses the server marks `"public"` are ever served across partitions. + +```python +from fastmcp import Client +from fastmcp.client.caching import KeyValueResponseCacheStore +from mcp.client.caching import CacheConfig +from key_value.aio.stores.redis import RedisStore + +backend = RedisStore(url="redis://localhost") +store = KeyValueResponseCacheStore(storage=backend) + +config = CacheConfig(store=store, partition="tenant-a", target_id="weather-api") +client = Client("https://example.com/mcp", mode="auto", cache=config) +``` + +The adapter serializes each result through a type-tagged envelope validated against an allowlist of cacheable result models, so a value naming an unknown type is treated as a cache miss rather than deserialized blindly. Each store instance owns its own collection namespace; `clear()` affects only that namespace, never another tenant's entries. + ## Operations FastMCP clients interact with three types of server components. diff --git a/docs/development/v4-notes/change-register.mdx b/docs/development/v4-notes/change-register.mdx index 6476d96cd..0b9e56e8b 100644 --- a/docs/development/v4-notes/change-register.mdx +++ b/docs/development/v4-notes/change-register.mdx @@ -219,6 +219,21 @@ Proxy forwarding handlers stash the request context so a backend that issues a s *Verify:* commit `1ac166bd`, `server/providers/proxy.py`. +### Shared response cache via `KeyValueResponseCacheStore` — New + +The SDK's client response cache (SEP-2549) reads and writes through a pluggable `ResponseCacheStore`; the default is a per-client in-memory LRU. FastMCP adds `KeyValueResponseCacheStore`, an adapter over the same `AsyncKeyValue` key-value abstraction the event store and OAuth proxy already use, so a fleet of clients (e.g. proxy replicas) can share one Redis-backed response cache. Pass it via `CacheConfig(store=...)`; a custom store requires an explicit `partition` (SDK) and `target_id` (FastMCP). Results serialize through a type-tagged envelope validated against an allowlist of cacheable result models — an unknown tag is a cache miss, never an import-by-name — and each adapter owns its own collection so `clear()` never touches another tenant. + +```python +from fastmcp.client.caching import KeyValueResponseCacheStore +from mcp.client.caching import CacheConfig +from key_value.aio.stores.redis import RedisStore + +store = KeyValueResponseCacheStore(storage=RedisStore(url="redis://localhost")) +config = CacheConfig(store=store, partition="tenant-a", target_id="weather-api") +``` + +*Verify:* `fastmcp_slim/fastmcp/client/caching.py`, `tests/client/client/test_kv_response_cache.py`. + ## HTTP The maintainer asked whether FastMCP can now delete its custom HTTP app and let the SDK's `Server.streamable_http_app()` handle everything. The answer for this PR is **no** — every override earns its keep. Convergence is a v4 project gated on three upstream additions (see [Feature Program](/development/v4-notes/feature-program)). diff --git a/fastmcp_slim/fastmcp/client/caching.py b/fastmcp_slim/fastmcp/client/caching.py new file mode 100644 index 000000000..75423819f --- /dev/null +++ b/fastmcp_slim/fastmcp/client/caching.py @@ -0,0 +1,223 @@ +"""A client response cache store backed by AsyncKeyValue. + +The MCP SDK's client response cache (SEP-2549) reads and writes through a +pluggable `ResponseCacheStore` protocol; the default is a per-client in-memory +LRU. This module adapts that protocol onto the `AsyncKeyValue` key-value +abstraction FastMCP already uses for its other state-management surfaces (the +event store, the OAuth proxy, the response-caching middleware), so a fleet of +FastMCP clients — for example a set of proxy replicas — can share one +Redis-backed response cache. + +Because a shared store mingles cached responses across principals, the SDK +requires an explicit `partition` on any custom store (and FastMCP additionally +requires a `target_id`). The partition is folded into every stored key so +entries can never collide or leak across authorization contexts, and the +adapter round-trips each result through a small type-tagged envelope validated +against an allowlist of cacheable result models — a stored value that names an +unknown type is treated as a miss, never imported by name. + +Example: + ```python + from fastmcp import Client + from fastmcp.client.caching import KeyValueResponseCacheStore + from mcp.client.caching import CacheConfig + from key_value.aio.stores.redis import RedisStore + + store = KeyValueResponseCacheStore(storage=RedisStore(url="redis://localhost")) + config = CacheConfig(store=store, partition="tenant-a", target_id="weather-api") + client = Client("https://example.com/mcp", mode="auto", cache=config) + ``` +""" + +from __future__ import annotations + +import hashlib +import time +from types import UnionType +from typing import Literal, get_args + +from key_value.aio.adapters.pydantic import PydanticAdapter +from key_value.aio.protocols import AsyncKeyValue +from key_value.aio.protocols.key_value import ( + AsyncDestroyCollectionProtocol, + AsyncEnumerateKeysProtocol, +) +from key_value.aio.stores.memory import MemoryStore +from mcp.client.caching import CacheEntry, CacheKey +from mcp_types import CacheableResult +from mcp_types.methods import MONOLITH_RESULTS + +from fastmcp.utilities.logging import get_logger +from fastmcp.utilities.types import FastMCPBaseModel + +logger = get_logger(__name__) + +DEFAULT_CACHE_COLLECTION = "fastmcp_response_cache" +"""Collection namespace owned by one adapter instance; `clear()` never reaches beyond it.""" + + +def _cacheable_result_models() -> dict[str, type[CacheableResult]]: + """Allowlist of `{class name: model}` for every cacheable result type. + + Derived from `MONOLITH_RESULTS` (the SDK's per-method result registry) so it + tracks the CACHEABLE_METHODS surface automatically. The class name is the + type tag written into the envelope; reconstruction looks the model up here + rather than importing an arbitrary name from store contents. + """ + models: dict[str, type[CacheableResult]] = {} + for row in MONOLITH_RESULTS.values(): + arms = get_args(row) if isinstance(row, UnionType) else (row,) + for arm in arms: + if isinstance(arm, type) and issubclass(arm, CacheableResult): + models[arm.__name__] = arm + return models + + +CACHEABLE_RESULT_MODELS = _cacheable_result_models() +"""Type tag -> model class allowlist for envelope reconstruction.""" + + +class _CacheEnvelope(FastMCPBaseModel): + """Serializable form of a `CacheEntry` for a remote store. + + A `CacheEntry.value` is a cacheable result model; a remote store cannot hold + it as an object, so it is serialized to `value_json` under a `type_tag` + (the model class name) and reconstructed against the allowlist on read. The + freshness/sharing metadata (`scope`, `expires_at`) round-trips alongside it. + """ + + type_tag: str + value_json: str + scope: str + expires_at: float | None + + +class KeyValueResponseCacheStore: + """A `ResponseCacheStore` backed by any `AsyncKeyValue` store. + + Implements the SDK client response cache contract (`get`/`set`/`delete`/ + `clear`) over the key-value abstraction FastMCP already uses elsewhere, so a + distributed deployment can point every client at one shared backend (memory, + Redis, etc.). Pass an instance as `CacheConfig(store=...)`; the SDK requires + an explicit `partition` on any custom store, and FastMCP additionally + requires a `target_id`. + + Each adapter instance owns one collection (`collection`), so `clear()` only + affects its own namespace and never another tenant's data. `clear()` needs + the backend to support collection destruction or key enumeration; against a + backend that supports neither it is a no-op and entries age out by TTL (a + warning is logged once). + + The SDK wraps every store call defensively — a raised operation degrades to + a cache miss rather than failing the request — so this adapter does not + re-wrap its own operations. + + Args: + storage: The `AsyncKeyValue` backend. Defaults to an in-process `MemoryStore`. + collection: Collection namespace for this adapter's entries. + """ + + def __init__( + self, + storage: AsyncKeyValue | None = None, + *, + collection: str = DEFAULT_CACHE_COLLECTION, + ) -> None: + self._storage: AsyncKeyValue = storage or MemoryStore() + self._collection = collection + self._adapter: PydanticAdapter[_CacheEnvelope] = PydanticAdapter[ + _CacheEnvelope + ]( + key_value=self._storage, + pydantic_model=_CacheEnvelope, + default_collection=collection, + ) + self._warned_clear_unsupported = False + + @staticmethod + def _string_key(key: CacheKey) -> str: + """Derive a stable store key from every `CacheKey` field. + + `CacheKey` is `(method, params_key, partition)`, where the coordinator has + already packed scope, negotiated protocol version, server arm id, and the + caller's partition into the `partition` field as a JSON array. Every field + is folded into the digest, so entries cannot collide across partitions, + protocol eras, or servers. The fields are length-prefixed before hashing + so no two distinct field tuples can produce the same pre-image. + """ + parts = [key.method, key.params_key, key.partition] + preimage = "".join(f"{len(part)}:{part}" for part in parts) + return hashlib.sha256(preimage.encode("utf-8")).hexdigest() + + async def get(self, key: CacheKey) -> CacheEntry | None: + envelope = await self._adapter.get(key=self._string_key(key)) + if envelope is None: + return None + model = CACHEABLE_RESULT_MODELS.get(envelope.type_tag) + if model is None: + # An unknown tag is never imported by name; a wrong-shape entry is a miss. + return None + value = model.model_validate_json(envelope.value_json) + scope: Literal["public", "private"] = ( + "public" if envelope.scope == "public" else "private" + ) + return CacheEntry(value=value, scope=scope, expires_at=envelope.expires_at) + + async def set(self, key: CacheKey, entry: CacheEntry) -> None: + value = entry.value + if not isinstance(value, CacheableResult): + return + type_tag = type(value).__name__ + if type_tag not in CACHEABLE_RESULT_MODELS: + return + envelope = _CacheEnvelope( + type_tag=type_tag, + value_json=value.model_dump_json(by_alias=True), + scope=entry.scope, + expires_at=entry.expires_at, + ) + ttl = self._entry_ttl(entry) + await self._adapter.put(key=self._string_key(key), value=envelope, ttl=ttl) + + async def delete(self, key: CacheKey) -> None: + await self._adapter.delete(key=self._string_key(key)) + + async def clear(self) -> None: + """Clear this adapter's collection only. + + Prefers deleting each enumerated key (which leaves the collection + usable), and falls back to whole-collection destruction. Against a + backend that supports neither, this is a no-op (entries age out by TTL) + and a warning is logged once. Either path is scoped to this adapter's + own collection, so a shared store's other tenants are never touched. + """ + storage = self._storage + if isinstance(storage, AsyncEnumerateKeysProtocol): + keys = await storage.keys(collection=self._collection) + for stored_key in keys: + await storage.delete(key=stored_key, collection=self._collection) + return + if isinstance(storage, AsyncDestroyCollectionProtocol): + await storage.destroy_collection(collection=self._collection) + return + if not self._warned_clear_unsupported: + self._warned_clear_unsupported = True + logger.warning( + "Response cache store backend %s supports neither collection " + "destruction nor key enumeration; clear() is a no-op and entries " + "will age out by TTL.", + type(storage).__name__, + ) + + def _entry_ttl(self, entry: CacheEntry) -> float | None: + """Seconds until the entry's own expiry, so the backend can evict it independently. + + The SDK gates freshness on `expires_at`, but a shared backend should not + retain a stale entry indefinitely; a store TTL lets it reclaim space. A + non-positive remaining lifetime stores with no backend TTL (the SDK will + still treat the already-stale entry as a miss). + """ + if entry.expires_at is None: + return None + remaining = entry.expires_at - time.time() + return remaining if remaining > 0 else None diff --git a/tests/client/client/test_kv_response_cache.py b/tests/client/client/test_kv_response_cache.py new file mode 100644 index 000000000..108bf3e60 --- /dev/null +++ b/tests/client/client/test_kv_response_cache.py @@ -0,0 +1,279 @@ +"""`KeyValueResponseCacheStore`: the AsyncKeyValue-backed client response cache store. + +The SDK's client response cache (SEP-2549) reads and writes through a pluggable +`ResponseCacheStore`. `KeyValueResponseCacheStore` adapts that contract onto the +`AsyncKeyValue` abstraction FastMCP already uses for its other state surfaces, so +a fleet of clients can share one backend (memory, Redis, etc.). + +These tests cover the store in isolation (round-trip, partition isolation, +clear, allowlist) and end-to-end: two independent `fastmcp.Client` instances +sharing one adapter-backed store, where the second client serves the first's +cached `tools/list` with zero wire calls. +""" + +from __future__ import annotations + +import json +import time + +import pytest +from key_value.aio.stores.memory import MemoryStore +from mcp.client.caching import CacheConfig, CacheEntry, CacheKey +from mcp.server.caching import CacheHint +from mcp.server.mcpserver import MCPServer +from mcp_types import ListToolsResult, Tool + +from fastmcp import Client, FastMCP +from fastmcp.client.caching import ( + CACHEABLE_RESULT_MODELS, + KeyValueResponseCacheStore, + _CacheEnvelope, +) + + +def _tools_result() -> ListToolsResult: + return ListToolsResult( + tools=[Tool(name="add", input_schema={"type": "object"})], + ttl_ms=60000, + cache_scope="public", + ) + + +def _key( + partition: str, *, method: str = "tools/list", params_key: str = "" +) -> CacheKey: + # The coordinator packs scope/version/arm into CacheKey.partition as a JSON + # array; mirror that shape so the derived string key is realistic. + arm = json.dumps(["public", "2026-07-28", "srv", partition]) + return CacheKey(method, params_key, arm) + + +def _cached_server(ttl_ms: int = 60000) -> MCPServer: + """An SDK MCPServer whose tools/list carries a positive ttlMs hint at 2026.""" + server = MCPServer( + "cached", cache_hints={"tools/list": CacheHint(ttl_ms=ttl_ms, scope="public")} + ) + + @server.tool() + def add(a: int, b: int) -> int: + return a + b + + return server + + +class TestRoundTrip: + async def test_set_get_reconstructs_model(self): + """A stored entry round-trips back to an equal result model object.""" + store = KeyValueResponseCacheStore() + result = _tools_result() + key = _key("p1") + + await store.set( + key, CacheEntry(value=result, scope="public", expires_at=time.time() + 60) + ) + got = await store.get(key) + + assert got is not None + assert isinstance(got.value, ListToolsResult) + assert got.value == result + assert got.scope == "public" + + async def test_get_miss_returns_none(self): + store = KeyValueResponseCacheStore() + assert await store.get(_key("p1")) is None + + async def test_delete_removes_entry(self): + store = KeyValueResponseCacheStore() + key = _key("p1") + await store.set( + key, + CacheEntry( + value=_tools_result(), scope="public", expires_at=time.time() + 60 + ), + ) + await store.delete(key) + assert await store.get(key) is None + + async def test_private_scope_roundtrips(self): + store = KeyValueResponseCacheStore() + key = _key("p1") + result = ListToolsResult( + tools=[Tool(name="add", input_schema={"type": "object"})] + ) + await store.set( + key, CacheEntry(value=result, scope="private", expires_at=time.time() + 60) + ) + got = await store.get(key) + assert got is not None + assert got.scope == "private" + + +class TestPartitionIsolation: + async def test_two_partitions_do_not_bleed(self): + """Entries written under different CacheKey.partition arms never collide.""" + store = KeyValueResponseCacheStore() + result = _tools_result() + + await store.set( + _key("tenant-a"), + CacheEntry(value=result, scope="public", expires_at=time.time() + 60), + ) + + # A different partition is a distinct key: a clean miss, not a shared hit. + assert await store.get(_key("tenant-b")) is None + assert await store.get(_key("tenant-a")) is not None + + async def test_method_and_params_key_isolate(self): + """Distinct method / params_key never collide in the derived string key.""" + store = KeyValueResponseCacheStore() + result = _tools_result() + await store.set( + _key("p1", method="resources/read", params_key="file:///a"), + CacheEntry(value=result, scope="public", expires_at=time.time() + 60), + ) + assert ( + await store.get(_key("p1", method="resources/read", params_key="file:///b")) + is None + ) + assert await store.get(_key("p1", method="tools/list")) is None + + +class TestClear: + async def test_clear_empties_and_keeps_collection_usable(self): + store = KeyValueResponseCacheStore() + key = _key("p1") + await store.set( + key, + CacheEntry( + value=_tools_result(), scope="public", expires_at=time.time() + 60 + ), + ) + + await store.clear() + assert await store.get(key) is None + + # The collection remains usable for subsequent writes. + await store.set( + key, + CacheEntry( + value=_tools_result(), scope="public", expires_at=time.time() + 60 + ), + ) + assert await store.get(key) is not None + + async def test_clear_scoped_to_own_collection(self): + """Two adapters over one backend clear independently.""" + backend = MemoryStore() + store_a = KeyValueResponseCacheStore(backend, collection="cache_a") + store_b = KeyValueResponseCacheStore(backend, collection="cache_b") + key = _key("p1") + entry = CacheEntry( + value=_tools_result(), scope="public", expires_at=time.time() + 60 + ) + + await store_a.set(key, entry) + await store_b.set(key, entry) + + await store_a.clear() + + assert await store_a.get(key) is None + assert await store_b.get(key) is not None # untouched + + +class TestAllowlist: + async def test_unknown_type_tag_is_a_miss(self): + """A stored envelope naming a type outside the allowlist is a miss, never imported.""" + store = KeyValueResponseCacheStore() + key = _key("p1") + forged = _CacheEnvelope( + type_tag="EvilResult", + value_json="{}", + scope="public", + expires_at=time.time() + 60, + ) + await store._adapter.put(key=store._string_key(key), value=forged) + + assert await store.get(key) is None + + def test_allowlist_matches_cacheable_methods(self): + """The allowlist covers exactly the SDK's cacheable result models.""" + assert set(CACHEABLE_RESULT_MODELS) == { + "DiscoverResult", + "ListPromptsResult", + "ListResourceTemplatesResult", + "ListResourcesResult", + "ListToolsResult", + "ReadResourceResult", + } + + +class TestFastMCPConstruction: + def test_custom_store_without_target_id_raises(self): + """FastMCP requires a target_id for a custom shared store on an in-memory transport.""" + store = KeyValueResponseCacheStore() + with pytest.raises(ValueError, match="requires CacheConfig.target_id"): + Client(FastMCP("x"), cache=CacheConfig(store=store, partition="p")) + + def test_custom_store_without_partition_raises(self): + """The SDK requires an explicit partition for any custom store.""" + with pytest.raises(ValueError, match="requires an explicit partition"): + CacheConfig(store=KeyValueResponseCacheStore(), target_id="srv") + + def test_custom_store_builds_cache(self): + store = KeyValueResponseCacheStore() + config = CacheConfig(store=store, partition="p", target_id="srv") + client = Client(FastMCP("x"), mode="auto", cache=config) + assert client._response_cache is not None + + +class TestDistributedSharing: + async def test_second_client_serves_first_clients_cache(self): + """Two independent Clients sharing one adapter-backed store: client B's first + list_tools is served from the entry client A populated, with zero wire calls.""" + backend = MemoryStore() + store = KeyValueResponseCacheStore(backend) + + def make_client() -> Client: + config = CacheConfig(store=store, partition="tenant-a", target_id="cached") + return Client(_cached_server(), mode="auto", cache=config) + + async with make_client() as client_a: + first = await client_a.list_tools() + assert [t.name for t in first] == ["add"] + + async with make_client() as client_b: + calls = {"n": 0} + original = client_b.session.list_tools + + async def spy(**kwargs): + calls["n"] += 1 + return await original(**kwargs) + + client_b.session.list_tools = spy # type: ignore[method-assign] + served = await client_b.list_tools() + + assert calls["n"] == 0 # served from the shared store, no wire round-trip + assert [t.name for t in served] == ["add"] + + async def test_distinct_partitions_do_not_share(self): + """Two clients on the same store but different partitions each hit the wire.""" + backend = MemoryStore() + store = KeyValueResponseCacheStore(backend) + + config_a = CacheConfig(store=store, partition="tenant-a", target_id="cached") + async with Client(_cached_server(), mode="auto", cache=config_a) as client_a: + await client_a.list_tools() + + config_b = CacheConfig(store=store, partition="tenant-b", target_id="cached") + async with Client(_cached_server(), mode="auto", cache=config_b) as client_b: + calls = {"n": 0} + original = client_b.session.list_tools + + async def spy(**kwargs): + calls["n"] += 1 + return await original(**kwargs) + + client_b.session.list_tools = spy # type: ignore[method-assign] + await client_b.list_tools() + + assert calls["n"] == 1 # different partition -> not shared, hits the wire