Derive transport headers from a constructor protocol_version pin

streamablehttp_client() and StreamableHTTPTransport now take
protocol_version: str | None = None, seeding the existing
self.protocol_version field that _prepare_headers() already reads.
_body_derived_headers (which sniffed params._meta) is replaced by
_per_message_headers, gated on the pin and reading message.method
directly so requests and notifications are handled uniformly.

The _meta envelope is request-only per spec and stays the session's
responsibility; the transport no longer treats the body as the source
of truth for connection-level headers. The constructor pin also wins
over the InitializeResult snoop.

ClientSession: pinned sessions set cancel_on_abandon=False so the
dispatcher never emits notifications/cancelled (a stateless server
cannot correlate it); the envelope keys now overwrite caller-supplied
_meta values rather than setdefault.

For now the pin is passed to both streamablehttp_client and
ClientSession; the high-level Client will collapse this to one
argument.

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

Author: maxisbey

src/mcp/client/session.py

@@ -231,16 +231,18 @@ async def send_request(
         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[PROTOCOL_VERSION_META_KEY] = self._pinned_version
+            envelope_meta[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),
+            envelope_meta[CLIENT_CAPABILITIES_META_KEY] = self._build_capabilities().model_dump(
+                by_alias=True, mode="json", exclude_none=True
             )
         opts: CallOptions = {}
+        if self._pinned_version is not None:
+            # Stateless pinned mode: disconnect-as-cancel is the spec mechanism, so the
+            # dispatcher must not emit notifications/cancelled when the caller abandons.
+            opts["cancel_on_abandon"] = False
         timeout = (
             request_read_timeout_seconds
             if request_read_timeout_seconds is not None

src/mcp/client/streamable_http.py

@@ -25,7 +25,6 @@
     INTERNAL_ERROR,
     INVALID_REQUEST,
     PARSE_ERROR,
-    PROTOCOL_VERSION_META_KEY,
     ErrorData,
     InitializeResult,
     JSONRPCError,
@@ -66,24 +65,6 @@ def _encode_header_value(value: str) -> str:
     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."""
 
@@ -106,15 +87,39 @@ class RequestContext:
 class StreamableHTTPTransport:
     """StreamableHTTP client transport implementation."""
 
-    def __init__(self, url: str) -> None:
+    def __init__(self, url: str, protocol_version: str | None = None) -> None:
         """Initialize the StreamableHTTP transport.
 
         Args:
             url: The endpoint URL.
+            protocol_version: Pin the MCP-Protocol-Version header from the first request
+                instead of waiting to snoop it from an InitializeResult. Required for
+                stateless 2026-07-28 sessions that never send initialize.
         """
         self.url = url
         self.session_id: str | None = None
-        self.protocol_version: str | None = None
+        self.protocol_version: str | None = protocol_version
+
+    # TODO: header derivation from the pin (not body _meta) per spec; body envelope is request-only.
+    def _per_message_headers(self, message: JSONRPCMessage) -> dict[str, str]:
+        """Per-POST routing headers (Mcp-Method, Mcp-Name) for 2026-07-28+ pinned transports.
+
+        MCP-Protocol-Version is not emitted here — `_prepare_headers()` already adds it
+        from `self.protocol_version` for every request.
+        """
+        if self.protocol_version is None or self.protocol_version < "2026-07-28":
+            return {}
+        if not isinstance(message, JSONRPCRequest | JSONRPCNotification):
+            return {}
+        headers: dict[str, str] = {MCP_METHOD: message.method}
+        if (
+            isinstance(message, JSONRPCRequest)
+            and message.method == "tools/call"
+            and message.params
+            and isinstance(name := message.params.get("name"), str)
+        ):
+            headers[MCP_NAME] = _encode_header_value(name)
+        return headers
 
     def _prepare_headers(self) -> dict[str, str]:
         """Build MCP-specific request headers.
@@ -150,6 +155,9 @@ def _maybe_extract_session_id_from_response(self, response: httpx.Response) -> N
 
     def _maybe_extract_protocol_version_from_message(self, message: JSONRPCMessage) -> None:
         """Extract protocol version from initialization response message."""
+        if self.protocol_version is not None:
+            # Constructor pin wins over snooping the InitializeResult.
+            return
         if isinstance(message, JSONRPCResponse) and message.result:  # pragma: no branch
             try:
                 # Parse the result as InitializeResult for type safety
@@ -289,7 +297,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))
+        headers.update(self._per_message_headers(message))
         is_initialization = self._is_initialization_request(message)
 
         async with ctx.client.stream(
@@ -556,6 +564,7 @@ async def streamable_http_client(
     *,
     http_client: httpx.AsyncClient | None = None,
     terminate_on_close: bool = True,
+    protocol_version: str | None = None,
 ) -> AsyncGenerator[TransportStreams, None]:
     """Client transport for StreamableHTTP.
 
@@ -565,6 +574,8 @@ async def streamable_http_client(
             client with recommended MCP timeouts will be created. To configure headers,
             authentication, or other HTTP settings, create an httpx.AsyncClient and pass it here.
         terminate_on_close: If True, send a DELETE request to terminate the session when the context exits.
+        protocol_version: Pin the MCP-Protocol-Version header for stateless 2026-07-28 sessions.
+            Tracer-bullet duplication — also pass to `ClientSession(protocol_version=...)`.
 
     Yields:
         Tuple containing:
@@ -582,7 +593,7 @@ async def streamable_http_client(
         # Create default client with recommended MCP timeouts
         client = create_mcp_http_client()
 
-    transport = StreamableHTTPTransport(url)
+    transport = StreamableHTTPTransport(url, protocol_version=protocol_version)
 
     logger.debug(f"Connecting to StreamableHTTP endpoint: {url}")
 

tests/client/test_streamable_http.py

@@ -1,56 +1,60 @@
 """Unit tests for the streamable-HTTP client transport.
 
 The full client<->server round trip is pinned by the interaction suite under
-tests/interaction/transports/; these tests cover the private header-derivation helpers
-directly because the headers are an HTTP-seam observation the public client never exposes.
+tests/interaction/transports/; these tests cover the transport's per-message header
+derivation directly because the headers are an HTTP-seam observation the public client
+never exposes.
 """
 
 import base64
 
 import pytest
 from inline_snapshot import snapshot
 
-from mcp.client.streamable_http import _body_derived_headers, _encode_header_value
-from mcp.types import PROTOCOL_VERSION_META_KEY, JSONRPCMessage, JSONRPCNotification, JSONRPCRequest
-
-_ENVELOPE = {PROTOCOL_VERSION_META_KEY: "2026-07-28"}
+from mcp.client.streamable_http import StreamableHTTPTransport, _encode_header_value
+from mcp.types import JSONRPCMessage, JSONRPCNotification, JSONRPCRequest, JSONRPCResponse
 
 
 @pytest.mark.parametrize(
     ("message", "expected"),
     [
         (
-            JSONRPCRequest(
-                jsonrpc="2.0", id=1, method="tools/call", params={"name": "add", "arguments": {}, "_meta": _ENVELOPE}
-            ),
-            snapshot({"mcp-protocol-version": "2026-07-28", "mcp-method": "tools/call", "mcp-name": "add"}),
-        ),
-        (
-            JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/list", params={"_meta": _ENVELOPE}),
-            snapshot({"mcp-protocol-version": "2026-07-28", "mcp-method": "tools/list"}),
+            JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={"name": "add", "arguments": {}}),
+            snapshot({"mcp-method": "tools/call", "mcp-name": "add"}),
         ),
         (
-            JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/call", params={"_meta": _ENVELOPE}),
-            snapshot({"mcp-protocol-version": "2026-07-28", "mcp-method": "tools/call"}),
+            JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/list", params={}),
+            snapshot({"mcp-method": "tools/list"}),
         ),
         (
-            JSONRPCRequest(jsonrpc="2.0", id=3, method="tools/call", params={"name": "add", "arguments": {}}),
-            snapshot({}),
+            JSONRPCNotification(jsonrpc="2.0", method="notifications/cancelled"),
+            snapshot({"mcp-method": "notifications/cancelled"}),
         ),
         (
-            JSONRPCNotification(jsonrpc="2.0", method="notifications/initialized"),
+            JSONRPCResponse(jsonrpc="2.0", id=3, result={}),
             snapshot({}),
         ),
     ],
 )
-def test_body_derived_headers_reflect_the_envelope_on_the_request_body(
+def test_per_message_headers_for_pinned_transport_carry_method_and_name(
     message: JSONRPCMessage, expected: dict[str, str]
 ) -> None:
-    """An envelope-bearing body yields the three stateless headers; a legacy body yields none.
+    """A 2026-07-28-pinned transport derives ``Mcp-Method`` (and ``Mcp-Name`` for tools/call) from the body.
 
-    Legacy bodies returning ``{}`` is what keeps the unpinned wire byte-identical to a pre-2026 client.
+    ``MCP-Protocol-Version`` is not in the per-message set: ``_prepare_headers()`` adds it from the
+    pin for every request, so only the method/name advisory headers vary per POST. Responses yield
+    nothing because the spec only defines the headers for requests and notifications.
     """
-    assert _body_derived_headers(message) == expected
+    transport = StreamableHTTPTransport("http://test/mcp", protocol_version="2026-07-28")
+    assert transport._per_message_headers(message) == expected  # pyright: ignore[reportPrivateUsage]
+
+
+@pytest.mark.parametrize("protocol_version", [None, "2025-11-25"])
+def test_per_message_headers_are_empty_for_legacy_or_unpinned_transport(protocol_version: str | None) -> None:
+    """An unpinned or 2025-era transport emits no per-message headers, keeping the wire byte-identical to v1."""
+    transport = StreamableHTTPTransport("http://test/mcp", protocol_version=protocol_version)
+    message = JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={"name": "add", "arguments": {}})
+    assert transport._per_message_headers(message) == {}  # pyright: ignore[reportPrivateUsage]
 
 
 @pytest.mark.parametrize(
@@ -78,3 +82,19 @@ def test_mcp_name_header_values_are_base64_wrapped_when_unsafe_for_an_http_field
         assert base64.b64decode(encoded.removeprefix("=?base64?").removesuffix("?=")).decode() == raw
     else:
         assert encoded == raw
+
+
+def test_constructor_pin_is_not_overwritten_by_an_initialize_result() -> None:
+    """A protocol_version passed at construction wins over the InitializeResult snoop."""
+    transport = StreamableHTTPTransport("http://test/mcp", protocol_version="2026-07-28")
+    init = JSONRPCResponse(
+        jsonrpc="2.0",
+        id=1,
+        result={
+            "protocolVersion": "2025-11-25",
+            "capabilities": {},
+            "serverInfo": {"name": "s", "version": "0"},
+        },
+    )
+    transport._maybe_extract_protocol_version_from_message(init)  # pyright: ignore[reportPrivateUsage]
+    assert transport.protocol_version == "2026-07-28"

tests/interaction/transports/test_client_transport_http_modern.py

@@ -1,10 +1,11 @@
 """Behaviour of the streamable-HTTP client transport under the 2026-07-28 stateless protocol.
 
 A pinned session stamps the ``io.modelcontextprotocol/*`` `_meta` envelope onto every outgoing
-request, and the streamable-HTTP transport derives the ``MCP-Protocol-Version`` / ``Mcp-Method`` /
-``Mcp-Name`` headers from that body. These tests pin the composition through a real ``httpx``
-request against a canned ``httpx.MockTransport`` -- no in-process 2026 server exists yet to record
-the headers against. The header-derivation helpers themselves are unit-tested in
+request, and a pinned streamable-HTTP transport adds the ``MCP-Protocol-Version`` / ``Mcp-Method`` /
+``Mcp-Name`` headers to each POST. The pin is a tracer-bullet duplication: both
+``streamable_http_client()`` and ``ClientSession()`` take ``protocol_version`` until the higher-level
+client wires them together. These tests drive the composition through a real ``httpx`` request
+against a canned ``httpx.MockTransport``; the per-message header derivation itself is unit-tested in
 ``tests/client/test_streamable_http.py``.
 """
 
@@ -27,14 +28,13 @@
 @requirement("client-transport:http:body-derived-headers")
 @requirement("lifecycle:stateless:request-envelope")
 async def test_pinned_session_post_carries_body_derived_headers_on_the_wire() -> None:
-    """A pinned ``call_tool`` over streamable HTTP lands as a POST whose headers were derived from its body.
+    """A pinned ``call_tool`` over streamable HTTP lands as a POST carrying the three stateless headers.
 
     Spec-mandated for the body-derived headers and the request envelope: this is the wire-seam proof
-    that the ``ClientSession`` envelope stamp and the transport's header derivation are actually
-    composed -- the streamable-HTTP POST wiring is driven through a real ``httpx`` request. A canned
-    ``httpx.MockTransport`` stands in for the (not-yet-existing) 2026 server; the ``isError`` result
-    skips the client's implicit ``tools/list`` output-schema fetch so the recorded log is the single
-    POST.
+    that the ``ClientSession`` envelope stamp and the pinned transport's header derivation are
+    actually composed -- the streamable-HTTP POST wiring is driven through a real ``httpx`` request.
+    A canned ``httpx.MockTransport`` stands in for the server; the ``isError`` result skips the
+    client's implicit ``tools/list`` output-schema fetch so the recorded log is the single POST.
     """
     recorded: list[httpx.Request] = []
 
@@ -47,7 +47,7 @@ def handler(request: httpx.Request) -> httpx.Response:
     with anyio.fail_after(5):
         async with (
             httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http,
-            streamable_http_client(f"{BASE_URL}/mcp", http_client=http) as (read, write),
+            streamable_http_client(f"{BASE_URL}/mcp", http_client=http, protocol_version="2026-07-28") as (read, write),
             ClientSession(
                 read,
                 write,
@@ -103,7 +103,7 @@ def handler(request: httpx.Request) -> httpx.Response:
     with anyio.fail_after(5):
         async with (
             httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http,
-            streamable_http_client(f"{BASE_URL}/mcp", http_client=http) as (read, write),
+            streamable_http_client(f"{BASE_URL}/mcp", http_client=http, protocol_version="2026-07-28") as (read, write),
             ClientSession(read, write, protocol_version="2026-07-28") as session,
         ):
             await session.call_tool("add", {"a": 2, "b": 3})

tests/interaction/transports/test_hosting_http_modern.py

@@ -220,7 +220,10 @@ async def on_response(response: httpx.Response) -> None:
     with anyio.fail_after(5):
         async with (
             mounted_app(_server(), on_request=on_request, on_response=on_response) as (http, _),
-            streamable_http_client(f"{BASE_URL}/mcp", http_client=http) as (read, write),
+            streamable_http_client(f"{BASE_URL}/mcp", http_client=http, protocol_version=MODERN_VERSION) as (
+                read,
+                write,
+            ),
             ClientSession(read, write, client_info=client_info, protocol_version=MODERN_VERSION) as session,
         ):
             result = await session.call_tool("add", {"a": 2, "b": 3})