|
| 1 | +"""Server-side caching hints (SEP-2549, protocol revision 2026-07-28). |
| 2 | +
|
| 3 | +Results for the cacheable methods carry `ttlMs`/`cacheScope` freshness hints. |
| 4 | +A handler sets them by returning a result with explicit `ttl_ms`/`cache_scope` |
| 5 | +values; `Server(cache_hints={method: CacheHint(...)})` fills them for handlers |
| 6 | +that don't. Fields the handler set win, per field, so a server-wide hint never |
| 7 | +overrides a handler's explicit choice. |
| 8 | +""" |
| 9 | + |
| 10 | +from __future__ import annotations |
| 11 | + |
| 12 | +from collections.abc import Mapping |
| 13 | +from dataclasses import dataclass |
| 14 | +from typing import Any, Final, Literal, TypeVar, get_args |
| 15 | + |
| 16 | +import mcp_types as types |
| 17 | + |
| 18 | +__all__ = ["CACHEABLE_METHODS", "CacheHint", "CacheableMethod", "apply_cache_hint", "validate_cache_hints"] |
| 19 | + |
| 20 | +CacheableMethod = Literal[ |
| 21 | + "prompts/list", |
| 22 | + "resources/list", |
| 23 | + "resources/read", |
| 24 | + "resources/templates/list", |
| 25 | + "server/discover", |
| 26 | + "tools/list", |
| 27 | +] |
| 28 | +"""The methods whose results carry `ttlMs`/`cacheScope`. Closed set: the spec |
| 29 | +defines caching hints on exactly these six (tests pin it to which result models |
| 30 | +mix in `CacheableResult`).""" |
| 31 | + |
| 32 | +CACHEABLE_METHODS: Final[frozenset[str]] = frozenset(get_args(CacheableMethod)) |
| 33 | +"""Runtime mirror of `CacheableMethod`, for callers the type checker can't see.""" |
| 34 | + |
| 35 | + |
| 36 | +@dataclass(frozen=True, slots=True) |
| 37 | +class CacheHint: |
| 38 | + """Freshness hint for one cacheable method's results. |
| 39 | +
|
| 40 | + `ttl_ms` is how long, in milliseconds, a client may consider the result |
| 41 | + fresh (`0` means immediately stale). `scope` is whether a cached result may |
| 42 | + be shared across authorization contexts (`"public"`) or only reused within |
| 43 | + the one that produced it (`"private"`). |
| 44 | + """ |
| 45 | + |
| 46 | + ttl_ms: int = 0 |
| 47 | + scope: Literal["public", "private"] = "private" |
| 48 | + |
| 49 | + def __post_init__(self) -> None: |
| 50 | + if self.ttl_ms < 0: |
| 51 | + raise ValueError(f"ttl_ms must be >= 0, got {self.ttl_ms}") |
| 52 | + if self.scope not in ("public", "private"): |
| 53 | + raise ValueError(f"scope must be 'public' or 'private', got {self.scope!r}") |
| 54 | + |
| 55 | + |
| 56 | +CacheableResultT = TypeVar("CacheableResultT", bound=types.CacheableResult) |
| 57 | + |
| 58 | + |
| 59 | +def apply_cache_hint(result: CacheableResultT, hint: CacheHint) -> CacheableResultT: |
| 60 | + """Fill `ttl_ms`/`cache_scope` on `result` from `hint`. |
| 61 | +
|
| 62 | + Per-field: a field the handler set explicitly - even to its default value, |
| 63 | + tracked via `model_fields_set` - is left alone; only unset fields take the |
| 64 | + hint. A handler constructing results with `model_construct` bypasses that |
| 65 | + tracking and is treated as having set nothing. |
| 66 | + """ |
| 67 | + update: dict[str, int | str] = {} |
| 68 | + if "ttl_ms" not in result.model_fields_set: |
| 69 | + update["ttl_ms"] = hint.ttl_ms |
| 70 | + if "cache_scope" not in result.model_fields_set: |
| 71 | + update["cache_scope"] = hint.scope |
| 72 | + return result.model_copy(update=update) if update else result |
| 73 | + |
| 74 | + |
| 75 | +def validate_cache_hints(cache_hints: Mapping[Any, Any] | None) -> dict[str, CacheHint]: |
| 76 | + """Validate a `cache_hints` constructor argument into a plain dict. |
| 77 | +
|
| 78 | + The `Server`/`MCPServer` signatures already close the key set and value |
| 79 | + type for type-checked callers; this runtime gate is deliberately loose in |
| 80 | + its parameter so it covers everyone else (e.g. a map deserialized from |
| 81 | + config) - a bad entry fails at construction, not on the first request to |
| 82 | + that method. |
| 83 | +
|
| 84 | + Raises: |
| 85 | + ValueError: If a key is not a cacheable method. |
| 86 | + TypeError: If a value is not a `CacheHint`. |
| 87 | + """ |
| 88 | + if cache_hints is None: |
| 89 | + return {} |
| 90 | + unknown = sorted(method for method in cache_hints if method not in CACHEABLE_METHODS) |
| 91 | + if unknown: |
| 92 | + raise ValueError(f"cache_hints keys must be cacheable methods (see CacheableMethod); got: {', '.join(unknown)}") |
| 93 | + validated: dict[str, CacheHint] = {} |
| 94 | + for method, hint in cache_hints.items(): |
| 95 | + if not isinstance(hint, CacheHint): |
| 96 | + raise TypeError(f"cache_hints[{method!r}] must be a CacheHint, got {type(hint).__name__}") |
| 97 | + validated[method] = hint |
| 98 | + return validated |
0 commit comments