feat: DispatchContext.message_metadata passes SessionMessage.metadata through verbatim

Scaffolding for the swap: ServerRunner._make_context will read this to
populate ServerRequestContext.request / close_sse_stream / etc. the same
way the current Server._handle_request does.

Marked TODO(maxisbey): remove for Context rework — the redesign replaces
this with the per-transport context shape.

Author: maxisbey

src/mcp/shared/direct_dispatcher.py

@@ -24,6 +24,7 @@
 
 from mcp.shared.dispatcher import CallOptions, OnNotify, OnRequest, ProgressFnT
 from mcp.shared.exceptions import MCPError, NoBackChannelError
+from mcp.shared.message import MessageMetadata
 from mcp.shared.transport_context import TransportContext
 from mcp.types import INTERNAL_ERROR, REQUEST_TIMEOUT
 
@@ -47,6 +48,8 @@ class _DirectDispatchContext:
     transport: TransportContext
     _back_request: _Request
     _back_notify: _Notify
+    message_metadata: MessageMetadata = None  # TODO(maxisbey): remove for Context rework
+    """Always ``None``: in-memory dispatch attaches no transport metadata."""
     _on_progress: ProgressFnT | None = None
     cancel_requested: anyio.Event = field(default_factory=anyio.Event)
 

src/mcp/shared/dispatcher.py

@@ -22,6 +22,7 @@
 import anyio
 import anyio.abc
 
