Client gains mode= and prior_discover= policy knobs (legacy and version-pin)

- mode='legacy' (default) performs the initialize handshake; a version
  string (e.g. '2026-07-28') adopts that version directly via .adopt()
- prior_discover= reuses a known DiscoverResult; omitting it synthesizes
  a minimal one
- 'auto' (server/discover probe with fallback) follows once .discover() lands
- Interaction-suite connect fixture passes mode= for the modern arm and
  yields Client for all arms again; the W1b-era ClientSession adapter and
  type suppression are removed

Author: maxisbey

src/mcp/client/client.py

@@ -4,10 +4,11 @@
 
 from contextlib import AsyncExitStack
 from dataclasses import KW_ONLY, dataclass, field
-from typing import Any
+from typing import Any, Literal
 
 from typing_extensions import deprecated
 
+from mcp import types
 from mcp.client._memory import InMemoryTransport
 from mcp.client._transport import Transport
 from mcp.client.session import ClientSession, ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
@@ -36,6 +37,17 @@
 )
 
 
+def _synthesize_discover(protocol_version: str) -> types.DiscoverResult:
+    return types.DiscoverResult(
+        supported_versions=[protocol_version],
+        capabilities=types.ServerCapabilities(),
+        server_info=types.Implementation(name="", version=""),
+        result_type="complete",
+        ttl_ms=0,
+        cache_scope="public",
+    )
+
+
 @dataclass
 class Client:
     """A high-level MCP client for connecting to MCP servers.
@@ -95,6 +107,15 @@ async def main():
     client_info: Implementation | None = None
     """Client implementation info to send to server."""
 
+    mode: Literal["legacy"] | str = "legacy"
+    """'legacy' performs the initialize handshake. A protocol-version string (e.g. '2026-07-28') adopts that
+    version directly without a handshake — supply prior_discover to reuse a known DiscoverResult, or omit it
+    to synthesize a minimal one."""
+
+    prior_discover: types.DiscoverResult | None = None
+    """A previously-obtained DiscoverResult to install via .adopt() when mode is a version pin.
+    Ignored when mode='legacy'."""
+
     elicitation_callback: ElicitationFnT | None = None
     """Callback for handling elicitation requests."""
 
@@ -132,7 +153,10 @@ async def __aenter__(self) -> Client:
                 )
             )
 
-            await self._session.initialize()
+            if self.mode == "legacy":
+                await self._session.initialize()
+            else:
+                self._session.adopt(self.prior_discover or _synthesize_discover(self.mode))
 
             # Transfer ownership to self for __aexit__ to handle
             self._exit_stack = exit_stack.pop_all()

tests/interaction/_connect.py

@@ -20,7 +20,7 @@
 from starlette.routing import Mount, Route
 
 from mcp.client.client import Client
-from mcp.client.session import ClientSession, ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
+from mcp.client.session import ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
@@ -35,13 +35,11 @@
 from mcp.types import (
     LATEST_PROTOCOL_VERSION,
     ClientCapabilities,
-    DiscoverResult,
     Implementation,
     InitializeRequestParams,
     JSONRPCMessage,
     JSONRPCRequest,
     JSONRPCResponse,
-    ServerCapabilities,
     jsonrpc_message_adapter,
 )
 from tests.interaction.transports._bridge import StreamingASGITransport
@@ -129,9 +127,9 @@ async def connect_over_streamable_http(
     resumability tests pass an `event_store` (with `retry_interval=0` so the client's
     reconnection wait is a no-op).
 
-    When `spec_version` is a modern (2026-07-28+) revision, the modern path is exercised: a bare
-    `ClientSession` is built over the streams and adopted from a synthesized `DiscoverResult`
-    instead of negotiating via `Client`'s legacy initialize handshake.
+    When `spec_version` is a modern (2026-07-28+) revision the Client is opened with
+    `mode=<version>`, which adopts a synthesized DiscoverResult instead of running the legacy
+    initialize handshake.
     """
     app = server.streamable_http_app(
         stateless_http=stateless_http,
@@ -143,48 +141,19 @@ async def connect_over_streamable_http(
     async with (
         server.session_manager.run(),
         httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http_client,
+        Client(
+            streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client),
+            mode=spec_version if spec_version in MODERN_PROTOCOL_VERSIONS else "legacy",
+            read_timeout_seconds=read_timeout_seconds,
+            sampling_callback=sampling_callback,
+            list_roots_callback=list_roots_callback,
+            logging_callback=logging_callback,
+            message_handler=message_handler,
+            client_info=client_info,
+            elicitation_callback=elicitation_callback,
+        ) as client,
     ):
-        if spec_version in MODERN_PROTOCOL_VERSIONS:
-            async with (
-                streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client) as (read, write),
-                ClientSession(
-                    read,
-                    write,
-                    read_timeout_seconds=read_timeout_seconds,
-                    sampling_callback=sampling_callback,
-                    list_roots_callback=list_roots_callback,
-                    logging_callback=logging_callback,
-                    message_handler=message_handler,
-                    client_info=client_info,
-                    elicitation_callback=elicitation_callback,
-                ) as session,
-            ):
-                session.adopt(
-                    DiscoverResult(
-                        supported_versions=[spec_version],
-                        capabilities=ServerCapabilities(),
-                        server_info=Implementation(name="test", version="0"),
-                        result_type="complete",
-                        ttl_ms=0,
-                        cache_scope="public",
-                    )
-                )
-                # ClientSession quacks as Client for every modern-arm requirement; the surfaces
-                # diverge (cursor= vs params=, .session, nullable initialize_result) so widening
-                # Connect to the union cascades ~58 errors. Contained here until the two converge.
-                yield session  # pyright: ignore[reportReturnType]
-        else:
-            async with Client(
-                streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client),
-                read_timeout_seconds=read_timeout_seconds,
-                sampling_callback=sampling_callback,
-                list_roots_callback=list_roots_callback,
-                logging_callback=logging_callback,
-                message_handler=message_handler,
-                client_info=client_info,
-                elicitation_callback=elicitation_callback,
-            ) as client:
-                yield client
+        yield client
 
 
 connect_over_streamable_http_stateless: Connect = partial(connect_over_streamable_http, stateless_http=True)

