Thread per-message headers from session to transport via CallOptions/metadata sidecar

Additive infrastructure for the client-side outbound stamp:

- CallOptions gains a headers key; ClientMessageMetadata gains a headers field
- _plan_outbound projects opts['headers'] onto the metadata (same path
  resumption tokens take); JSONRPCDispatcher.notify accepts opts and threads
  headers through
- Outbound.notify Protocol grows opts=None; all implementers updated
  (Connection, _NoChannelOutbound, _SingleExchangeDispatchContext, peer,
  context, DirectDispatcher, test stubs)
- StreamableHTTPTransport's POST path merges metadata.headers into the
  request (alongside existing _prepare_headers/_per_message_headers, which
  are removed in the next commit)
- MCP_METHOD_HEADER, MCP_NAME_HEADER, encode_header_value moved to
  shared/inbound (single source for the header names)
- Tests pin both new paths

Author: maxisbey

src/mcp/client/streamable_http.py

@@ -2,10 +2,8 @@
 
 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
@@ -20,7 +18,7 @@
 from mcp.shared._compat import resync_tracer
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
 from mcp.shared._httpx_utils import create_mcp_http_client
-from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
+from mcp.shared.inbound import MCP_METHOD_HEADER, MCP_NAME_HEADER, MCP_PROTOCOL_VERSION_HEADER, encode_header_value
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.shared.version import MODERN_PROTOCOL_VERSIONS
 from mcp.types import (
@@ -47,24 +45,12 @@
 StreamReader = ContextReceiveStream[SessionMessage]
 
 MCP_SESSION_ID = "mcp-session-id"
-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 value == value.strip() and not _B64_SENTINEL.fullmatch(value):
-        return value
-    return f"=?base64?{base64.b64encode(value.encode('utf-8')).decode('ascii')}?="
-
 
 class StreamableHTTPError(Exception):
     """Base exception for StreamableHTTP transport errors."""
@@ -112,7 +98,7 @@ def _per_message_headers(self, message: JSONRPCMessage) -> dict[str, str]:
             return {}
         if not isinstance(message, JSONRPCRequest | JSONRPCNotification):
             return {}
-        headers: dict[str, str] = {MCP_METHOD: message.method}
+        headers: dict[str, str] = {MCP_METHOD_HEADER: message.method}
         # TODO: Mcp-Name is also REQUIRED for prompts/get (params.name) and resources/read
         # (params.uri); a method->param-key map replaces this gate when those land.
         if (
@@ -121,7 +107,7 @@ def _per_message_headers(self, message: JSONRPCMessage) -> dict[str, str]:
             and message.params
             and isinstance(name := message.params.get("name"), str)
         ):
-            headers[MCP_NAME] = _encode_header_value(name)
+            headers[MCP_NAME_HEADER] = encode_header_value(name)
         return headers
 
     def _prepare_headers(self) -> dict[str, str]:
@@ -302,6 +288,8 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
         headers = self._prepare_headers()
         message = ctx.session_message.message
         headers.update(self._per_message_headers(message))
+        if ctx.metadata is not None and ctx.metadata.headers is not None:
+            headers.update(ctx.metadata.headers)
         is_initialization = self._is_initialization_request(message)
 
         async with ctx.client.stream(

src/mcp/server/_streamable_http_modern.py

@@ -77,7 +77,7 @@ async def send_raw_request(
     ) -> dict[str, Any]:
         raise NoBackChannelError(method)
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         # TODO(D-005a): buffer and stream as SSE once the JSON-vs-SSE response mode lands.
         return None
 

src/mcp/server/connection.py

@@ -93,7 +93,7 @@ async def send_raw_request(
     ) -> dict[str, Any]:
         raise NoBackChannelError(method)
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         logger.debug("dropped %s: no standalone channel", method)
 
 
@@ -277,14 +277,14 @@ async def send_request(
         cls = result_type if result_type is not None else _RESULT_FOR[type(req)]
         return cls.model_validate(raw, by_name=False)
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         """Send a best-effort notification on the standalone stream.
 
         Never raises. If there's no standalone channel or the stream is broken,
         the notification is dropped and debug-logged.
         """
         try:
-            await self.outbound.notify(method, params)
+            await self.outbound.notify(method, params, opts)
         except (anyio.BrokenResourceError, anyio.ClosedResourceError):
             logger.debug("dropped %s: standalone stream closed", method)
 

src/mcp/shared/context.py

@@ -73,9 +73,9 @@ async def send_raw_request(
         """
         return await self._dctx.send_raw_request(method, params, opts)
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         """Send a notification to the peer on the back-channel."""
-        await self._dctx.notify(method, params)
+        await self._dctx.notify(method, params, opts)
 
     async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
         """Report progress for this request, if the peer supplied a progress token.

src/mcp/shared/direct_dispatcher.py

@@ -63,7 +63,7 @@ class _DirectDispatchContext:
     def can_send_request(self) -> bool:
         return self.transport.can_send_request
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         await self._back_notify(method, params)
 
     async def send_raw_request(
@@ -133,12 +133,13 @@ async def send_raw_request(
             raise RuntimeError("DirectDispatcher.send_raw_request called before run()")
         return await self._peer._dispatch_request(method, params, opts)
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         """Send a notification by invoking the peer's `on_notify` directly.
 
         Fire-and-forget: usable before `run()` (delivery waits for the peer to
         start), and after close it is silently dropped, matching
-        `JSONRPCDispatcher.notify`.
+        `JSONRPCDispatcher.notify`. `opts` is accepted for `Dispatcher`
+        conformance; there is no HTTP layer here so `headers` is ignored.
         """
         if self._peer is None:
             raise RuntimeError("DirectDispatcher has no peer; use create_direct_dispatcher_pair()")

src/mcp/shared/dispatcher.py

@@ -85,6 +85,9 @@ class CallOptions(TypedDict, total=False):
     resumption is removed in the next protocol revision.
     """
 
+    headers: dict[str, str]
+    """Transport-layer hint: HTTP transports merge these onto the outgoing request; non-HTTP transports ignore."""
+
 
 @runtime_checkable
 class Outbound(Protocol):
@@ -111,7 +114,7 @@ async def send_raw_request(
         """
         ...
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         """Send a fire-and-forget notification."""
         ...
 

src/mcp/shared/inbound.py

@@ -8,6 +8,8 @@
 status.
 """
 
+import base64
+import re
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from types import MappingProxyType
@@ -34,13 +36,33 @@
     "ERROR_CODE_HTTP_STATUS",
     "InboundLadderRejection",
     "InboundModernRoute",
+    "MCP_METHOD_HEADER",
+    "MCP_NAME_HEADER",
     "MCP_PROTOCOL_VERSION_HEADER",
     "classify_inbound_request",
+    "encode_header_value",
 ]
 
 MCP_PROTOCOL_VERSION_HEADER: Final = "mcp-protocol-version"
 """Canonical lowercase name of the HTTP header carrying the MCP protocol version."""
 
+MCP_METHOD_HEADER: Final = "mcp-method"
+"""Canonical lowercase name of the HTTP header carrying the JSON-RPC method."""
+
+MCP_NAME_HEADER: Final = "mcp-name"
+"""Canonical lowercase name of the HTTP header carrying the resource name (tool/prompt/resource URI)."""
+
+_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 value == value.strip() and not _B64_SENTINEL.fullmatch(value):
+        return value
+    return f"=?base64?{base64.b64encode(value.encode('utf-8')).decode('ascii')}?="
+
+
 # INTERNAL_ERROR is deliberately unmapped (→ HTTP 200): the spec assigns no status to
 # -32603, and whether handler-origin errors get 5xx is an open S4 question — see TODO(L66).
 ERROR_CODE_HTTP_STATUS: Final[Mapping[int, int]] = MappingProxyType(

src/mcp/shared/jsonrpc_dispatcher.py

@@ -132,11 +132,11 @@ def request_id(self) -> RequestId | None:
     def can_send_request(self) -> bool:
         return self.transport.can_send_request and not self._closed
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         if self._closed:
             logger.debug("dropped %s: dispatch context closed", method)
             return
-        await self._dispatcher.notify(method, params, _related_request_id=self._request_id)
+        await self._dispatcher.notify(method, params, opts, _related_request_id=self._request_id)
 
     async def send_raw_request(
         self,
@@ -209,6 +209,7 @@ def _plan_outbound(related_request_id: RequestId | None, opts: CallOptions | Non
     cancel_on_abandon = opts.get("cancel_on_abandon", True)
     token = opts.get("resumption_token")
     on_token = opts.get("on_resumption_token")
+    headers = opts.get("headers")
     if related_request_id is not None:
         if token is not None or on_token is not None:
             logger.debug(
@@ -217,9 +218,11 @@ def _plan_outbound(related_request_id: RequestId | None, opts: CallOptions | Non
         return _OutboundPlan(ServerMessageMetadata(related_request_id=related_request_id), cancel_on_abandon)
     if token is not None or on_token is not None:
         return _OutboundPlan(
-            ClientMessageMetadata(resumption_token=token, on_resumption_token_update=on_token),
+            ClientMessageMetadata(resumption_token=token, on_resumption_token_update=on_token, headers=headers),
             cancel_on_abandon=False,
         )
+    if headers:
+        return _OutboundPlan(ClientMessageMetadata(headers=headers), cancel_on_abandon)
     return _OutboundPlan(None, cancel_on_abandon)
 
 
@@ -395,6 +398,7 @@ async def notify(
         self,
         method: str,
         params: Mapping[str, Any] | None,
+        opts: CallOptions | None = None,
         *,
         _related_request_id: RequestId | None = None,
     ) -> None:
@@ -414,7 +418,7 @@ async def notify(
         else:
             msg = JSONRPCNotification(jsonrpc="2.0", method=method)
         try:
-            await self._write(msg, _plan_outbound(_related_request_id, None).metadata)
+            await self._write(msg, _plan_outbound(_related_request_id, opts).metadata)
         except (anyio.BrokenResourceError, anyio.ClosedResourceError):
             # Transport tore down before run() noticed EOF.
             logger.debug("dropped %s: write stream closed", method)

src/mcp/shared/message.py

@@ -24,6 +24,8 @@ class ClientMessageMetadata:
 
     resumption_token: ResumptionToken | None = None
     on_resumption_token_update: Callable[[ResumptionToken], Awaitable[None]] | None = None
+    # Per-message HTTP headers (e.g. MCP-Protocol-Version, Mcp-Method) the transport should set.
+    headers: dict[str, str] | None = None
 
 
 @dataclass

src/mcp/shared/peer.py

@@ -81,8 +81,8 @@ async def send_raw_request(
     ) -> dict[str, Any]:
         return await self._outbound.send_raw_request(method, params, opts)
 
-    async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
-        await self._outbound.notify(method, params)
+    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
+        await self._outbound.notify(method, params, opts)
 
     @overload
     @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)