ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add KeyValueResponseCacheStore for distributed client response caching (#4479)

Browse source 
* Add KeyValueResponseCacheStore adapter for client response cache

* Test KeyValueResponseCacheStore round-trip, isolation, and distributed sharing

* Document distributed response cache store
This commit is contained in:
Jeremiah Lowin 2026-07-17 16:41:35 -04:00 • committed by GitHub
commit 836ceac30e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 538 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

21
docs/clients/client.mdx
View file

@ -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.

15
docs/development/v4-notes/change-register.mdx
View file

@ -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)).

223
fastmcp_slim/fastmcp/client/caching.py Normal file
View file

@ -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

279
tests/client/client/test_kv_response_cache.py Normal file
View file

@ -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