Add cache_hints constructor map for SEP-2549 caching hints (#3015)

src/mcp/server/__init__.py

@@ -1,6 +1,7 @@
+from .caching import CacheHint
 from .context import ServerRequestContext
 from .lowlevel import NotificationOptions, Server
 from .mcpserver import MCPServer
 from .models import InitializationOptions
 
-__all__ = ["Server", "ServerRequestContext", "MCPServer", "NotificationOptions", "InitializationOptions"]
+__all__ = ["CacheHint", "Server", "ServerRequestContext", "MCPServer", "NotificationOptions", "InitializationOptions"]

src/mcp/server/caching.py

@@ -0,0 +1,98 @@
+"""Server-side caching hints (SEP-2549, protocol revision 2026-07-28).
+
+Results for the cacheable methods carry `ttlMs`/`cacheScope` freshness hints.
+A handler sets them by returning a result with explicit `ttl_ms`/`cache_scope`
+values; `Server(cache_hints={method: CacheHint(...)})` fills them for handlers
+that don't. Fields the handler set win, per field, so a server-wide hint never
+overrides a handler's explicit choice.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any, Final, Literal, TypeVar, get_args
+
+import mcp_types as types
+
+__all__ = ["CACHEABLE_METHODS", "CacheHint", "CacheableMethod", "apply_cache_hint", "validate_cache_hints"]
+
+CacheableMethod = Literal[
+    "prompts/list",
+    "resources/list",
+    "resources/read",
+    "resources/templates/list",
+    "server/discover",
+    "tools/list",
+]
+"""The methods whose results carry `ttlMs`/`cacheScope`. Closed set: the spec
+defines caching hints on exactly these six (tests pin it to which result models
+mix in `CacheableResult`)."""
+
+CACHEABLE_METHODS: Final[frozenset[str]] = frozenset(get_args(CacheableMethod))
+"""Runtime mirror of `CacheableMethod`, for callers the type checker can't see."""
+
+
+@dataclass(frozen=True, slots=True)
+class CacheHint:
+    """Freshness hint for one cacheable method's results.
+
+    `ttl_ms` is how long, in milliseconds, a client may consider the result
+    fresh (`0` means immediately stale). `scope` is whether a cached result may
+    be shared across authorization contexts (`"public"`) or only reused within
+    the one that produced it (`"private"`).
+    """
+
+    ttl_ms: int = 0
+    scope: Literal["public", "private"] = "private"
+
+    def __post_init__(self) -> None:
+        if self.ttl_ms < 0:
+            raise ValueError(f"ttl_ms must be >= 0, got {self.ttl_ms}")
+        if self.scope not in ("public", "private"):
+            raise ValueError(f"scope must be 'public' or 'private', got {self.scope!r}")
+
+
+CacheableResultT = TypeVar("CacheableResultT", bound=types.CacheableResult)
+
+
+def apply_cache_hint(result: CacheableResultT, hint: CacheHint) -> CacheableResultT:
+    """Fill `ttl_ms`/`cache_scope` on `result` from `hint`.
+
+    Per-field: a field the handler set explicitly - even to its default value,
+    tracked via `model_fields_set` - is left alone; only unset fields take the
+    hint. A handler constructing results with `model_construct` bypasses that
+    tracking and is treated as having set nothing.
+    """
+    update: dict[str, int | str] = {}
+    if "ttl_ms" not in result.model_fields_set:
+        update["ttl_ms"] = hint.ttl_ms
+    if "cache_scope" not in result.model_fields_set:
+        update["cache_scope"] = hint.scope
+    return result.model_copy(update=update) if update else result
+
+
+def validate_cache_hints(cache_hints: Mapping[Any, Any] | None) -> dict[str, CacheHint]:
+    """Validate a `cache_hints` constructor argument into a plain dict.
+
+    The `Server`/`MCPServer` signatures already close the key set and value
+    type for type-checked callers; this runtime gate is deliberately loose in
+    its parameter so it covers everyone else (e.g. a map deserialized from
+    config) - a bad entry fails at construction, not on the first request to
+    that method.
+
+    Raises:
+        ValueError: If a key is not a cacheable method.
+        TypeError: If a value is not a `CacheHint`.
+    """
+    if cache_hints is None:
+        return {}
+    unknown = sorted(method for method in cache_hints if method not in CACHEABLE_METHODS)
+    if unknown:
+        raise ValueError(f"cache_hints keys must be cacheable methods (see CacheableMethod); got: {', '.join(unknown)}")
+    validated: dict[str, CacheHint] = {}
+    for method, hint in cache_hints.items():
+        if not isinstance(hint, CacheHint):
+            raise TypeError(f"cache_hints[{method!r}] must be a CacheHint, got {type(hint).__name__}")
+        validated[method] = hint
+    return validated

src/mcp/server/lowlevel/server.py

@@ -38,7 +38,7 @@ from __future__ import annotations
 
 import logging
 import warnings
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable, Mapping
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from dataclasses import dataclass
 from importlib.metadata import version as importlib_version
@@ -59,6 +59,7 @@ from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAut
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
 from mcp.server.auth.routes import build_resource_metadata_url, create_auth_routes, create_protected_resource_routes
 from mcp.server.auth.settings import AuthSettings
+from mcp.server.caching import CacheableMethod, CacheHint, validate_cache_hints
 from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
 from mcp.server.runner import serve_loop
@@ -140,6 +141,7 @@ class Server(Generic[LifespanResultT]):
         instructions: str | None = None,
         website_url: str | None = None,
         icons: list[types.Icon] | None = None,
+        cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
         lifespan: Callable[
             [Server[LifespanResultT]],
             AbstractAsyncContextManager[LifespanResultT],
@@ -222,6 +224,7 @@ class Server(Generic[LifespanResultT]):
         instructions: str | None = None,
         website_url: str | None = None,
         icons: list[types.Icon] | None = None,
+        cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
         lifespan: Callable[
             [Server[LifespanResultT]],
             AbstractAsyncContextManager[LifespanResultT],
@@ -313,6 +316,7 @@ class Server(Generic[LifespanResultT]):
         instructions: str | None = None,
         website_url: str | None = None,
         icons: list[types.Icon] | None = None,
+        cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
         lifespan: Callable[
             [Server[LifespanResultT]],
             AbstractAsyncContextManager[LifespanResultT],
@@ -420,6 +424,9 @@ class Server(Generic[LifespanResultT]):
         self.instructions = instructions
         self.website_url = website_url
         self.icons = icons
+        # Per-method `ttl_ms`/`cache_scope` fills, applied by `ServerRunner`
+        # after the handler returns; fields the handler set explicitly win.
+        self.cache_hints: dict[str, CacheHint] = validate_cache_hints(cache_hints)
         self.lifespan = lifespan
         self._request_handlers: dict[str, HandlerEntry[LifespanResultT]] = {}
         self._notification_handlers: dict[str, HandlerEntry[LifespanResultT]] = {}

src/mcp/server/mcpserver/server.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import base64
 import inspect
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Any, Generic, Literal, TypeVar, overload
 
@@ -58,6 +58,7 @@ from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
 from mcp.server.auth.settings import AuthSettings
+from mcp.server.caching import CacheableMethod, CacheHint
 from mcp.server.context import HandlerResult, ServerRequestContext
 from mcp.server.extension import (
     Extension,
@@ -169,6 +170,7 @@ class MCPServer(Generic[LifespanResultT]):
         lifespan: Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None = None,
         auth: AuthSettings | None = None,
         resource_security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
+        cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
     ):
         self._resource_security = resource_security
         self.settings = Settings(
@@ -196,6 +198,7 @@ class MCPServer(Generic[LifespanResultT]):
             website_url=website_url,
             icons=icons,
             version=version,
+            cache_hints=cache_hints,
             on_list_tools=self._handle_list_tools,
             on_call_tool=self._handle_call_tool,
             on_list_resources=self._handle_list_resources,

src/mcp/server/runner.py

@@ -28,6 +28,7 @@ from mcp_types import (
     INVALID_PARAMS,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
+    CacheableResult,
     ErrorData,
     Implementation,
     InitializeRequestParams,
@@ -40,6 +41,7 @@ from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS, LATEST_HANDSHAKE_VERS
 from pydantic import BaseModel, ValidationError
 from typing_extensions import TypeVar
 
+from mcp.server.caching import apply_cache_hint
 from mcp.server.connection import Connection
 from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
@@ -196,6 +198,12 @@ class ServerRunner(Generic[LifespanT]):
             if isinstance(result, ErrorData):
                 # Raise inside the chain so middleware observes the failure.
                 raise MCPError.from_error_data(result)
+            # Fill cache hints on the typed result, before the serialize sieve
+            # decides whether the negotiated version carries the fields at all.
+            # `input_required` interim results are not `CacheableResult` models,
+            # so the MRTR carve-out (no hints on them) holds by shape.
+            if isinstance(result, CacheableResult) and (hint := self.server.cache_hints.get(method)) is not None:
+                result = apply_cache_hint(result, hint)
             # Dump and serialize inside the chain so the OpenTelemetry span (the
             # outermost middleware) records a failing handler return shape too.
             return self._serialize(method, version, result)