+from mcp.shared.message import MessageMetadata
 from mcp.shared.transport_context import TransportContext
 
 __all__ = [
@@ -107,6 +108,19 @@ def transport(self) -> TransportT_co:
         """Transport-specific metadata for this inbound message."""
         ...
 
+    @property
+    def message_metadata(self) -> MessageMetadata:
+        """The metadata the transport attached to this inbound message, if any.
+
+        This is `SessionMessage.metadata` passed through verbatim: HTTP
+        transports attach `ServerMessageMetadata` (the HTTP request, SSE
+        stream-close callbacks); stdio and in-memory dispatch attach nothing.
+        Tied to the `SessionMessage` wire format — goes away when transports
+        stop delivering messages that way.
+        """
+        # TODO(maxisbey): remove for context rework
+        ...
+
     @property
     def cancel_requested(self) -> anyio.Event:
         """Set when the peer sends ``notifications/cancelled`` for this request."""

src/mcp/shared/jsonrpc_dispatcher.py

@@ -116,6 +116,13 @@ class _JSONRPCDispatchContext(Generic[TransportT]):
     transport: TransportT
     _dispatcher: JSONRPCDispatcher[TransportT]
     _request_id: RequestId | None
+    message_metadata: MessageMetadata = None  # TODO(maxisbey): remove for Context rework
+    """The transport-attached `SessionMessage.metadata` for this inbound message.
+
+    Carries `ServerMessageMetadata` (HTTP request, SSE stream-close callbacks)
+    that the server lifts onto its request context. ``None`` for transports
+    that attach nothing.
+    """
     _progress_token: ProgressToken | None = None
     _closed: bool = False
     cancel_requested: anyio.Event = field(default_factory=anyio.Event)
@@ -398,6 +405,7 @@ def _dispatch_request(
             transport=transport_ctx,
             _dispatcher=self,
             _request_id=req.id,
+            message_metadata=metadata,
             _progress_token=progress_token,
         )
         scope = anyio.CancelScope()
@@ -439,7 +447,9 @@ def _dispatch_notification(
                     pass
             # fall through: progress is also teed to on_notify
         transport_ctx = self._transport_builder(None, metadata)
-        dctx = _JSONRPCDispatchContext(transport=transport_ctx, _dispatcher=self, _request_id=None)
+        dctx = _JSONRPCDispatchContext(
+            transport=transport_ctx, _dispatcher=self, _request_id=None, message_metadata=metadata
+        )
         self._spawn(on_notify, dctx, msg.method, msg.params, sender_ctx=sender_ctx)
 
     def _resolve_pending(self, request_id: RequestId | None, outcome: dict[str, Any] | ErrorData) -> None:

tests/shared/test_dispatcher.py

@@ -196,6 +196,16 @@ async def server_on_request(
     assert result == {"ok": True}
 
 
+@pytest.mark.anyio
+async def test_ctx_message_metadata_is_none_when_transport_attaches_nothing(pair_factory: PairFactory):
+    """Plain requests carry no transport metadata, so handlers see ``None``."""
+    async with running_pair(pair_factory) as (client, _server, _crec, srec):
+        with anyio.fail_after(5):
+            await client.send_raw_request("tools/call", None)
+    assert len(srec.contexts) == 1
+    assert srec.contexts[0].message_metadata is None
+
+
 @pytest.mark.anyio
 async def test_direct_send_raw_request_wraps_non_mcperror_exception_as_internal_error_with_cause():
     """DirectDispatcher-specific: the original exception is chained via __cause__."""

tests/shared/test_jsonrpc_dispatcher.py

@@ -22,7 +22,7 @@
     _outbound_metadata,
     _Pending,
 )
-from mcp.shared.message import ClientMessageMetadata, ServerMessageMetadata, SessionMessage
+from mcp.shared.message import ClientMessageMetadata, MessageMetadata, ServerMessageMetadata, SessionMessage
 from mcp.shared.transport_context import TransportContext
 from mcp.types import (
     CONNECTION_CLOSED,
@@ -274,6 +274,77 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) ->
             s.close()
 
 
+@pytest.mark.anyio
+async def test_ctx_message_metadata_carries_inbound_request_metadata():
+    """Transport-attached metadata (HTTP request, SSE close hooks) is readable off the dispatch context."""
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send)
+    metadata = ServerMessageMetadata(request_context="request-scoped-data")
+    seen: list[MessageMetadata] = []
+
+    async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
+        seen.append(ctx.message_metadata)
+        return {}
+
+    async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None:
+        raise NotImplementedError
+
+    try:
+        async with anyio.create_task_group() as tg:
+            await tg.start(server.run, on_request, on_notify)
+            await c2s_send.send(
+                SessionMessage(
+                    message=JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params=None),
+                    metadata=metadata,
+                )
+            )
+            with anyio.fail_after(5):
+                await s2c_recv.receive()  # response sent ⇒ the handler has run
+            tg.cancel_scope.cancel()
+    finally:
+        for s in (c2s_send, c2s_recv, s2c_send, s2c_recv):
+            s.close()
+    assert len(seen) == 1
+    assert seen[0] is metadata  # the exact object, passed through verbatim
+
+
+@pytest.mark.anyio
+async def test_ctx_message_metadata_carries_inbound_notification_metadata():
+    """Notifications get the same metadata pass-through as requests."""
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send)
+    metadata = ServerMessageMetadata(request_context="request-scoped-data")
+    seen: list[MessageMetadata] = []
+    notified = anyio.Event()
+
+    async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
+        raise NotImplementedError
+
+    async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None:
+        seen.append(ctx.message_metadata)
+        notified.set()
+
+    try:
+        async with anyio.create_task_group() as tg:
+            await tg.start(server.run, on_request, on_notify)
+            await c2s_send.send(
+                SessionMessage(
+                    message=JSONRPCNotification(jsonrpc="2.0", method="notifications/initialized", params=None),
+                    metadata=metadata,
+                )
+            )
+            with anyio.fail_after(5):
+                await notified.wait()
+            tg.cancel_scope.cancel()
+    finally:
+        for s in (c2s_send, c2s_recv, s2c_send, s2c_recv):
+            s.close()
+    assert len(seen) == 1
+    assert seen[0] is metadata
+
+
 @pytest.mark.anyio
 async def test_ctx_progress_with_only_progress_value_omits_total_and_message():
     received: list[tuple[float, float | None, str | None]] = []