Drop request-scoped messages a JSON response cannot carry at the router

In JSON-response mode a message tied to an in-flight request has no
wire form and no POST could replay it, so message_router now drops it
before storing or queueing rather than leaving it for a consumer to
discard. The per-request queue then carries exactly the one response,
and the JSON POST handler collapses from a discard loop to a single
receive with a real arm for a session torn down mid-request. This
removes the four coverage pragmas the discard loop needed.

Author: maxisbey

src/mcp/server/streamable_http.py

@@ -612,43 +612,24 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 session_message = SessionMessage(message, metadata=metadata)
                 await writer.send(session_message)
                 try:
-                    # Process messages from the request-specific stream
-                    # We need to collect all messages until we get a response
-                    response_message = None
-
-                    # Use similar approach to SSE writer for consistency
-                    async for event_message in request_stream_reader:  # pragma: no branch
-                        # If it's a response, this is what we're waiting for
-                        if isinstance(event_message.message, JSONRPCResponse | JSONRPCError):
-                            response_message = event_message.message
-                            break
-                        # For notifications and requests, keep waiting
-                        else:  # pragma: no cover
-                            logger.debug(f"received: {event_message.message.method}")
-
-                    # At this point we should have a response
-                    if response_message:
-                        # Create JSON response
-                        response = self._create_json_response(response_message)
-                        await response(scope, receive, send)
-                    else:  # pragma: no cover
-                        # This shouldn't happen in normal operation
-                        logger.error("No response message received before stream closed")
-                        response = self._create_error_response(
-                            "Error processing request: No response received",
-                            HTTPStatus.INTERNAL_SERVER_ERROR,
-                        )
-                        await response(scope, receive, send)
-                except Exception:  # pragma: no cover
-                    logger.exception("Error processing JSON response")
+                    # `message_router` deposits only this request's own response
+                    # here: anything else scoped to the request has no wire in
+                    # JSON-response mode.
+                    event_message = await request_stream_reader.receive()
+                except (anyio.EndOfStream, anyio.ClosedResourceError):
+                    # The stream closed with no response: the session was
+                    # terminated while this request was in flight.
+                    logger.debug(f"Session terminated with request {request_id} in flight; no response to send")
                     response = self._create_error_response(
-                        "Error processing request",
+                        "Session terminated before the request completed",
                         HTTPStatus.INTERNAL_SERVER_ERROR,
                         INTERNAL_ERROR,
                     )
-                    await response(scope, receive, send)
+                else:
+                    response = self._create_json_response(event_message.message)
                 finally:
                     await self._clean_up_memory_streams(request_id)
+                await response(scope, receive, send)
             else:
                 # Mint the priming event before any per-request state exists:
                 # `EventStore.store_event` is user code and may raise, in which
@@ -1049,7 +1030,19 @@ async def message_router():
                             )
                             and session_message.metadata.related_request_id is not None
                         ):
-                            target_request_id = str(session_message.metadata.related_request_id)
+                            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"
+                                )
+                                continue
+                            target_request_id = str(related_request_id)
 
                         request_stream_id = target_request_id if target_request_id is not None else GET_STREAM_KEY
 

tests/interaction/transports/test_streamable_http.py

@@ -134,6 +134,35 @@ async def test_json_response_streamable_http_rejects_request_scoped_server_reque
     assert exc_info.value.error.code == INVALID_REQUEST
 
 
+@requirement("transport:streamable-http:json-response-restrictions")
+@requirement("transport:streamable-http:unrelated-messages")
+@requirement("hosting:http:standalone-sse")
+async def test_json_response_streamable_http_delivers_only_unrelated_notifications() -> None:
+    """In JSON-response mode the call's own log notification has no stream to ride and never
+    reaches the client, while the tool result comes back as the JSON body and the unrelated
+    resource-updated notification arrives on the standalone stream. The handler writes both
+    notifications before returning, so once the result and the unrelated message are in, no
+    request-scoped message can still be in flight."""
+    received: list[IncomingMessage] = []
+    server_message_seen = anyio.Event()
+
+    async def collect(message: IncomingMessage) -> None:
+        received.append(message)
+        server_message_seen.set()
+
+    async with connect_over_streamable_http(_smoke_server(), json_response=True, message_handler=collect) as client:
+        with anyio.fail_after(5):
+            result = await client.call_tool("announce", {})
+            await server_message_seen.wait()
+
+    assert result == snapshot(
+        CallToolResult(content=[TextContent(text="announced")], structured_content={"result": "announced"})
+    )
+    assert received == snapshot(
+        [ResourceUpdatedNotification(params=ResourceUpdatedNotificationParams(uri="file:///watched.txt"))]
+    )
+
+
 @requirement("transport:streamable-http:notifications")
 @requirement("transport:streamable-http:unrelated-messages")
 @requirement("hosting:http:standalone-sse")

tests/server/test_streamable_http_router.py

@@ -118,6 +118,48 @@ async def asgi_send(message: Message) -> None:
     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": [
+            (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)
+            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
+
+
 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)