Dispatch ClientSession.protocol_version by era instead of restricting to a Literal

A pin to a stateful version now flows through initialize() as the
requested version (instead of LATEST_PROTOCOL_VERSION); only a pin to a
2026-07-28+ stateless version raises. The envelope-stamp and
cancel_on_abandon=False in send_request key off the same
is_version_at_least gate. StatelessProtocolVersion is dropped.

Author: maxisbey

src/mcp/client/session.py

@@ -21,7 +21,7 @@
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.shared.session import RequestResponder
 from mcp.shared.transport_context import TransportContext
-from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS, StatelessProtocolVersion
+from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS, is_version_at_least
 from mcp.types import (
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
@@ -149,13 +149,14 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         client_info: types.Implementation | None = None,
         *,
-        protocol_version: StatelessProtocolVersion | None = None,
+        protocol_version: str | None = None,
         sampling_capabilities: types.SamplingCapability | None = None,
         dispatcher: Dispatcher[Any] | None = None,
     ) -> None:
         self._session_read_timeout_seconds = read_timeout_seconds
         self._client_info = client_info or DEFAULT_CLIENT_INFO
         self._pinned_version = protocol_version
+        self._stateless_pinned = protocol_version is not None and is_version_at_least(protocol_version, "2026-07-28")
         self._sampling_callback = sampling_callback or _default_sampling_callback
         self._sampling_capabilities = sampling_capabilities
         self._elicitation_callback = elicitation_callback or _default_elicitation_callback
@@ -229,7 +230,7 @@ async def send_request(
         data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
         method: str = data["method"]
         opts: CallOptions = {}
-        if self._pinned_version is not None:
+        if self._stateless_pinned:
             params = data.setdefault("params", {})
             envelope_meta = params.setdefault("_meta", {})
             envelope_meta[PROTOCOL_VERSION_META_KEY] = self._pinned_version
@@ -299,13 +300,15 @@ def _build_capabilities(self) -> types.ClientCapabilities:
         return types.ClientCapabilities(sampling=sampling, elicitation=elicitation, experimental=None, roots=roots)
 
     async def initialize(self) -> types.InitializeResult:
-        if self._pinned_version is not None:
+        if self._stateless_pinned:
             raise RuntimeError("initialize() must not be called on a session pinned to a stateless protocol version")
         capabilities = self._build_capabilities()
         result = await self.send_request(
             types.InitializeRequest(
                 params=types.InitializeRequestParams(
-                    protocol_version=types.LATEST_PROTOCOL_VERSION,
+                    protocol_version=self._pinned_version
+                    if self._pinned_version is not None
+                    else types.LATEST_PROTOCOL_VERSION,
                     capabilities=capabilities,
                     client_info=self._client_info,
                 ),

src/mcp/client/streamable_http.py

@@ -21,7 +21,7 @@
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
 from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
-from mcp.shared.version import StatelessProtocolVersion, is_version_at_least
+from mcp.shared.version import is_version_at_least
 from mcp.types import (
     INTERNAL_ERROR,
     INVALID_REQUEST,
@@ -567,7 +567,7 @@ async def streamable_http_client(
     *,
     http_client: httpx.AsyncClient | None = None,
     terminate_on_close: bool = True,
-    protocol_version: StatelessProtocolVersion | None = None,
+    protocol_version: str | None = None,
 ) -> AsyncGenerator[TransportStreams, None]:
     """Client transport for StreamableHTTP.
 

src/mcp/shared/version.py

@@ -7,13 +7,10 @@
 ordering questions go through KNOWN_PROTOCOL_VERSIONS.
 """
 
-from typing import Final, Literal
+from typing import Final
 
 from mcp.types import LATEST_PROTOCOL_VERSION
 
-StatelessProtocolVersion = Literal["2026-07-28"]
-"""Protocol revisions that use the stateless per-request envelope (no `initialize`)."""
-
 KNOWN_PROTOCOL_VERSIONS: Final[tuple[str, ...]] = (
     "2024-11-05",
     "2025-03-26",

tests/client/test_session.py

@@ -1401,19 +1401,46 @@ async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
 
 
 @pytest.mark.anyio
-async def test_initialize_on_a_pinned_session_raises_before_any_frame_is_sent():
+async def test_initialize_on_a_stateless_pinned_session_raises_before_any_frame_is_sent():
     """A session pinned to the 2026-07-28 stateless protocol rejects ``initialize()`` locally.
 
     The 2026-07-28 lifecycle replaces the initialize handshake with a per-request ``_meta``
-    envelope, so calling ``initialize()`` on a pinned session is a programmer error and raises
-    immediately rather than reaching the wire.
+    envelope, so calling ``initialize()`` on a stateless-pinned session is a programmer error
+    and raises immediately rather than reaching the wire.
     """
     async with raw_client_session(protocol_version="2026-07-28") as (session, _send, from_client):
         with pytest.raises(RuntimeError, match="pinned to a stateless"):
             await session.initialize()
         assert from_client.statistics().current_buffer_used == 0
 
 
+@pytest.mark.anyio
+async def test_initialize_on_a_stateful_pin_requests_the_pinned_version():
+    """A session pinned to a pre-2026 stateful version still runs the handshake, but the
+    outgoing ``initialize`` frame requests the pinned version rather than ``LATEST``."""
+    async with raw_client_session(protocol_version="2025-06-18") as (session, to_client, from_client):
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(session.initialize)
+            out = await from_client.receive()
+            assert isinstance(out.message, JSONRPCRequest)
+            assert out.message.params is not None
+            assert out.message.params["protocolVersion"] == "2025-06-18"
+            result = InitializeResult(
+                protocol_version="2025-06-18",
+                capabilities=ServerCapabilities(),
+                server_info=Implementation(name="mock-server", version="0.1.0"),
+            )
+            await to_client.send(
+                SessionMessage(
+                    JSONRPCResponse(
+                        jsonrpc="2.0",
+                        id=out.message.id,
+                        result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
+                    )
+                )
+            )
+
+
 @pytest.mark.anyio
 async def test_send_notification_after_close_is_dropped_silently():
     """Post-close `send_notification` is fire-and-forget: the notification is dropped,