Move the JSON-mode back-channel verdict onto the message metadata

Instead of the transport supplying a `transport_builder` that the
session manager has to ferry into the dispatcher, the transport now
states its verdict on the message it already emits:
ServerMessageMetadata grows `can_send_request`, which
StreamableHTTPServerTransport stamps as
`mcp_session_id is not None and not is_json_response_enabled`, and the
dispatcher's default builder honors it.

This drops the public transport_context_for method, the serve_loop
keyword, and the manager's stateless special-case builder: both legs
are now correct with no wiring at all, and so is a transport driven by
hand. The metadata already carries transport facts the dispatcher
consumes (related_request_id, on_request_unanswered), so the verdict
follows an existing channel rather than opening a new one.

Author: maxisbey

src/mcp/server/runner.py

@@ -15,7 +15,7 @@
 
 import contextvars
 import logging
-from collections.abc import AsyncIterator, Awaitable, Callable, Mapping
+from collections.abc import AsyncIterator, Awaitable, Mapping
 from contextlib import asynccontextmanager
 from dataclasses import KW_ONLY, dataclass, replace
 from functools import cached_property, partial
@@ -473,7 +473,6 @@ async def serve_loop(
     session_id: str | None = None,
     init_options: InitializationOptions | None = None,
     raise_exceptions: bool = False,
-    transport_builder: Callable[[MessageMetadata], TransportContext] | None = None,
 ) -> None:
     """Drive ``server`` in handshake-only loop mode over a stream pair until the channel closes.
 
@@ -483,20 +482,10 @@ async def serve_loop(
     this; `Server.run` drives `serve_dual_era_loop`, which extends the same
     dispatcher recipe (notably the `inline_methods={"initialize"}` rule) with
     era routing.
-
-    Args:
-        transport_builder: Builds each inbound message's `TransportContext` from
-            its `SessionMessage.metadata`. Omitted, every message reads as a full
-            duplex pipe (`can_send_request=True`) - the truth for a plain stream
-            pair; a transport whose response cannot carry a server-initiated
-            request (streamable HTTP in JSON-response mode) passes its own
-            `StreamableHTTPServerTransport.transport_context_for` so the
-            request-scoped channel refuses instead of stalling.
     """
     dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
         read_stream,
         write_stream,
-        transport_builder=transport_builder,
         raise_handler_exceptions=raise_exceptions,
         # Handle `initialize` inline so a client that pipelines it with the
         # next request (spec: SHOULD NOT, not MUST NOT) sees the initialized

src/mcp/server/streamable_http.py

@@ -44,8 +44,7 @@
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
-from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
-from mcp.shared.transport_context import TransportContext
+from mcp.shared.message import ServerMessageMetadata, SessionMessage
 
 logger = logging.getLogger(__name__)
 