tests/interaction/lowlevel/test_cancellation.py

@@ -148,8 +148,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
     server = Server("calm", on_list_tools=list_tools, on_call_tool=call_tool)
 
     async with connect(server) as client:
-        session = client if isinstance(client, ClientSession) else client.session  # modern arm yields a bare session
-        await session.send_notification(
+        await client.session.send_notification(
             types.CancelledNotification(params=types.CancelledNotificationParams(request_id=9999))
         )
         result = await client.call_tool("echo", {})

tests/interaction/lowlevel/test_pagination.py

@@ -5,9 +5,6 @@
 pagination scheme.
 """
 
-from collections.abc import Callable, Coroutine
-from typing import Any, TypeVar
-
 import pytest
 from inline_snapshot import snapshot
 
@@ -29,19 +26,6 @@
 
 pytestmark = pytest.mark.anyio
 
-_ResultT = TypeVar("_ResultT")
-
-
-def _page(list_fn: Callable[..., Coroutine[Any, Any, _ResultT]], cursor: str | None) -> Coroutine[Any, Any, _ResultT]:
-    """Call a paginated list method with a cursor on whichever client surface `connect` yielded.
-
-    `Client.list_*` takes `cursor=`; `ClientSession.list_*` takes `params=PaginatedRequestParams(...)`.
-    """
-    try:
-        return list_fn(cursor=cursor)
-    except TypeError:
-        return list_fn(params=types.PaginatedRequestParams(cursor=cursor))
-
 
 @requirement("tools:list:pagination")
 async def test_next_cursor_round_trips_through_the_client(connect: Connect) -> None:
@@ -65,7 +49,7 @@ async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestPa
 
     async with connect(server) as client:
         first_page = await client.list_tools()
-        second_page = await _page(client.list_tools, first_page.next_cursor)
+        second_page = await client.list_tools(cursor=first_page.next_cursor)
 
     assert first_page.next_cursor == cursor
     assert seen_cursors == [None, cursor]
@@ -95,7 +79,7 @@ async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestPa
     requests_made = 0
     async with connect(server) as client:
         while True:
-            result = await _page(client.list_tools, cursor)
+            result = await client.list_tools(cursor=cursor)
             requests_made += 1
             assert requests_made <= len(pages), "the server kept returning next_cursor past the last page"
             collected.extend(tool.name for tool in result.tools)
@@ -138,7 +122,7 @@ async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestPa
     cursor: str | None = None
     async with connect(server) as client:
         while True:
-            result = await _page(client.list_tools, cursor)
+            result = await client.list_tools(cursor=cursor)
             page_sizes.append(len(result.tools))
             if result.next_cursor is None:
                 break
@@ -166,7 +150,7 @@ async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestPa
 
     async with connect(server) as client:
         with pytest.raises(MCPError) as exc_info:
-            await _page(client.list_tools, "never-issued")
+            await client.list_tools(cursor="never-issued")
 
     assert exc_info.value.error.code == INVALID_PARAMS
 
@@ -190,7 +174,7 @@ async def list_resources(
 
     async with connect(server) as client:
         first_page = await client.list_resources()
-        second_page = await _page(client.list_resources, first_page.next_cursor)
+        second_page = await client.list_resources(cursor=first_page.next_cursor)
 
     assert first_page.next_cursor == cursor
     assert seen_cursors == [None, cursor]
@@ -223,7 +207,7 @@ async def list_resource_templates(
 
     async with connect(server) as client:
         first_page = await client.list_resource_templates()
-        second_page = await _page(client.list_resource_templates, first_page.next_cursor)
+        second_page = await client.list_resource_templates(cursor=first_page.next_cursor)
 
     assert first_page.next_cursor == cursor
     assert seen_cursors == [None, cursor]
@@ -249,7 +233,7 @@ async def list_prompts(ctx: ServerRequestContext, params: types.PaginatedRequest
 
     async with connect(server) as client:
         first_page = await client.list_prompts()
-        second_page = await _page(client.list_prompts, first_page.next_cursor)
+        second_page = await client.list_prompts(cursor=first_page.next_cursor)
 
     assert first_page.next_cursor == cursor
     assert seen_cursors == [None, cursor]