Return 405 for listen-mode GET on stateless streamable HTTP transport

src/mcp/server/streamable_http.py

@@ -169,6 +169,7 @@ def __init__(
         security_settings: TransportSecuritySettings | None = None,
         retry_interval: int | None = None,
         idle_timeout: float | None = None,
+        stateless: bool = False,
     ) -> None:
         """Initialize a new StreamableHTTP server transport.
 
@@ -196,6 +197,10 @@ def __init__(
                          (available once `connect()` has been entered) around the session's
                          message loop to end the session when it fires. Default is None: no
                          `idle_scope`, the session never expires.
+            stateless: If True, this transport never sends server-initiated messages, so the
+                      listen-mode GET (`Accept: text/event-stream`) answers 405 Method Not
+                      Allowed instead of opening a stream that would otherwise sit open
+                      forever with nothing to write. Default is False.
 
         Raises:
             ValueError: If the session ID contains invalid characters, or if `idle_timeout`
@@ -211,6 +216,7 @@ def __init__(
         self._event_store = event_store
         self._security = TransportSecurityMiddleware(security_settings)
         self._retry_interval = retry_interval
+        self._stateless = stateless
         self._request_streams: dict[
             RequestId,
             tuple[
@@ -736,6 +742,18 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
         if writer is None:  # pragma: no cover
             raise ValueError("No read stream writer available. Ensure connect() is called first.")
 
+        if self._stateless:
+            # A stateless transport never sends server-initiated messages, so the
+            # listen stream would sit open with nothing to write. Per the
+            # 2025-03-26 spec, a server that doesn't support server-initiated
+            # messages MUST answer the listen-mode GET with 405.
+            response = self._create_error_response(
+                "Method Not Allowed: This server does not support server-initiated messages",
+                HTTPStatus.METHOD_NOT_ALLOWED,
+            )
+            await response(request.scope, request.receive, send)
+            return
+
         # Validate Accept header - must include text/event-stream
         _, has_sse = check_accept_headers(request)
 

src/mcp/server/streamable_http_manager.py

@@ -214,6 +214,7 @@ async def _handle_stateless_request(
             is_json_response_enabled=self.json_response,
             event_store=None,  # No event store in stateless mode
             security_settings=self.security_settings,
+            stateless=True,
         )
 
         # Start server in a new task

tests/shared/test_streamable_http.py

@@ -333,6 +333,7 @@ async def running_app(
     event_store: EventStore | None = None,
     retry_interval: int | None = None,
     server: Server[Any] | None = None,
+    stateless: bool = False,
 ) -> AsyncIterator[Starlette]:
     """Serve the test server's streamable HTTP app in process for the duration.
 
@@ -341,6 +342,7 @@ async def running_app(
         event_store: Optional event store for testing resumability.
         retry_interval: Retry interval in milliseconds for SSE polling.
         server: Server to mount; defaults to the file's shared test server.
+        stateless: If True, run the session manager in stateless mode.
     """
     # DNS-rebinding protection validates Host/Origin headers against a network attack that cannot
     # exist for an in-process app; the protection itself is pinned by
@@ -349,6 +351,7 @@ async def running_app(
         app=server if server is not None else _create_server(),
         event_store=event_store,
         json_response=is_json_response_enabled,
+        stateless=stateless,
         security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False),
         retry_interval=retry_interval,
     )
@@ -384,6 +387,13 @@ async def json_app() -> AsyncIterator[Starlette]:
         yield app
 
 
+@pytest.fixture
+async def stateless_app() -> AsyncIterator[Starlette]:
+    """The test server's app in stateless mode."""
+    async with running_app(stateless=True) as app:
+        yield app
+
+
 @pytest.fixture
 def event_store() -> SimpleEventStore:
     """Create a test event store."""
@@ -902,6 +912,15 @@ async def test_get_sse_stream(basic_app: Starlette) -> None:
             assert second_get.status_code == 409
 
 
+@pytest.mark.anyio
+async def test_get_sse_stream_returns_405_when_stateless(stateless_app: Starlette) -> None:
+    """A stateless server never pushes, so the listen-mode GET answers 405 instead of hanging (#3492)."""
+    async with make_client(stateless_app) as client:
+        with anyio.fail_after(5):
+            get_response = await client.get("/mcp", headers={"Accept": "text/event-stream"})
+        assert get_response.status_code == 405
+
+
 @pytest.mark.anyio
 async def test_get_validation(basic_app: Starlette) -> None:
     """A GET without an Accept header covering text/event-stream is rejected with 406."""