@@ -176,9 +175,9 @@ def __init__(
                             Must contain only visible ASCII characters (0x21-0x7E).
             is_json_response_enabled: If True, answer each request POST with a single
                                     JSON body instead of an SSE stream. The body carries
-                                    only the response, so `transport_context_for` reports
-                                    `can_send_request=False` for it: a server-initiated
-                                    request on the request-scoped channel raises
+                                    only the response, so this transport marks its messages
+                                    as having no request-scoped back-channel: a
+                                    server-initiated request on that channel raises
                                     `NoBackChannelError` and request-scoped notifications
                                     are dropped. Default is False.
             event_store: Event store for resumability support. If provided,
@@ -218,24 +217,14 @@ def is_terminated(self) -> bool:
         """Check if this transport has been explicitly terminated."""
         return self._terminated
 
-    def transport_context_for(self, metadata: MessageMetadata) -> TransportContext:
-        """Build the `TransportContext` for an inbound message this transport delivered.
-
-        The transport owns what a request's response can carry, so it supplies
-        the loop dispatcher's `transport_builder`. A JSON body holds exactly one
-        JSON-RPC response, so in JSON-response mode the request-scoped channel has
-        no room for a server-initiated request and `can_send_request` is `False`:
-        `ctx.elicit()` raises `NoBackChannelError` at once instead of parking a
-        waiter no reply can reach. An SSE response streams whatever the handler
-        emits. The connection's standalone GET stream is a separate channel and
-        is not described here.
+    @property
+    def _request_channel_can_send_request(self) -> bool:
+        """Whether the request-scoped channel of a message this transport delivers can carry a
+        server-initiated request. It cannot in JSON-response mode (the POST is answered with one
+        JSON body) nor with no session (the client's reply would have no session to land on);
+        stamped on each message's `ServerMessageMetadata` so any dispatcher's builder reads it.
         """
-        request = metadata.request_context if isinstance(metadata, ServerMessageMetadata) else None
-        return TransportContext(
-            kind="streamable-http",
-            can_send_request=not self.is_json_response_enabled,
-            headers=request.headers if isinstance(request, Request) else None,
-        )
+        return self.mcp_session_id is not None and not self.is_json_response_enabled
 
     def close_sse_stream(self, request_id: RequestId) -> None:
         """Close SSE connection for a specific request without terminating the stream.
@@ -312,9 +301,14 @@ async def close_standalone_stream_callback() -> None:
                 close_sse_stream=close_stream_callback,
                 close_standalone_sse_stream=close_standalone_stream_callback,
                 on_request_unanswered=end_stream,
+                can_send_request=self._request_channel_can_send_request,
             )
         else:
-            metadata = ServerMessageMetadata(request_context=request, on_request_unanswered=end_stream)
+            metadata = ServerMessageMetadata(
+                request_context=request,
+                on_request_unanswered=end_stream,
+                can_send_request=self._request_channel_can_send_request,
+            )
 
         return SessionMessage(message, metadata=metadata)
 
@@ -582,7 +576,9 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 await response(scope, receive, send)
 
                 # Process the message after sending the response
-                metadata = ServerMessageMetadata(request_context=request)
+                metadata = ServerMessageMetadata(
+                    request_context=request, can_send_request=self._request_channel_can_send_request
+                )
                 session_message = SessionMessage(message, metadata=metadata)
                 await writer.send(session_message)
 
@@ -608,6 +604,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 metadata = ServerMessageMetadata(
                     request_context=request,
                     on_request_unanswered=partial(self._terminate_unanswered_request, message.id),
+                    can_send_request=self._request_channel_can_send_request,
                 )
                 session_message = SessionMessage(message, metadata=metadata)
                 await writer.send(session_message)
@@ -992,12 +989,6 @@ async def connect(
     ]:
         """Context manager that provides read and write streams for a connection.
 
-        Drive the yielded streams through a `JSONRPCDispatcher` built with
-        `transport_context_for` as its `transport_builder`, as the session
-        manager does: that is what makes the request-scoped channel refuse
-        server-initiated requests in JSON-response mode instead of parking
-        them where nothing can deliver them.
-
         Yields:
             Tuple of (read_stream, write_stream) for bidirectional communication
         """

src/mcp/server/streamable_http_manager.py

@@ -6,7 +6,6 @@
 import logging
 from collections import deque
 from collections.abc import AsyncIterator
-from dataclasses import replace
 from typing import TYPE_CHECKING, Any, Final
 from uuid import uuid4
 
@@ -215,14 +214,11 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                     read_stream,
                     write_stream,
                     inline_methods=frozenset({"initialize"}),
-                    # The transport says what its response stream can carry; on
-                    # top of that, no session ID means the client's reply to a
-                    # server request has nowhere to land, so the request-scoped
-                    # channel refuses (`NoBackChannelError`) even in SSE mode
-                    # while still forwarding notifications.
-                    transport_builder=lambda md: replace(
-                        http_transport.transport_context_for(md), can_send_request=False
-                    ),
+                    # No `transport_builder`: with no session ID the transport
+                    # marks each message's request-scoped channel unable to
+                    # carry a server-initiated request (the client's reply has
+                    # nowhere to land), so requests raise `NoBackChannelError`
+                    # while notifications still flow.
                 )
                 # Born-ready, no standalone channel: the legacy stateless path
                 # never opens a GET stream and need not see `initialize`. The
@@ -324,14 +320,13 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                                 # Drive via `serve_loop` (not `Server.run()`) so the
                                 # manager's already-entered lifespan is reused
                                 # rather than re-entered per session; the transport
-                                # rules on what each request's response can carry.
+                                # marks each message with what its channel can carry.
                                 await serve_loop(
                                     self.app,
                                     read_stream,
                                     write_stream,
                                     lifespan_state=self._lifespan_state,
                                     session_id=http_transport.mcp_session_id,
-                                    transport_builder=http_transport.transport_context_for,
                                 )
 
                             if idle_scope.cancelled_caught:

src/mcp/shared/jsonrpc_dispatcher.py

