feat(mcp): speak the 2026-07-28 stateless protocol

Phase 2 of the MCP 2026-07-28 migration (#69931), on top of the SDK 2.x
migration (#88180):

- Protocol-era negotiation (_negotiate_session): per-server `protocol`
  config key — auto (default, handshake-first with server/discover
  fallback on -32022/-32601), stateless (discover-first), legacy
  (handshake only). Auto is handshake-first deliberately: zero extra
  round-trips and zero behavior change for the entire existing server
  fleet, while 2026-07-28-only servers now connect via the fallback.
  All four transport call sites (stdio, SSE, new HTTP, legacy HTTP)
  route through the one choke point, so the CLI/desktop probe path
  inherits it too.
- SEP-2549 list caching: tools/list ttlMs/cacheScope hints are captured
  during discovery and bound to the lazy-startup schema cache — TTL'd
  entries expire and force a live re-probe; hint-less (pre-2026)
  servers keep the never-expires behavior. Pagination continuation now
  speaks both SDK generations (params= vs cursor=).
- SEP-837: OAuth client metadata declares application_type=native
  (config-overridable), with a fallback for 1.x-era metadata models.
  (RFC 9207 iss validation and SEP-2352 issuer-keyed credentials are
  native to SDK 2.0's OAuthClientProvider — verified, no client-side
  gap.)
- SEP-2577 deprecation posture: SamplingHandler docstring marks the
  Sampling feature as upstream-deprecated (12-month window) — kept
  fully functional, closed to new capability.
- Docs: `protocol` key in the MCP config reference.

tests/tools/test_mcp_protocol_negotiation.py

@@ -0,0 +1,144 @@
+"""MCP 2026-07-28 protocol-era negotiation (_negotiate_session).
+
+The negotiation helper decides between the legacy ``initialize`` handshake
+and the stateless ``server/discover`` probe (SEP-2575) per the per-server
+``protocol`` config key. These tests drive it with duck-typed sessions —
+the live-path integration is covered by the real-server E2E in the PR.
+"""
+
+import asyncio
+
+import pytest
+
+from tools.mcp_tool import (
+    MCPServerTask,
+    _handshake_rejected_as_modern,
+    _JSONRPC_UNSUPPORTED_PROTOCOL_VERSION,
+)
+
+
+class _Err(Exception):
+    def __init__(self, code, msg="err"):
+        super().__init__(msg)
+        self.error = type("E", (), {"code": code})()
+
+
+class _Session:
+    def __init__(self, init=None, disc=None):
+        self._init = init
+        self._disc = disc
+        self.calls = []
+
+    async def initialize(self):
+        self.calls.append("initialize")
+        if isinstance(self._init, Exception):
+            raise self._init
+        return self._init
+
+    async def discover(self):
+        self.calls.append("discover")
+        if isinstance(self._disc, Exception):
+            raise self._disc
+        return self._disc
+
+
+class _LegacySession:
+    """mcp 1.x sessions have no discover() attribute at all."""
+
+    def __init__(self, init=None):
+        self._init = init
+        self.calls = []
+
+    async def initialize(self):
+        self.calls.append("initialize")
+        if isinstance(self._init, Exception):
+            raise self._init
+        return self._init
+
+
+def _task(protocol=None):
+    t = MCPServerTask("negotest")
+    t._config = {} if protocol is None else {"protocol": protocol}
+    return t
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+class TestAutoMode:
+    def test_handshake_first_no_discover_on_success(self):
+        s = _Session(init="INIT_RESULT")
+        out = _run(_task()._negotiate_session(s, 5))
+        assert out == "INIT_RESULT"
+        assert s.calls == ["initialize"]
+
+    def test_falls_back_to_discover_on_unsupported_protocol_version(self):
+        s = _Session(init=_Err(_JSONRPC_UNSUPPORTED_PROTOCOL_VERSION), disc="DISC_RESULT")
+        out = _run(_task()._negotiate_session(s, 5))
+        assert out == "DISC_RESULT"
+        assert s.calls == ["initialize", "discover"]
+
+    def test_falls_back_on_method_not_found(self):
+        s = _Session(init=_Err(-32601, "Method not found: initialize"), disc="DISC_RESULT")
+        out = _run(_task()._negotiate_session(s, 5))
+        assert out == "DISC_RESULT"
+
+    def test_unrelated_error_propagates_without_discover(self):
+        s = _Session(init=_Err(-32000, "borked"))
+        with pytest.raises(_Err):
+            _run(_task()._negotiate_session(s, 5))
+        assert s.calls == ["initialize"]
+
+    def test_timeout_propagates_not_swallowed(self):
+        class _Hang(_Session):
+            async def initialize(self):
+                await asyncio.sleep(30)
+
+        with pytest.raises(asyncio.TimeoutError):
+            _run(_task()._negotiate_session(_Hang(), 0.05))
+
+
+class TestExplicitModes:
+    def test_stateless_probes_discover_first(self):
+        s = _Session(init="INIT_RESULT", disc="DISC_RESULT")
+        out = _run(_task("stateless")._negotiate_session(s, 5))
+        assert out == "DISC_RESULT"
+        assert s.calls == ["discover"]
+
+    def test_stateless_falls_back_to_handshake(self):
+        s = _Session(init="INIT_RESULT", disc=_Err(-32601))
+        out = _run(_task("stateless")._negotiate_session(s, 5))
+        assert out == "INIT_RESULT"
+        assert s.calls == ["discover", "initialize"]
+
+    def test_legacy_never_discovers(self):
+        s = _Session(init=_Err(_JSONRPC_UNSUPPORTED_PROTOCOL_VERSION), disc="DISC_RESULT")
+        with pytest.raises(_Err):
+            _run(_task("legacy")._negotiate_session(s, 5))
+        assert s.calls == ["initialize"]
+
+    def test_unknown_mode_treated_as_auto(self):
+        s = _Session(init="INIT_RESULT")
+        out = _run(_task("bogus")._negotiate_session(s, 5))
+        assert out == "INIT_RESULT"
+
+    def test_legacy_sdk_session_without_discover_reraises(self):
+        # mcp 1.x ClientSession has no .discover(): the auto fallback must
+        # re-raise the original handshake error, not AttributeError.
+        s = _LegacySession(init=_Err(_JSONRPC_UNSUPPORTED_PROTOCOL_VERSION))
+        with pytest.raises(_Err):
+            _run(_task()._negotiate_session(s, 5))
+        assert s.calls == ["initialize"]
+
+
+class TestModernRejectionClassifier:
+    def test_structural_codes(self):
+        assert _handshake_rejected_as_modern(_Err(-32022))
+        assert _handshake_rejected_as_modern(_Err(-32601))
+        assert not _handshake_rejected_as_modern(_Err(-32000))
+
+    def test_substring_fallbacks(self):
+        assert _handshake_rejected_as_modern(Exception("Unsupported protocol version"))
+        assert _handshake_rejected_as_modern(Exception("Unknown method: initialize"))
+        assert not _handshake_rejected_as_modern(Exception("connection reset by peer"))

tests/tools/test_mcp_schema_cache_ttl.py

@@ -0,0 +1,51 @@
+"""SEP-2549 schema-cache TTL expiry (tools/mcp_schema_cache.py)."""
+
+import time
+
+import pytest
+
+from tools import mcp_schema_cache as sc
+
+
+@pytest.fixture(autouse=True)
+def _isolated_cache(tmp_path, monkeypatch):
+    monkeypatch.setattr(sc, "_cache_path", lambda: tmp_path / "cache.json")
+    yield
+
+
+def test_entry_without_ttl_never_expires():
+    sc.write_cache_entry("srv", "fp", tools=[{"name": "t"}])
+    assert sc.get_cached_entry("srv", "fp") is not None
+
+
+def test_entry_within_ttl_served():
+    sc.write_cache_entry("srv", "fp", tools=[{"name": "t"}], ttl_ms=60_000)
+    entry = sc.get_cached_entry("srv", "fp")
+    assert entry is not None
+    assert entry["ttl_ms"] == 60_000
+    assert "written_at" in entry
+
+
+def test_entry_past_ttl_is_a_miss(monkeypatch):
+    sc.write_cache_entry("srv", "fp", tools=[{"name": "t"}], ttl_ms=1_000)
+    real_time = time.time
+    monkeypatch.setattr(sc.time, "time", lambda: real_time() + 2.0)
+    assert sc.get_cached_entry("srv", "fp") is None
+
+
+def test_ttl_rewrite_advances_written_at():
+    sc.write_cache_entry("srv", "fp", tools=[{"name": "t"}], ttl_ms=60_000)
+    first = sc.get_cached_entry("srv", "fp")["written_at"]
+    time.sleep(0.01)
+    # Identical payload would previously short-circuit; TTL'd entries must
+    # rewrite so written_at advances on every live reconfirmation.
+    sc.write_cache_entry("srv", "fp", tools=[{"name": "t"}], ttl_ms=60_000)
+    second = sc.get_cached_entry("srv", "fp")["written_at"]
+    assert second > first
+
+
+def test_cache_scope_round_trips():
+    sc.write_cache_entry(
+        "srv", "fp", tools=[{"name": "t"}], ttl_ms=60_000, cache_scope="private"
+    )
+    assert sc.get_cached_entry("srv", "fp")["cache_scope"] == "private"

tools/mcp_oauth.py

@@ -1364,11 +1364,24 @@ def _build_client_metadata(cfg: dict) -> "OAuthClientMetadata":
         "grant_types": ["authorization_code", "refresh_token"],
         "response_types": ["code"],
         "token_endpoint_auth_method": auth_method,
+        # SEP-837 (2026-07-28 spec): clients MUST declare an application_type
+        # during registration so OIDC-strict authorization servers stop
+        # rejecting loopback redirect_uris. Hermes is a CLI/desktop app
+        # redirecting to 127.0.0.1/localhost — that is exactly "native".
+        # Overridable for the rare hosted-dashboard deployment fronting a
+        # real https redirect.
+        "application_type": cfg.get("application_type", "native"),
     }
     if scope:
         metadata_kwargs["scope"] = scope
 
-    return OAuthClientMetadata.model_validate(metadata_kwargs)
+    try:
+        return OAuthClientMetadata.model_validate(metadata_kwargs)
+    except Exception:
+        # mcp 1.x metadata models predate SEP-837 and reject the unknown
+        # field — retry without it rather than failing the whole flow.
+        metadata_kwargs.pop("application_type", None)
+        return OAuthClientMetadata.model_validate(metadata_kwargs)
 
 
 def _invalidate_tokens_on_client_change(

tools/mcp_schema_cache.py

@@ -12,6 +12,7 @@ import hashlib
 import json
 import logging
 import threading
+import time
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
@@ -64,13 +65,27 @@ def _save_all(data: Dict[str, Any]) -> None:
 
 
 def get_cached_entry(server_name: str, fingerprint: str) -> Optional[dict]:
-    """Return cached entry when fingerprint matches, else None."""
+    """Return cached entry when fingerprint matches (and TTL holds), else None.
+
+    MCP 2026-07-28 (SEP-2549): ``tools/list`` results carry ``ttlMs`` as a
+    freshness hint. When the live discovery path recorded one, an entry
+    older than its TTL is treated as a miss so the next startup re-probes
+    the server instead of serving a stale manifest forever. Entries without
+    a recorded TTL (pre-2026 servers) keep the old never-expires behavior.
+    ``cacheScope`` is irrelevant here: this cache is per-user local disk,
+    which satisfies even ``private``.
+    """
     with _cache_lock:
         entry = _load_all().get(server_name)
     if not isinstance(entry, dict):
         return None
     if entry.get("fingerprint") != fingerprint:
         return None
+    ttl_ms = entry.get("ttl_ms")
+    written_at = entry.get("written_at")
+    if isinstance(ttl_ms, (int, float)) and isinstance(written_at, (int, float)):
+        if (time.time() - written_at) * 1000.0 >= float(ttl_ms):
+            return None
     return entry
 
 
@@ -84,19 +99,34 @@ def write_cache_entry(
     *,
     tools: List[dict],
     utility_tools: Optional[List[dict]] = None,
+    ttl_ms: Optional[float] = None,
+    cache_scope: Optional[str] = None,
 ) -> None:
-    """Persist tool schemas after a successful live connect."""
+    """Persist tool schemas after a successful live connect.
+
+    ``ttl_ms``/``cache_scope`` are the SEP-2549 hints from the server's
+    ``tools/list`` result (2026-07-28 servers). ``written_at`` anchors TTL
+    expiry in :func:`get_cached_entry`.
+    """
     entry = {
         "fingerprint": fingerprint,
         "tools": tools,
         "utility_tools": utility_tools or [],
     }
+    if isinstance(ttl_ms, (int, float)):
+        entry["ttl_ms"] = ttl_ms
+        entry["written_at"] = time.time()
+    if cache_scope:
+        entry["cache_scope"] = cache_scope
     with _cache_lock:
         data = _load_all()
         # Write-through fires on every registration (reconnects,
         # list_changed refreshes); skip the load-all+rewrite churn when the
-        # entry is byte-identical to what is already on disk.
-        if data.get(server_name) == entry:
+        # entry is byte-identical to what is already on disk. TTL'd entries
+        # always rewrite: written_at must advance or the entry would expire
+        # at its ORIGINAL write time no matter how many live reconnects
+        # confirmed it since.
+        if "written_at" not in entry and data.get(server_name) == entry:
             return
         data[server_name] = entry
         _save_all(data)

tools/mcp_tool.py

@@ -732,6 +732,34 @@ def _exc_str(exc: BaseException) -> str:
 # lazy so this module never triggers the ~260ms `mcp` import at import time).
 _JSONRPC_METHOD_NOT_FOUND = -32601
 
+# 2026-07-28 stateless servers answering a legacy ``initialize`` reject it
+# with one of these: UnsupportedProtocolVersion (-32022, spec-reserved range)
+# or plain method-not-found when the handshake methods are gone entirely.
+# Structural codes only — checked via _handshake_rejected_as_modern().
+_JSONRPC_UNSUPPORTED_PROTOCOL_VERSION = -32022
+
+
+def _handshake_rejected_as_modern(exc: BaseException) -> bool:
+    """True when a failed ``initialize`` signals a 2026-07-28-only server.
+
+    Mirrors :func:`_is_method_not_found_error`'s structural-then-substring
+    shape (never ``isinstance`` on SDK exception types — the SDK wraps
+    task-group errors in ``ExceptionGroup`` and symbols drift across
+    generations; see references/sdk-exceptiongroup-wrapping.md).
+    """
+    err = getattr(exc, "error", None)
+    code = getattr(err, "code", None) or getattr(exc, "code", None)
+    if code in (_JSONRPC_UNSUPPORTED_PROTOCOL_VERSION, _JSONRPC_METHOD_NOT_FOUND):
+        return True
+    msg = str(exc).lower()
+    if not msg:
+        return False
+    return (
+        "unsupported protocol version" in msg
+        or str(_JSONRPC_UNSUPPORTED_PROTOCOL_VERSION) in msg
+        or _is_method_not_found_error(exc)
+    )
+
 
 def _is_method_not_found_error(exc: BaseException) -> bool:
     """Return True if *exc* is a JSON-RPC ``method not found`` (-32601).
@@ -837,7 +865,8 @@ def _prepend_path(env: dict, directory: str) -> dict:
 _MCP_LIST_MAX_PAGES = 50
 
 
-async def _paginate_full_list(list_method, items_attr: str, server_name: str):
+async def _paginate_full_list(list_method, items_attr: str, server_name: str,
+                              cache_meta_out: Optional[dict] = None):
     """Drain a paginated MCP ``list_*`` call by following ``nextCursor``.
 
     The MCP spec allows servers to paginate ``tools/list``,
@@ -853,6 +882,10 @@ async def _paginate_full_list(list_method, items_attr: str, server_name: str):
         items_attr: Result attribute holding the page's items
             (``"tools"``, ``"resources"``, or ``"prompts"``).
         server_name: For log messages.
+        cache_meta_out: Optional dict that receives the first page's
+            SEP-2549 cache hints (``ttl_ms``, ``cache_scope``) when the
+            server provides them (2026-07-28 servers MUST; earlier ones
+            won't). Callers use ``ttl_ms`` to bound the schema cache.
 
     Returns:
         Combined list of items across all pages. Callers must hold the
@@ -862,7 +895,27 @@ async def _paginate_full_list(list_method, items_attr: str, server_name: str):
     items: list = []
     cursor = None
     for _ in range(_MCP_LIST_MAX_PAGES):
-        result = await (list_method(cursor=cursor) if cursor else list_method())
+        if not cursor:
+            result = await list_method()
+        else:
+            # Cursor continuation differs by SDK generation: mcp 1.x
+            # accepts ``cursor=``, mcp 2.0 takes ``params=`` (a
+            # PaginatedRequestParams). Try modern first, fall back.
+            try:
+                _params_cls = getattr(_mcp_types(), "PaginatedRequestParams", None)
+                if _params_cls is not None:
+                    result = await list_method(params=_params_cls(cursor=cursor))
+                else:
+                    result = await list_method(cursor=cursor)
+            except TypeError:
+                result = await list_method(cursor=cursor)
+        if cache_meta_out is not None and not items:
+            _ttl = mcp_field(result, "ttl_ms", "ttlMs")
+            _scope = mcp_field(result, "cache_scope", "cacheScope")
+            if _ttl is not None:
+                cache_meta_out["ttl_ms"] = _ttl
+            if _scope is not None:
+                cache_meta_out["cache_scope"] = _scope
         items.extend(getattr(result, items_attr, None) or [])
         cursor = mcp_field(result, "next_cursor", "nextCursor")
         # Per the MCP spec the cursor is an opaque string; anything else
@@ -878,6 +931,12 @@ async def _paginate_full_list(list_method, items_attr: str, server_name: str):
     return items
 
 
+def _mcp_types():
+    """Late import of ``mcp.types`` (module keeps the SDK import lazy)."""
+    import mcp.types as _t
+    return _t
+
+
 def _resolve_stdio_command(command: str, env: dict) -> tuple[str, dict]:
     """Resolve a stdio MCP command against the exact subprocess environment.
 
@@ -1650,6 +1709,14 @@ def _safe_numeric(value, default, coerce=int, minimum=1):
 class SamplingHandler:
     """Handles sampling/createMessage requests for a single MCP server.
 
+    .. deprecated-upstream:: MCP 2026-07-28 deprecates the Sampling feature
+       (SEP-2577, 12-month window; suggested migration is direct LLM-provider
+       integration server-side). This handler stays fully functional for the
+       deprecation window because handshake-era servers in the wild still
+       issue sampling/createMessage — but do NOT grow new capability here;
+       modern servers use MRTR (``resultType: "input_required"``) instead of
+       server-initiated requests, which the SDK's session layer handles.
+
     Each MCPServerTask that has sampling enabled creates one SamplingHandler.
     The handler is callable and passed directly to ``ClientSession`` as
     the ``sampling_callback``.  All state (rate-limit timestamps, metrics,
@@ -2264,7 +2331,7 @@ class MCPServerTask:
         "_pending_call_context",
         "_lifecycle_started_at", "_last_tool_call_at",
         "_idle_timeout_seconds", "_max_lifetime_seconds", "_recycled_reason",
-        "initialize_result", "_ping_unsupported",
+        "initialize_result", "_ping_unsupported", "_list_cache_meta",
         "_reconnect_retries", "_session_proven", "_was_parked",
     )
 
@@ -2332,6 +2399,8 @@ class MCPServerTask:
         # ``.capabilities.prompts``) instead of assuming every ``ClientSession``
         # method attribute corresponds to a supported server method. See #18051.
         self.initialize_result: Optional[Any] = None
+        # SEP-2549 cache hints from the last tools/list (ttl_ms, cache_scope).
+        self._list_cache_meta: dict = {}
         # Set True the first time a keepalive ``ping`` returns JSON-RPC
         # -32601 (method not found): the server is tool-capable but doesn't
         # implement the optional ``ping`` utility. Subsequent keepalives fall
@@ -2363,6 +2432,87 @@ class MCPServerTask:
             return True
         return getattr(caps, "tools", None) is not None
 
+    async def _negotiate_session(self, session, connect_timeout: float):
+        """Negotiate the protocol era with the server and return its result.
+
+        MCP 2026-07-28 replaced the ``initialize``/``initialized`` handshake
+        with a stateless core: every request is self-describing and clients
+        MAY probe ``server/discover`` up front (SEP-2575). The SDK exposes
+        both paths on ``ClientSession`` (``initialize()`` / ``discover()``)
+        and ``adopt()``s whichever result installs the outbound stamp, so
+        the rest of this file is era-agnostic.
+
+        Per-server ``protocol`` config key:
+
+        - ``auto`` (default): try the legacy handshake FIRST, and fall back
+          to ``server/discover`` when the server signals it is modern-only
+          (``UnsupportedProtocolVersion`` -32022, or ``initialize`` missing
+          -32601). This is the reverse of the SDK's own discover-first auto
+          mode, on purpose: nearly every configured/catalog server today
+          speaks the handshake era, and initialize-first means ZERO extra
+          round-trips and zero behavior change for all of them, while
+          stateless-only servers still connect via the fallback.
+        - ``stateless``: probe ``server/discover`` first (one legacy retry
+          on MCPError, so a handshake-only server still connects).
+        - ``legacy``: handshake only, no fallback (escape hatch for servers
+          that misbehave on unknown methods).
+
+        Both result types expose ``.capabilities``, so downstream gates
+        (``_advertises_tools``, ``_select_utility_schemas``, the config
+        probe) work unchanged on either.
+        """
+        mode = str((self._config or {}).get("protocol", "auto")).lower().strip()
+        if mode in ("stateless", "modern", "2026-07-28"):
+            try:
+                return await asyncio.wait_for(
+                    session.discover(), timeout=connect_timeout
+                )
+            except asyncio.TimeoutError:
+                raise
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                logger.info(
+                    "MCP server '%s': server/discover rejected (%s) despite "
+                    "protocol=%s — falling back to the legacy handshake",
+                    self.name, exc, mode,
+                )
+                return await asyncio.wait_for(
+                    session.initialize(), timeout=connect_timeout
+                )
+        if mode in ("legacy", "handshake"):
+            return await asyncio.wait_for(
+                session.initialize(), timeout=connect_timeout
+            )
+        if mode != "auto":
+            logger.warning(
+                "MCP server '%s': unknown protocol=%r — treating as 'auto' "
+                "(valid: auto, stateless, legacy)", self.name, mode,
+            )
+        try:
+            return await asyncio.wait_for(
+                session.initialize(), timeout=connect_timeout
+            )
+        except asyncio.TimeoutError:
+            raise
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            if not _handshake_rejected_as_modern(exc):
+                raise
+            if not hasattr(session, "discover"):
+                # Legacy SDK generation (mcp 1.x) has no server/discover
+                # client — nothing to fall back to.
+                raise
+            logger.info(
+                "MCP server '%s': legacy handshake rejected (%s) — "
+                "retrying via server/discover (2026-07-28 stateless server)",
+                self.name, exc,
+            )
+            return await asyncio.wait_for(
+                session.discover(), timeout=connect_timeout
+            )
+
     def _is_recycled_stdio(self) -> bool:
         """Return True when a stdio server was intentionally recycled."""
         return not self._is_http() and self._recycled_reason is not None
@@ -2963,8 +3113,8 @@ class MCPServerTask:
                     connect_timeout = float(
                         config.get("connect_timeout", _DEFAULT_CONNECT_TIMEOUT)
                     )
-                    self.initialize_result = await asyncio.wait_for(
-                        session.initialize(), timeout=connect_timeout
+                    self.initialize_result = await self._negotiate_session(
+                        session, connect_timeout
                     )
                     self.session = session
                     self._mark_lifecycle_started()
@@ -3335,8 +3485,8 @@ class MCPServerTask:
                         # stdio path (#59349): an endpoint that accepts the
                         # connection but never answers ``initialize`` parks this
                         # coroutine forever on the background loop.
-                        self.initialize_result = await asyncio.wait_for(
-                            session.initialize(), timeout=float(connect_timeout)
+                        self.initialize_result = await self._negotiate_session(
+                            session, float(connect_timeout)
                         )
                         self.session = session
                         await self._discover_tools()
@@ -3400,8 +3550,8 @@ class MCPServerTask:
                         read_stream, write_stream = _streams[0], _streams[1]
                         async with ClientSession(read_stream, write_stream, **sampling_kwargs) as session:
                             # Bound the handshake (#59349) — see stdio path.
-                            self.initialize_result = await asyncio.wait_for(
-                                session.initialize(), timeout=float(connect_timeout)
+                            self.initialize_result = await self._negotiate_session(
+                                session, float(connect_timeout)
                             )
                             self.session = session
                             await self._discover_tools()
@@ -3447,8 +3597,8 @@ class MCPServerTask:
                 ):
                     async with ClientSession(read_stream, write_stream, **sampling_kwargs) as session:
                         # Bound the handshake (#59349) — see stdio path.
-                        self.initialize_result = await asyncio.wait_for(
-                            session.initialize(), timeout=float(connect_timeout)
+                        self.initialize_result = await self._negotiate_session(
+                            session, float(connect_timeout)
                         )
                         self.session = session
                         await self._discover_tools()
@@ -3497,8 +3647,10 @@ class MCPServerTask:
             self._register_discovered_tools_if_needed()
             return
         async with self._rpc_lock:
+            self._list_cache_meta = {}
             self._tools = await _paginate_full_list(
-                self.session.list_tools, "tools", self.name
+                self.session.list_tools, "tools", self.name,
+                cache_meta_out=self._list_cache_meta,
             )
         self._register_discovered_tools_if_needed()
 
@@ -6796,6 +6948,8 @@ def _register_server_tools(name: str, server: MCPServerTask, config: dict) -> Li
                 config_fingerprint(config),
                 tools=tools_payload,
                 utility_tools=utility_payload,
+                ttl_ms=(getattr(server, "_list_cache_meta", None) or {}).get("ttl_ms"),
+                cache_scope=(getattr(server, "_list_cache_meta", None) or {}).get("cache_scope"),
             )
         except Exception as exc:
             logger.debug("MCP schema cache write failed for '%s': %s", name, exc)

website/docs/reference/mcp-config-reference.md

@@ -56,6 +56,7 @@ mcp_servers:
 | `enabled` | bool | both | Skip the server entirely when false |
 | `timeout` | number | both | Tool call timeout in seconds (default: `300`) |
 | `connect_timeout` | number | both | Initial connection timeout in seconds (default: `60`) |
+| `protocol` | string | both | Protocol-era negotiation: `auto` (default — legacy `initialize` handshake first, falling back to the 2026-07-28 `server/discover` stateless probe when the server rejects the handshake as modern-only), `stateless` (probe `server/discover` first; one legacy retry), or `legacy` (handshake only, no fallback) |
 | `supports_parallel_tool_calls` | bool | both | Allow tools from this server to run concurrently |
 | `skip_preflight` | bool | HTTP | Bypass the fail-fast content-type probe for valid Streamable HTTP endpoints whose HEAD/GET answers a non-MCP content type (default: `false`) |
 | `transport` | string | HTTP | Set to `sse` to use the SSE transport instead of Streamable HTTP |