Add negotiate_auto: denylist probe classifier for mode='auto'

The mode='auto' connect path now goes through client/_probe.py's
negotiate_auto, which inverts the previous allowlist into a denylist:
every MCPError from the server/discover probe falls back to initialize(),
the sole exception being -32022 with a disjoint modern-only supported list.
An unparseable probe result also falls back. Any non-MCPError exception
(network/connection errors, anyio resource errors) propagates — an outage
is never an era verdict.

ClientSession gains send_discover(version) (the raw probe, no retry, no
adopt), and discover() is reimplemented on top of it. The __aenter__
mode='auto' arm collapses to a single negotiate_auto call.

tests/client/test_probe.py covers the verdict table directly; the
interaction-suite fallback test broadens to a parametrized rpc-error set,
and the previous "INTERNAL_ERROR raises" assertion is replaced with a
network-error case (under the denylist, INTERNAL_ERROR now falls back).

Author: maxisbey

src/mcp/client/_probe.py

@@ -0,0 +1,79 @@
+"""Connect-time era negotiation for ``mode='auto'``.
+
+The ``server/discover`` probe is sent at the newest modern version. Anything
+that is not positive evidence the peer is a modern MCP server falls back to
+the legacy ``initialize`` handshake — a *denylist* (only the disjoint-modern
+case raises) rather than an allowlist of fallback codes.
+
+Every ``MCPError`` falls back except ``-32022`` with a disjoint modern-only
+``supported`` list. The streamable-HTTP transport already maps HTTP-layer
+4xx rejections (no JSON-RPC body) into ``MCPError`` codes, so those reach
+the same path. Any non-``MCPError`` exception (network/connection errors,
+anyio cancellation, the ``RuntimeError`` from ``adopt()`` on no-mutual)
+propagates to the caller; an outage or in-process bug is never an era verdict.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import ValidationError
+
+from mcp import types
+from mcp.client.session import ClientSession
+from mcp.shared.exceptions import MCPError
+from mcp.shared.version import (
+    HANDSHAKE_PROTOCOL_VERSIONS,
+    LATEST_MODERN_VERSION,
+    MODERN_PROTOCOL_VERSIONS,
+)
+from mcp.types import UNSUPPORTED_PROTOCOL_VERSION
+
+
+def _parse_supported(data: Any) -> list[str] | None:
+    """Pull ``data.supported`` off a -32022 error, or ``None`` if not actionable."""
+    try:
+        return types.UnsupportedProtocolVersionErrorData.model_validate(data).supported
+    except ValidationError:
+        return None
+
+
+async def negotiate_auto(session: ClientSession) -> None:
+    """Drive the ``mode='auto'`` connect-time policy on ``session``.
+
+    Probes ``server/discover`` once (twice if the server names a mutual
+    modern version via -32022), then either ``adopt()``s the result or falls
+    back to ``initialize()``. Idempotent only in the sense that one of
+    ``session.discover_result`` / ``session.initialize_result`` is set on
+    return.
+
+    Raises:
+        MCPError: The server is modern-only and shares no version with this
+            client (-32022 with a disjoint ``supported`` list).
+        Exception: Any transport/network error from the probe propagates as-is.
+    """
+    version = LATEST_MODERN_VERSION
+    for attempt in range(2):
+        try:
+            raw = await session.send_discover(version)
+        except MCPError as e:
+            if e.code == UNSUPPORTED_PROTOCOL_VERSION:
+                supported = _parse_supported(e.error.data)
+                mutual = [v for v in MODERN_PROTOCOL_VERSIONS if v in (supported or ())]
+                if mutual and attempt == 0:
+                    version = mutual[-1]
+                    continue
+                if supported is not None and not any(v in HANDSHAKE_PROTOCOL_VERSIONS for v in supported):
+                    raise  # server is modern-only and disjoint — real incompatibility
+            await session.initialize()  # every other rpc-error → legacy (the denylist)
+            return
+        # any other exception (httpx.TransportError, ConnectionError, anyio errors,
+        # RuntimeError from adopt) → propagate
+        try:
+            result = types.DiscoverResult.model_validate(raw)
+        except ValidationError:
+            await session.initialize()  # unparseable result → not modern evidence
+            return
+        session.adopt(result)
+        return
+    raise AssertionError("unreachable")  # pragma: no cover — loop body always returns or raises

src/mcp/client/client.py

@@ -12,6 +12,7 @@
 
 from mcp import types
 from mcp.client._memory import InMemoryTransport
+from mcp.client._probe import negotiate_auto
 from mcp.client._transport import Transport
 from mcp.client.session import ClientSession, ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
 from mcp.client.streamable_http import streamable_http_client
@@ -20,13 +21,10 @@
 from mcp.server.runner import modern_on_request
 from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
 from mcp.shared.dispatcher import Dispatcher, ProgressFnT
-from mcp.shared.exceptions import MCPDeprecationWarning, MCPError
+from mcp.shared.exceptions import MCPDeprecationWarning
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
 from mcp.types import (
-    INVALID_REQUEST,
-    METHOD_NOT_FOUND,
-    REQUEST_TIMEOUT,
     CallToolResult,
     CompleteResult,
     EmptyResult,
@@ -183,8 +181,7 @@ async def main():
     client_info: Implementation | None = None
     """Client implementation info to send to server."""
 
-    # TODO(maxisbey): flip default to 'auto' once the in-proc test suite is era-decoupled
-    # and the probe-timeout fallback is transport-aware (stdio→fallback / HTTP→reject).
+    # TODO(maxisbey): flip default to 'auto' once the in-proc test suite is era-decoupled.
     mode: ConnectMode = "legacy"
     """'legacy' performs the initialize handshake. 'auto' probes server/discover and falls back to initialize()
     on legacy servers. A modern protocol-version string (e.g. '2026-07-28') adopts that version directly without
