Add protocol_version pin to ClientSession for stateless 2026-07-28 mode

When ClientSession is constructed with protocol_version="2026-07-28",
each outgoing request carries the io.modelcontextprotocol/* envelope
(protocolVersion, clientInfo, clientCapabilities) in params._meta, and
initialize() raises if called. Capabilities derivation is extracted to
_build_capabilities() so both paths share it.

The streamable-HTTP transport derives MCP-Protocol-Version, Mcp-Method
and (for tools/call) Mcp-Name headers per POST from the body's envelope;
non-header-safe values are Base64-sentinel-encoded per the spec.
Envelope-less bodies get no derived headers, so unpinned behaviour is
unchanged. Session-id capture, the standalone GET stream and DELETE on
close are gated on traffic the pinned mode never produces.

Claude-Session: https://claude.ai/code/session_017S3aJaxEHeMvftp6whnHWK

Author: maxisbey

src/mcp/client/session.py

@@ -22,7 +22,15 @@
 from mcp.shared.session import RequestResponder
 from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
-from mcp.types import INTERNAL_ERROR, METHOD_NOT_FOUND, RequestId, RequestParamsMeta
+from mcp.types import (
+    CLIENT_CAPABILITIES_META_KEY,
+    CLIENT_INFO_META_KEY,
+    INTERNAL_ERROR,
+    METHOD_NOT_FOUND,
+    PROTOCOL_VERSION_META_KEY,
+    RequestId,
+    RequestParamsMeta,
+)
 from mcp.types import methods as _methods
 
 DEFAULT_CLIENT_INFO = types.Implementation(name="mcp", version="0.1.0")
@@ -141,11 +149,13 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         client_info: types.Implementation | 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._sampling_callback = sampling_callback or _default_sampling_callback
         self._sampling_capabilities = sampling_capabilities
         self._elicitation_callback = elicitation_callback or _default_elicitation_callback
@@ -218,6 +228,18 @@ async def send_request(
         """
         data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
         method: str = data["method"]
+        if self._pinned_version is not None:
+            params = data.setdefault("params", {})
+            envelope_meta = params.setdefault("_meta", {})
+            envelope_meta.setdefault(PROTOCOL_VERSION_META_KEY, self._pinned_version)
+            envelope_meta.setdefault(
+                CLIENT_INFO_META_KEY,
+                self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True),
+            )
+            envelope_meta.setdefault(
+                CLIENT_CAPABILITIES_META_KEY,
+                self._build_capabilities().model_dump(by_alias=True, mode="json", exclude_none=True),
+            )
         opts: CallOptions = {}
         timeout = (
             request_read_timeout_seconds
@@ -254,7 +276,7 @@ async def send_notification(self, notification: types.ClientNotification) -> Non
         data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
         await self._dispatcher.notify(data["method"], data.get("params"))
 
-    async def initialize(self) -> types.InitializeResult:
+    def _build_capabilities(self) -> types.ClientCapabilities:
         sampling = (
             (self._sampling_capabilities or types.SamplingCapability())
             if self._sampling_callback is not _default_sampling_callback
@@ -273,17 +295,17 @@ async def initialize(self) -> types.InitializeResult:
             if self._list_roots_callback is not _default_list_roots_callback
             else None
         )
+        return types.ClientCapabilities(sampling=sampling, elicitation=elicitation, experimental=None, roots=roots)
 
+    async def initialize(self) -> types.InitializeResult:
+        if self._pinned_version is not None:
+            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,
-                    capabilities=types.ClientCapabilities(
-                        sampling=sampling,
-                        elicitation=elicitation,
-                        experimental=None,
-                        roots=roots,
-                    ),
+                    capabilities=capabilities,
                     client_info=self._client_info,
                 ),
             ),
@@ -309,7 +331,9 @@ def initialize_result(self) -> types.InitializeResult | None:
 
     @property
     def protocol_version(self) -> str | None:
-        """The negotiated protocol version. None until `initialize()` has completed."""
+        """Negotiated or pinned protocol version. None until initialize() unless pinned at construction."""
+        if self._pinned_version is not None:
+            return self._pinned_version
         return self._initialize_result.protocol_version if self._initialize_result else None
 
     async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:

src/mcp/client/streamable_http.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations as _annotations
 
+import base64
 import contextlib
 import logging
+import re
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
@@ -23,6 +25,7 @@
     INTERNAL_ERROR,
     INVALID_REQUEST,
     PARSE_ERROR,
+    PROTOCOL_VERSION_META_KEY,
     ErrorData,
     InitializeResult,
     JSONRPCError,
@@ -44,12 +47,42 @@
 
 MCP_SESSION_ID = "mcp-session-id"
 MCP_PROTOCOL_VERSION = "mcp-protocol-version"
+MCP_METHOD = "mcp-method"
+MCP_NAME = "mcp-name"
 LAST_EVENT_ID = "last-event-id"
 
 # Reconnection defaults
 DEFAULT_RECONNECTION_DELAY_MS = 1000  # 1 second fallback when server doesn't provide retry
 MAX_RECONNECTION_ATTEMPTS = 2  # Max retry attempts before giving up
 
+_B64_SENTINEL = re.compile(r"^=\?base64\?.*\?=$")
+# RFC 7230 token chars minus DEL; visible ASCII 0x20-0x7E is the practical bound for a header value.
+_HEADER_SAFE = re.compile(r"^[\x20-\x7E]*$")
+
+
+def _encode_header_value(value: str) -> str:
+    if _HEADER_SAFE.fullmatch(value) and not _B64_SENTINEL.fullmatch(value):
+        return value
+    return f"=?base64?{base64.b64encode(value.encode('utf-8')).decode('ascii')}?="
+
+
+def _body_derived_headers(message: JSONRPCMessage) -> dict[str, str]:
+    """Derive 2026-era headers from an envelope-bearing request body. Empty dict for legacy bodies."""
+    if not isinstance(message, JSONRPCRequest) or message.params is None:
+        return {}
+    meta = message.params.get("_meta")
+    if meta is None:
+        return {}
+    version = meta.get(PROTOCOL_VERSION_META_KEY)
+    if not isinstance(version, str):
+        return {}
+    headers: dict[str, str] = {MCP_PROTOCOL_VERSION: version, MCP_METHOD: message.method}
+    if message.method == "tools/call":
+        name = message.params.get("name")
+        if isinstance(name, str):
+            headers[MCP_NAME] = _encode_header_value(name)
+    return headers
+
 
 class StreamableHTTPError(Exception):
     """Base exception for StreamableHTTP transport errors."""
@@ -256,6 +289,7 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
         """Handle a POST request with response processing."""
         headers = self._prepare_headers()
         message = ctx.session_message.message
+        headers.update(_body_derived_headers(message))
         is_initialization = self._is_initialization_request(message)
 
         async with ctx.client.stream(