@@ -184,8 +184,21 @@ def close(self) -> None:
         self._closed = True
 
 
-def _default_transport_builder(_meta: MessageMetadata) -> TransportContext:
-    return TransportContext(kind="jsonrpc", can_send_request=True)
+def _default_transport_builder(metadata: MessageMetadata) -> TransportContext:
+    """The `TransportContext` for a message, honoring the transport's own verdict when it stamps one.
+
+    A message reads as riding a full duplex pipe (`can_send_request=True`)
+    unless the transport that framed it says otherwise on the metadata it
+    attached: streamable HTTP marks JSON-response-mode and sessionless
+    messages `ServerMessageMetadata(can_send_request=False)`, so their
+    request-scoped channel raises `NoBackChannelError` instead of parking a
+    waiter no reply can reach - with no wiring needed from whoever drives the
+    streams.
+    """
+    can_send_request = True
+    if isinstance(metadata, ServerMessageMetadata) and metadata.can_send_request is not None:
+        can_send_request = metadata.can_send_request
+    return TransportContext(kind="jsonrpc", can_send_request=can_send_request)
 
 
 def _shielded_progress(fn: ProgressFnT) -> ProgressFnT:

src/mcp/shared/message.py

@@ -45,6 +45,10 @@ class ServerMessageMetadata:
     # (e.g. it was cancelled), for a transport whose wire must still end the
     # request even though no response is written.
     on_request_unanswered: Callable[[], Awaitable[None]] | None = None
+    # The transport's verdict on whether this message's request-scoped channel
+    # can deliver a server-initiated request; `None` when the transport does
+    # not say, which the default `TransportContext` builder reads as True.
+    can_send_request: bool | None = None
 
 
 MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None

tests/server/test_streamable_http_router.py

@@ -3,7 +3,6 @@
 import anyio
 import pytest
 from mcp_types import JSONRPCMessage, JSONRPCResponse
-from starlette.requests import Request
 from starlette.types import Message, Scope
 
 from mcp.server.streamable_http import (
@@ -15,8 +14,7 @@
     StreamableHTTPServerTransport,
     StreamId,
 )
-from mcp.shared.message import ServerMessageMetadata, SessionMessage
-from mcp.shared.transport_context import TransportContext
+from mcp.shared.message import SessionMessage
 
 
 class _PrimingFailingStore(EventStore):
@@ -158,18 +156,3 @@ async def asgi_send(message: Message) -> None:
 
     assert sent[0]["type"] == "http.response.start"
     assert sent[0]["status"] == 500
-
-
-def test_transport_context_reports_response_mode_and_request_headers() -> None:
-    """The transport's own verdict: no request-scoped requests in JSON mode, headers off the carried Request."""
-    json_transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True)
-    assert json_transport.transport_context_for(None) == TransportContext(
-        kind="streamable-http", can_send_request=False
-    )
-    sse_transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=False)
-    assert sse_transport.transport_context_for(None) == TransportContext(kind="streamable-http", can_send_request=True)
-
-    request = Request({"type": "http", "method": "POST", "path": "/", "headers": [(b"x-test", b"1")]})
-    context = sse_transport.transport_context_for(ServerMessageMetadata(request_context=request))
-    assert context.headers is not None
-    assert context.headers["x-test"] == "1"

tests/shared/test_jsonrpc_dispatcher.py

@@ -1329,6 +1329,45 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) ->
     assert seen[0] is metadata  # the exact object, passed through verbatim
 
 
+@pytest.mark.anyio
+async def test_transport_stamped_can_send_request_makes_the_request_channel_refuse():
+    """A transport that marks a message `can_send_request=False` on its metadata gets a request-scoped
+    channel that raises `NoBackChannelError` immediately - the default builder reads the transport's
+    verdict off the message, so no driver has to wire it."""
+    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)
+    outcomes: list[bool | str] = []
+
+    async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
+        outcomes.append(ctx.can_send_request)
+        try:
+            await ctx.send_raw_request("elicitation/create", {})
+        except NoBackChannelError as exc:
+            outcomes.append(exc.method)
+        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=ServerMessageMetadata(can_send_request=False),
+                )
+            )
+            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 outcomes == [False, "elicitation/create"]
+
+
 @pytest.mark.anyio
 async def test_ctx_message_metadata_carries_inbound_notification_metadata():
     """Notifications get the same metadata pass-through as requests."""