@@ -250,15 +247,7 @@ async def __aenter__(self) -> Client:
             if self.mode == "legacy":
                 await session.initialize()
             elif self.mode == "auto":
-                try:
-                    await session.discover()
-                except MCPError as e:
-                    # TODO(L73): invert this allowlist into a `classify_probe_outcome` denylist —
-                    # fall back on every rpc-error/4xx that isn't a recognized modern error.
-                    if e.code in (METHOD_NOT_FOUND, INVALID_REQUEST, REQUEST_TIMEOUT):
-                        await session.initialize()
-                    else:
-                        raise
+                await negotiate_auto(session)
             else:
                 session.adopt(self.prior_discover or _synthesize_discover(self.mode))
 

src/mcp/client/session.py

@@ -393,6 +393,35 @@ def adopt(self, result: types.InitializeResult | types.DiscoverResult) -> None:
             self._discover_result = None
             self._negotiated_version = result.protocol_version
 
+    async def send_discover(self, version: str) -> dict[str, Any]:
+        """Send a single ``server/discover`` at ``version`` and return the raw result dict.
+
+        No retry, no ``adopt()``. The ``_meta`` envelope and the
+        ``Mcp-Protocol-Version`` header are stamped at ``version`` so the
+        server-side era router sees a coherent probe. Used by ``discover()`` and
+        the connect-time auto-negotiation policy.
+
+        Raises:
+            MCPError: The server returned a JSON-RPC error.
+            ProbeNotRecognized: The transport bounced the request at its own
+                layer (HTTP 4xx without a JSON-RPC error body).
+        """
+        client_info = self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True)
+        capabilities = self._build_capabilities().model_dump(by_alias=True, mode="json", exclude_none=True)
+        params = {
+            "_meta": {
+                PROTOCOL_VERSION_META_KEY: version,
+                CLIENT_INFO_META_KEY: client_info,
+                CLIENT_CAPABILITIES_META_KEY: capabilities,
+            }
+        }
+        opts: CallOptions = {
+            "timeout": DISCOVER_TIMEOUT_SECONDS,
+            "cancel_on_abandon": False,
+            "headers": {MCP_PROTOCOL_VERSION_HEADER: version},
+        }
+        return await self._dispatcher.send_raw_request("server/discover", params, opts)
+
     async def discover(self) -> types.DiscoverResult:
         """Probe `server/discover` and adopt the result.
 
@@ -412,26 +441,8 @@ async def discover(self) -> types.DiscoverResult:
         if self._discover_result is not None:
             return self._discover_result
 
-        client_info = self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True)
-        capabilities = self._build_capabilities().model_dump(by_alias=True, mode="json", exclude_none=True)
-
-        async def probe(version: str) -> dict[str, Any]:
-            params = {
-                "_meta": {
-                    PROTOCOL_VERSION_META_KEY: version,
-                    CLIENT_INFO_META_KEY: client_info,
-                    CLIENT_CAPABILITIES_META_KEY: capabilities,
-                }
-            }
-            opts: CallOptions = {
-                "timeout": DISCOVER_TIMEOUT_SECONDS,
-                "cancel_on_abandon": False,
-                "headers": {MCP_PROTOCOL_VERSION_HEADER: version},
-            }
-            return await self._dispatcher.send_raw_request("server/discover", params, opts)
-
         try:
-            raw = await probe(LATEST_MODERN_VERSION)
+            raw = await self.send_discover(LATEST_MODERN_VERSION)
         except MCPError as e:
             if e.code != UNSUPPORTED_PROTOCOL_VERSION:
                 raise
@@ -443,7 +454,7 @@ async def probe(version: str) -> dict[str, Any]:
             mutual = [v for v in MODERN_PROTOCOL_VERSIONS if v in data.supported]
             if not mutual:
                 raise
-            raw = await probe(mutual[-1])
+            raw = await self.send_discover(mutual[-1])
 
         result = types.DiscoverResult.model_validate(raw)
         self.adopt(result)

tests/client/test_client.py

@@ -415,13 +415,14 @@ async def test_client_auto_mode_probes_discover_then_adopts(simple_server: Serve
             assert (await client.list_resources()).resources[0].name == "Test Resource"
 
 
-@pytest.mark.parametrize("code", [types.METHOD_NOT_FOUND, types.REQUEST_TIMEOUT])
+@pytest.mark.parametrize("code", [types.METHOD_NOT_FOUND, types.REQUEST_TIMEOUT, types.INTERNAL_ERROR])
 async def test_client_auto_mode_falls_back_to_initialize_on_legacy_signal(code: int) -> None:
-    """`mode='auto'`: when `server/discover` is rejected with -32601 or -32001,
-    `Client.__aenter__` runs the legacy `initialize()` handshake and lands at a
-    handshake-era protocol version. The session itself does not fall back —
-    that policy lives here. A real `Server` always implements `server/discover`,
-    so the server side is hand-played."""
+    """`mode='auto'`: any JSON-RPC error from `server/discover` makes
+    `Client.__aenter__` run the legacy `initialize()` handshake and land at a
+    handshake-era protocol version. The denylist policy treats every server-sent
+    rpc-error as "not modern" — including INTERNAL_ERROR, since a legacy server
+    may crash on the unknown method before reaching its router. A real `Server`
+    always implements `server/discover`, so the server side is hand-played."""
     methods_seen: list[str] = []
 
     async def scripted_server(streams: MessageStream) -> None: