Address review: serve_loop helper, bare 405, INVALID_REQUEST split, migration.md fixes

Four review findings, reworked for by-construction:

1. Non-POST -> bare 405 + Allow header (HTTP-layer rejection happens before
   JSON-RPC parsing, so it doesn't go through the JSON-RPC status table).
   _write stays table-pure with no override parameter.

2. Stateful HTTP no longer enters lifespan twice. New serve_loop() free
   function in runner.py owns the loop-mode dispatcher recipe (notably
   inline_methods={'initialize'}); both Server.run() and the manager's
   stateful path call it. Manager owns lifespan for all HTTP modes;
   Server.run() owns it for stdio/memory. Stateful sessions also now have
   their session_id threaded onto Connection.

3. Well-formed JSON that isn't a single request object (notification,
   response, batch) returns INVALID_REQUEST instead of PARSE_ERROR.
   json.loads and model_validate are now distinct steps; INVALID_REQUEST
   added to ERROR_CODE_HTTP_STATUS. The classifier now receives the decoded
   body directly rather than a reconstructed dict.

4. migration.md: fixed the broken ctx.connection example (replaced with
   prose; the high-level Context surface for this is being finalised),
   documented stateless kwarg + StatelessModeNotSupported removals,
   updated the ServerSession constructor example, and extended the
   lifespan-once entry to cover stateful as well.

Author: maxisbey

docs/migration.md

@@ -501,17 +501,17 @@ app = Starlette(routes=[Mount("/", app=mcp.streamable_http_app(json_response=Tru
 
 If you were mutating these via `mcp.settings` after construction (e.g., `mcp.settings.port = 9000`), pass them to `run()` / `sse_app()` / `streamable_http_app()` instead — these fields no longer exist on `Settings`. The `debug` and `log_level` parameters remain on the constructor.
 
-### `stateless_http=True`: lifespan now entered once at startup
+### Streamable HTTP: lifespan now entered once at manager startup
 
-When serving streamable HTTP with `stateless_http=True`, the server's `lifespan` context manager is now entered once when `StreamableHTTPSessionManager.run()` starts, and the resulting state is shared across all requests. Previously each incoming request entered (and exited) `lifespan` independently.
+When serving streamable HTTP (stateful or `stateless_http=True`), the server's `lifespan` context manager is now entered once when `StreamableHTTPSessionManager.run()` starts, and the resulting state is shared across all sessions and requests. Previously each session (stateful) or each request (stateless) entered and exited `lifespan` independently.
 
-Lifespans that set up process-wide state (connection pools, caches, background tasks) are unaffected — they now run once instead of on every request. If your lifespan was acquiring per-connection resources, move that into the handler and register cleanup on the per-connection `exit_stack`:
+Lifespans that set up process-wide state (connection pools, caches, background tasks) are unaffected — they now run once instead of per session/request. If your lifespan was acquiring per-connection resources, move that acquisition into the handler body; per-connection cleanup belongs on the connection's `exit_stack` (the public surface for reaching it from high-level `@mcp.tool()` handlers is being finalised as part of the public-surface review).
 
-```python
-async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
-    db = await ctx.connection.exit_stack.enter_async_context(open_db())
-    ...
-```
+### `Server.run()` no longer takes a `stateless` flag; `StatelessModeNotSupported` removed
+
+The `stateless: bool` parameter on the lowlevel `Server.run()` has been removed. Stateless serving is now a property of how the connection is constructed (the streamable-HTTP manager builds a born-ready `Connection` per request), not a flag the loop driver inspects.
+
+`StatelessModeNotSupported` has been removed. Server-initiated requests that have no channel to travel on now raise `NoBackChannelError` (an `MCPError` subclass) — the same exception regardless of why the channel is absent. If you were catching `StatelessModeNotSupported`, catch `NoBackChannelError` instead.
 
 ### `MCPServer.get_context()` removed
 
@@ -1228,8 +1228,8 @@ from mcp.server import ServerRequestContext
 session = ServerSession(read_stream, write_stream, init_options, stateless=False)
 
 # After (v2)
-session = ServerSession(dispatcher, connection, stateless=False)
-# where `dispatcher` is a JSONRPCDispatcher and `connection` is a Connection
+session = ServerSession(request_outbound, connection)
+# where `request_outbound` is an Outbound and `connection` is a Connection
 ```
 
 In practice, replace direct `ServerSession` use with `Server.run(read_stream, write_stream, init_options)` and let the framework wire it up.

src/mcp/server/_streamable_http_modern.py

@@ -33,7 +33,7 @@
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata
 from mcp.shared.transport_context import TransportContext
 from mcp.types import (
-    METHOD_NOT_FOUND,
+    INVALID_REQUEST,
     PARSE_ERROR,
     ClientCapabilities,
     ErrorData,
@@ -148,23 +148,34 @@ async def handle_modern_request(
     # TODO(D-005a): validate Accept once the JSON-vs-SSE response mode is settled.
 
     if request.method != "POST":
-        rej = JSONRPCError(
-            jsonrpc="2.0",
-            id=None,
-            error=ErrorData(code=METHOD_NOT_FOUND, message=f"HTTP {request.method} not supported on this endpoint"),
-        )
-        await _write(rej, scope, receive, send, extra_headers={"Allow": "POST"})
+        # HTTP-layer rejection (Allow accompanies 405 per RFC 9110) — happens
+        # before JSON-RPC parsing, so it doesn't go through `_write`.
+        await Response(status_code=405, headers={"Allow": "POST"})(scope, receive, send)
         return
 
     body = await request.body()
     try:
-        req = JSONRPCRequest.model_validate_json(body)
-    except ValidationError:
+        decoded = json.loads(body)
+    except json.JSONDecodeError:
         rej = JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error"))
         await _write(rej, scope, receive, send)
         return
+    try:
+        req = JSONRPCRequest.model_validate(decoded)
+    except ValidationError:
+        # Well-formed JSON that isn't a single request object (a notification,
+        # a response, a batch). TODO(L57): the 2026 HTTP path carries no
+        # client→server notifications (cancellation is via SSE close), so a
+        # plain rejection is currently sufficient.
+        rej = JSONRPCError(
+            jsonrpc="2.0",
+            id=None,
+            error=ErrorData(code=INVALID_REQUEST, message="Body must be a single JSON-RPC request object"),
+        )
+        await _write(rej, scope, receive, send)
+        return
 
-    verdict = classify_inbound_request({"method": req.method, "params": req.params}, headers=dict(request.headers))
+    verdict = classify_inbound_request(decoded, headers=dict(request.headers))
     if isinstance(verdict, InboundLadderRejection):
         rej = JSONRPCError(
             jsonrpc="2.0", id=req.id, error=ErrorData(code=verdict.code, message=verdict.message, data=verdict.data)

src/mcp/server/lowlevel/server.py

@@ -56,17 +56,14 @@ async def main():
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
 from mcp.server.auth.routes import build_resource_metadata_url, create_auth_routes, create_protected_resource_routes
 from mcp.server.auth.settings import AuthSettings
-from mcp.server.connection import Connection
 from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
-from mcp.server.runner import serve_connection
+from mcp.server.runner import serve_loop
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._stream_protocols import ReadStream, WriteStream
-from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.message import SessionMessage
-from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import MODERN_PROTOCOL_VERSIONS
 
 logger = logging.getLogger(__name__)
@@ -442,27 +439,18 @@ async def run(
     ) -> None:
         """Serve a single connection over the given streams until the read side closes.
 
-        Thin wrapper over `serve_connection` (L28): enters the server lifespan,
-        builds a `JSONRPCDispatcher` over the streams and a loop-mode
-        `Connection` on it, then drives the dispatcher to completion.
+        Thin wrapper over `serve_loop` (L28): enters the server lifespan,
+        then drives the loop. Transports with their own lifespan owner
+        (the streamable-HTTP manager) call `serve_loop` directly instead.
         """
         async with self.lifespan(self) as lifespan_context:
-            dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
+            await serve_loop(
+                self,
                 read_stream,
                 write_stream,
-                raise_handler_exceptions=raise_exceptions,
-                # Handle `initialize` inline so a client that pipelines it with
-                # the next request (spec says SHOULD NOT, not MUST NOT) sees
-                # the initialized state instead of failing the init-gate.
-                inline_methods=frozenset({"initialize"}),
-            )
-            connection = Connection.for_loop(dispatcher)
-            await serve_connection(
-                self,
-                dispatcher,
-                connection=connection,
                 lifespan_state=lifespan_context,
                 init_options=initialization_options,
+                raise_exceptions=raise_exceptions,
             )
 
     def streamable_http_app(

src/mcp/server/runner.py

@@ -30,10 +30,11 @@
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
 from mcp.shared._otel import extract_trace_context, otel_span
+from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.dispatcher import DispatchContext, Dispatcher, DispatchMiddleware, OnNotify, OnRequest
 from mcp.shared.exceptions import MCPError
-from mcp.shared.jsonrpc_dispatcher import handler_exception_to_error_data
-from mcp.shared.message import ServerMessageMetadata
+from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher, handler_exception_to_error_data
+from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
 from mcp.types import (
@@ -436,6 +437,39 @@ async def serve_connection(
         await aclose_shielded(connection)
 
 
+async def serve_loop(
+    server: Server[LifespanT],
+    read_stream: ReadStream[SessionMessage | Exception],
+    write_stream: WriteStream[SessionMessage],
+    *,
+    lifespan_state: LifespanT,
+    session_id: str | None = None,
+    init_options: InitializationOptions | None = None,
+    raise_exceptions: bool = False,
+) -> None:
+    """Drive ``server`` in loop mode over a stream pair until the channel closes.
+
+    Builds the loop-mode `JSONRPCDispatcher` + `Connection` and hands them to
+    `serve_connection`, so the dispatcher-construction recipe (notably the
+    `inline_methods={"initialize"}` rule) lives in one place. Callers that own
+    a lifespan (the streamable-HTTP manager) pass it in; callers that don't
+    (`Server.run` for stdio/memory) enter the lifespan and then call this.
+    """
+    dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
+        read_stream,
+        write_stream,
+        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
+        # state instead of failing the init-gate.
+        inline_methods=frozenset({"initialize"}),
+    )
+    connection = Connection.for_loop(dispatcher, session_id=session_id)
+    await serve_connection(
+        server, dispatcher, connection=connection, lifespan_state=lifespan_state, init_options=init_options
+    )
+
+
 async def serve_one(
     server: Server[LifespanT],
     request: JSONRPCRequest,

src/mcp/server/streamable_http_manager.py

@@ -17,7 +17,7 @@
 from mcp.server._streamable_http_modern import handle_modern_request
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
 from mcp.server.connection import Connection
-from mcp.server.runner import serve_connection
+from mcp.server.runner import serve_connection, serve_loop
 from mcp.server.streamable_http import (
     MCP_PROTOCOL_VERSION_HEADER,
     MCP_SESSION_ID_HEADER,
@@ -296,7 +296,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                         task_status.started()
                         try:
                             # Use a cancel scope for idle timeout — when the
-                            # deadline passes the scope cancels app.run() and
+                            # deadline passes the scope cancels the loop and
                             # execution continues after the ``with`` block.
                             # Incoming requests push the deadline forward.
                             idle_scope = anyio.CancelScope()
@@ -305,10 +305,15 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                                 http_transport.idle_scope = idle_scope
 
                             with idle_scope:
-                                await self.app.run(
+                                # Drive via `serve_loop` (not `Server.run()`) so the
+                                # manager's already-entered lifespan is reused
+                                # rather than re-entered per session.
+                                await serve_loop(
+                                    self.app,
                                     read_stream,
                                     write_stream,
-                                    self.app.create_initialization_options(),
+                                    lifespan_state=self._lifespan_state,
+                                    session_id=http_transport.mcp_session_id,
                                 )
 
                             if idle_scope.cancelled_caught:

src/mcp/shared/inbound.py

@@ -22,6 +22,7 @@
 from mcp.types.jsonrpc import (
     HEADER_MISMATCH,
     INVALID_PARAMS,
+    INVALID_REQUEST,
     METHOD_NOT_FOUND,
     MISSING_REQUIRED_CLIENT_CAPABILITY,
     PARSE_ERROR,
@@ -42,6 +43,7 @@
 ERROR_CODE_HTTP_STATUS: Final[Mapping[int, int]] = MappingProxyType(
     {
         PARSE_ERROR: 400,
+        INVALID_REQUEST: 400,
         INVALID_PARAMS: 400,
         HEADER_MISMATCH: 400,
         MISSING_REQUIRED_CLIENT_CAPABILITY: 400,

tests/server/test_streamable_http_manager.py

@@ -103,11 +103,12 @@ async def running_manager():
 
 @pytest.mark.anyio
 async def test_stateful_session_cleanup_on_graceful_exit(running_manager: tuple[StreamableHTTPSessionManager, Server]):
-    manager, app = running_manager
+    manager, _app = running_manager
 
-    mock_mcp_run = AsyncMock(return_value=None)
-    # This will be called by StreamableHTTPSessionManager's run_server -> self.app.run
-    app.run = mock_mcp_run
+    # The manager's `run_server` task drives `serve_loop` directly (the manager
+    # owns lifespan); patch that seam so the loop returns immediately and we
+    # can observe the cleanup that follows.
+    mock_serve = AsyncMock(return_value=None)
 
     sent_messages: list[Message] = []
 
@@ -125,7 +126,8 @@ async def mock_receive():  # pragma: no cover
         return {"type": "http.request", "body": b"", "more_body": False}
 
     # Trigger session creation
-    await manager.handle_request(scope, mock_receive, mock_send)
+    with patch("mcp.server.streamable_http_manager.serve_loop", mock_serve):
+        await manager.handle_request(scope, mock_receive, mock_send)
 
     # Extract session ID from response headers
     session_id = None
@@ -140,10 +142,9 @@ async def mock_receive():  # pragma: no cover
 
     assert session_id is not None, "Session ID not found in response headers"
 
-    # Ensure MCPServer.run was called
-    mock_mcp_run.assert_called_once()
+    mock_serve.assert_called_once()
 
-    # At this point, mock_mcp_run has completed, and the finally block in
+    # At this point, mock_serve has completed, and the finally block in
     # StreamableHTTPSessionManager's run_server should have executed.
 
     # To ensure the task spawned by handle_request finishes and cleanup occurs:
@@ -158,10 +159,9 @@ async def mock_receive():  # pragma: no cover
 
 @pytest.mark.anyio
 async def test_stateful_session_cleanup_on_exception(running_manager: tuple[StreamableHTTPSessionManager, Server]):
-    manager, app = running_manager
+    manager, _app = running_manager
 
-    mock_mcp_run = AsyncMock(side_effect=TestException("Simulated crash"))
-    app.run = mock_mcp_run
+    mock_serve = AsyncMock(side_effect=TestException("Simulated crash"))
 
     sent_messages: list[Message] = []
 
@@ -184,7 +184,8 @@ async def mock_receive():  # pragma: no cover
         return {"type": "http.request", "body": b"", "more_body": False}
 
     # Trigger session creation
-    await manager.handle_request(scope, mock_receive, mock_send)
+    with patch("mcp.server.streamable_http_manager.serve_loop", mock_serve):
+        await manager.handle_request(scope, mock_receive, mock_send)
 
     session_id = None
     for msg in sent_messages:  # pragma: no branch
@@ -198,7 +199,7 @@ async def mock_receive():  # pragma: no cover
 
     assert session_id is not None, "Session ID not found in response headers"
 
-    mock_mcp_run.assert_called_once()
+    mock_serve.assert_called_once()
 
     # Give other tasks a chance to run to ensure the finally block executes
     await anyio.sleep(0.01)

tests/server/test_streamable_http_modern.py

@@ -24,7 +24,7 @@
 from mcp.types import (
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
-    METHOD_NOT_FOUND,
+    INVALID_REQUEST,
     PARSE_ERROR,
     PROTOCOL_VERSION_META_KEY,
     ListToolsResult,
@@ -60,17 +60,29 @@ async def app(scope: Scope, receive: Receive, send: Send) -> None:
     )
 
 
-async def test_handle_modern_request_rejects_non_post_with_method_not_found() -> None:
-    """A GET on the modern entry is rejected with HTTP 404 and a ``METHOD_NOT_FOUND`` JSON-RPC error.
-
-    SDK-defined: the 404 status comes from the SDK's error-code→HTTP-status table; spec-mandated: the
-    body carries the JSON-RPC ``METHOD_NOT_FOUND`` code.
-    """
+async def test_handle_modern_request_rejects_non_post_with_http_405_and_allow_header() -> None:
+    """SDK-defined: a GET on the modern entry is an HTTP-verb mismatch — 405 Method Not
+    Allowed with ``Allow: POST`` per RFC 9110. This is HTTP-layer (before JSON-RPC parsing)
+    so there is no JSON-RPC body."""
     async with _asgi_client(Server("test")) as http:
         response = await http.get("/mcp")
-    assert response.status_code == 404
+    assert response.status_code == 405
     assert response.headers["allow"] == "POST"
-    assert response.json()["error"]["code"] == METHOD_NOT_FOUND
+    assert response.content == b""
+
+
+async def test_handle_modern_request_rejects_a_notification_body_with_invalid_request() -> None:
+    """SDK-defined: well-formed JSON that isn't a single JSON-RPC request object (e.g. a
+    notification, which lacks ``id``) is ``INVALID_REQUEST`` — distinct from ``PARSE_ERROR``,
+    which is for malformed JSON."""
+    async with _asgi_client(Server("test")) as http:
+        response = await http.post(
+            "/mcp",
+            content=b'{"jsonrpc":"2.0","method":"notifications/cancelled","params":{"requestId":1}}',
+            headers={"content-type": "application/json"},
+        )
+    assert response.status_code == 400
+    assert response.json()["error"]["code"] == INVALID_REQUEST
 
 
 async def test_handle_modern_request_rejects_malformed_body_with_parse_error() -> None: