Stamp the JSON-mode verdict in one metadata factory

Route every ServerMessageMetadata the transport frames through a single
_message_metadata factory so the can_send_request stamp has one home a
future construction site cannot forget. The field becomes a plain bool
defaulting to True, dropping the None sentinel and the branch that read
it. Docstrings now defer to TransportContext.can_send_request for the
full rule instead of restating it, and the two direct-transport tests
share one fake-ASGI scaffold.

Author: maxisbey

src/mcp/server/streamable_http.py

@@ -44,7 +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 ServerMessageMetadata, SessionMessage
+from mcp.shared.message import CloseSSEStreamCallback, ServerMessageMetadata, SessionMessage
 
 logger = logging.getLogger(__name__)
 
@@ -174,12 +174,11 @@ def __init__(
             mcp_session_id: Optional session identifier for this connection.
                             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 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.
+                                    JSON body instead of an SSE stream, which removes
+                                    the request-scoped back-channel: a server-initiated
+                                    request tied to the call raises `NoBackChannelError`
+                                    and its notifications are dropped (see
+                                    `TransportContext.can_send_request`). Default is False.
             event_store: Event store for resumability support. If provided,
                         resumability will be enabled, allowing clients to
                         reconnect and resume messages.
@@ -217,15 +216,28 @@ def is_terminated(self) -> bool:
         """Check if this transport has been explicitly terminated."""
         return self._terminated
 
-    @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, which holds only the response. Stamped on each message's
-        `ServerMessageMetadata` so any dispatcher's builder reads it; whether a reply has a session
-        to land on is the session manager's fact, not the transport's, so it is not decided here.
+    def _message_metadata(
+        self,
+        request: Request,
+        *,
+        close_sse_stream: CloseSSEStreamCallback | None = None,
+        close_standalone_sse_stream: CloseSSEStreamCallback | None = None,
+        on_request_unanswered: Callable[[], Awaitable[None]] | None = None,
+    ) -> ServerMessageMetadata:
+        """The metadata this transport frames every inbound message with.
+
+        The one place `can_send_request` is stamped, so no construction site can
+        forget it: a JSON body carries only the response, so in JSON-response mode
+        the request-scoped channel cannot carry a server-initiated request (see
+        `TransportContext.can_send_request`).
         """
-        return not self.is_json_response_enabled
+        return ServerMessageMetadata(
+            request_context=request,
+            close_sse_stream=close_sse_stream,
+            close_standalone_sse_stream=close_standalone_sse_stream,
+            on_request_unanswered=on_request_unanswered,
+            can_send_request=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.
@@ -297,19 +309,14 @@ async def close_stream_callback() -> None:
             async def close_standalone_stream_callback() -> None:
                 self.close_standalone_sse_stream()
 
-            metadata = ServerMessageMetadata(
-                request_context=request,
+            metadata = self._message_metadata(
+                request,
                 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,
-                can_send_request=self._request_channel_can_send_request,
-            )
+            metadata = self._message_metadata(request, on_request_unanswered=end_stream)
 
         return SessionMessage(message, metadata=metadata)
 
@@ -577,10 +584,7 @@ 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, can_send_request=self._request_channel_can_send_request
-                )
-                session_message = SessionMessage(message, metadata=metadata)
+                session_message = SessionMessage(message, metadata=self._message_metadata(request))
                 await writer.send(session_message)
 
                 return
@@ -602,10 +606,8 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 )
                 request_stream_reader = self._request_streams[request_id][1]
                 # Process the message
-                metadata = ServerMessageMetadata(
-                    request_context=request,
-                    on_request_unanswered=partial(self._terminate_unanswered_request, message.id),
-                    can_send_request=self._request_channel_can_send_request,
+                metadata = self._message_metadata(
+                    request, on_request_unanswered=partial(self._terminate_unanswered_request, message.id)
                 )
                 session_message = SessionMessage(message, metadata=metadata)
                 await writer.send(session_message)
@@ -1030,15 +1032,10 @@ async def message_router():
                         ):
                             related_request_id = session_message.metadata.related_request_id
                             if self.is_json_response_enabled:
-                                # A JSON body carries exactly the one response: a
-                                # message that only rides this request's stream
-                                # has no wire form (and no POST could replay it),
-                                # so drop it before storing or queueing - never
-                                # park it in a queue nothing drains (#1764).
-                                logger.debug(
-                                    f"Dropped message related to request {related_request_id}: "
-                                    "a JSON response carries only its own response"
-                                )
+                                # A JSON body carries only the response: this message
+                                # has no wire form (nor a replay), so drop it before
+                                # storing or queueing rather than park it (#1764).
+                                logger.debug(f"Dropped message related to request {related_request_id} in JSON mode")
                                 continue
                             target_request_id = str(related_request_id)
 

src/mcp/shared/exceptions.py

@@ -55,13 +55,10 @@ def __str__(self) -> str:
 class NoBackChannelError(MCPError):
     """Raised when a server-initiated request has no channel that can deliver it.
 
-    The request-scoped channel is a dead end when the response has no room
-    (streamable HTTP in JSON-response mode, the 2026-07-28 single-exchange
-    entry), when the client's reply has nowhere to land (stateless HTTP), or
-    when the protocol forbids server-initiated requests (2026-07-28). Raised by
-    `DispatchContext.send_raw_request` when `can_send_request` is `False`, and
-    by a connection's standalone channel when it has none; serializes to an
-    `INVALID_REQUEST` error response.
+    Raised by `DispatchContext.send_raw_request` when its request-scoped channel
+    reports `TransportContext.can_send_request` as `False` (the cases are
+    documented on that field), and by a connection's standalone channel when it
+    has none; serializes to an `INVALID_REQUEST` error response.
     """
 
     def __init__(self, method: str):

src/mcp/shared/jsonrpc_dispatcher.py

@@ -189,14 +189,11 @@ def _default_transport_builder(metadata: MessageMetadata) -> TransportContext:
 
     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 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.
+    attached, so a transport whose response has no room for a server request
+    (streamable HTTP in JSON-response mode) needs no wiring from whoever drives
+    its streams.
     """
-    can_send_request = True
-    if isinstance(metadata, ServerMessageMetadata) and metadata.can_send_request is not None:
-        can_send_request = metadata.can_send_request
+    can_send_request = metadata.can_send_request if isinstance(metadata, ServerMessageMetadata) else True
     return TransportContext(kind="jsonrpc", can_send_request=can_send_request)
 
 

src/mcp/shared/message.py

@@ -46,9 +46,10 @@ class ServerMessageMetadata:
     # 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
+    # can deliver a server-initiated request (see
+    # `TransportContext.can_send_request`); a transport that says nothing leaves
+    # it True.
+    can_send_request: bool = True
 
 
 MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None

tests/server/test_streamable_http_router.py

@@ -25,6 +25,25 @@ async def replay_events_after(self, last_event_id: EventId, send_callback: Event
         raise NotImplementedError
 
 
+class _AsgiPost:
+    """A one-shot POST driven straight at `handle_request`, capturing what the transport sends."""
+
+    def __init__(self, body: bytes, headers: list[tuple[bytes, bytes]]) -> None:
+        self.scope: Scope = {"type": "http", "method": "POST", "path": "/", "query_string": b"", "headers": headers}
+        self.sent: list[Message] = []
+        self._body = body
+        self._body_sent = False
+
+    async def receive(self) -> Message:
+        if not self._body_sent:
+            self._body_sent = True
+            return {"type": "http.request", "body": self._body, "more_body": False}
+        raise NotImplementedError
+
+    async def send(self, message: Message) -> None:
+        self.sent.append(message)
+
+
 @pytest.mark.anyio
 async def test_router_unconsumed_request_stream_does_not_block_siblings() -> None:
     """A response whose `sse_writer` is not yet receiving must not park the router (#1764).
@@ -73,35 +92,18 @@ async def test_priming_store_failure_leaves_no_per_request_state() -> None:
         event_store=_PrimingFailingStore(),
     )
 
-    body = b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}'
-    scope: Scope = {
-        "type": "http",
-        "method": "POST",
-        "path": "/",
-        "query_string": b"",
-        "headers": [
+    post = _AsgiPost(
+        b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}',
+        [
             (b"accept", b"application/json, text/event-stream"),
             (b"content-type", b"application/json"),
             (b"mcp-protocol-version", b"2025-11-25"),
         ],
-    }
-    body_sent = False
-
-    async def receive() -> Message:
-        nonlocal body_sent
-        if not body_sent:
-            body_sent = True
-            return {"type": "http.request", "body": body, "more_body": False}
-        raise NotImplementedError
-
-    sent: list[Message] = []
-
-    async def asgi_send(message: Message) -> None:
-        sent.append(message)
+    )
 
     async with transport.connect() as (read_stream, _write_stream):
         async with anyio.create_task_group() as tg:
-            tg.start_soon(transport.handle_request, scope, receive, asgi_send)
+            tg.start_soon(transport.handle_request, post.scope, post.receive, post.send)
             with anyio.fail_after(5):
                 forwarded = await read_stream.receive()
             assert isinstance(forwarded, Exception)
@@ -110,49 +112,32 @@ async def asgi_send(message: Message) -> None:
         assert transport._request_streams == {}
         assert transport._sse_stream_writers == {}
 
-    assert sent[0]["type"] == "http.response.start"
-    assert sent[0]["status"] == 500
-    body = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
+    assert post.sent[0]["type"] == "http.response.start"
+    assert post.sent[0]["status"] == 500
+    body = b"".join(m.get("body", b"") for m in post.sent if m["type"] == "http.response.body")
     assert b"backend unavailable" not in body
 
 
 @pytest.mark.anyio
 async def test_json_post_answers_500_when_session_terminates_mid_request() -> None:
     """A JSON-mode POST whose session is torn down before the handler answers gets a 500, not a stall."""
     transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True)
-    body = b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}'
-    scope: Scope = {
-        "type": "http",
-        "method": "POST",
-        "path": "/",
-        "query_string": b"",
-        "headers": [
+    post = _AsgiPost(
+        b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}',
+        [
             (b"accept", b"application/json"),
             (b"content-type", b"application/json"),
             (b"mcp-session-id", b"sid"),
             (b"mcp-protocol-version", b"2025-11-25"),
         ],
-    }
-    body_sent = False
-
-    async def receive() -> Message:
-        nonlocal body_sent
-        if not body_sent:
-            body_sent = True
-            return {"type": "http.request", "body": body, "more_body": False}
-        raise NotImplementedError
-
-    sent: list[Message] = []
-
-    async def asgi_send(message: Message) -> None:
-        sent.append(message)
+    )
 
     async with transport.connect() as (read_stream, _write_stream):
         async with anyio.create_task_group() as tg:
-            tg.start_soon(transport.handle_request, scope, receive, asgi_send)
+            tg.start_soon(transport.handle_request, post.scope, post.receive, post.send)
             with anyio.fail_after(5):
                 await read_stream.receive()  # the request reached the session; the POST is parked
             await transport.terminate()
 
-    assert sent[0]["type"] == "http.response.start"
-    assert sent[0]["status"] == 500
+    assert post.sent[0]["type"] == "http.response.start"
+    assert post.sent[0]["status"] == 500