Rebuild the streamable HTTP server transport around per-request dispatch

docs/migration.md

@@ -721,6 +721,59 @@ When serving streamable HTTP (stateful or `stateless_http=True`), the server's `
 
 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` (a public way to reach it from high-level `@mcp.tool()` handlers is planned).
 
+### Streamable HTTP: `StreamableHTTPServerTransport` is driven per request, not per stream
+
+`StreamableHTTPServerTransport` no longer exposes a `connect()` context manager yielding a
+`(read_stream, write_stream)` pair for you to run a server loop over. Each HTTP request is now
+dispatched to the server's handlers directly: a request's outbound messages ride that request's
+own response stream (backed by the optional `EventStore` for `Last-Event-ID` resumability), the
+client's POSTed answers to server-initiated requests are correlated back by request id, and the
+standalone GET stream is a further per-connection channel. The transport is the per-session
+core; `StreamableHTTPSessionManager` binds one to the `Server` for each session (via the new
+keyword-only `app` / `lifespan_state` constructor arguments) and routes requests to it.
+
+Nothing changes if you serve through `streamable_http_app()` / `run(transport="streamable-http")`
+or mount `StreamableHTTPSessionManager` — the wire behaviour (session ids, GET stream, event
+store, `ctx.close_sse_stream()`, `related_request_id` routing) is unchanged. Only code that
+constructed a transport and consumed `transport.connect()` by hand needs to move to the session
+manager:
+
+```python
+# Before (v1)
+transport = StreamableHTTPServerTransport(mcp_session_id=session_id, ...)
+async with transport.connect() as (read_stream, write_stream):
+    await server.run(read_stream, write_stream, server.create_initialization_options())
+
+# After (v2): serve the app the SDK builds ...
+app = server.streamable_http_app(event_store=..., json_response=...)
+```
+
+... or, when composing your own Starlette/FastAPI app, mount a `StreamableHTTPSessionManager` and
+enter `session_manager.run()` in the lifespan — see [Mounting the ASGI app](run/asgi.md) for the
+full wiring.
+
+Behaviour clarified in the same change:
+
+- In JSON-response mode a *request-scoped* server-to-client request (`ctx.elicit()`, or any
+  `ctx.session` call carrying `related_request_id`) now raises `NoBackChannelError` — the POST's
+  single JSON body has no stream to carry the nested request, and previously the call would hang
+  waiting for an answer that could never be delivered. Connection-scoped sends (calls without
+  `related_request_id`) are unchanged and still ride the standalone GET stream.
+- A GET carrying `Last-Event-ID` on a server without an `EventStore` opens the standalone stream
+  as a plain GET would, since there is nothing to replay.
+- Two concurrent POSTs that share a JSON-RPC request id each keep their own response stream; the
+  second no longer silently takes over the first's queue.
+- Stream ids handed to your `EventStore` are minted by the transport in its own session-scoped
+  namespace (previously the raw `str(request_id)` and a single global GET-stream key), so two
+  sessions sharing one store no longer collide, and a `Last-Event-ID` replay only releases frames
+  of the requesting session's own streams. Treat the ids as opaque.
+- A failing `EventStore.store_event` degrades resumability for that message rather than taking
+  the stream down: the message is still delivered live (with no event id to resume from) and the
+  store's exception is logged, never sent to the client.
+- A server-to-client request that can reach no client at all (no attached stream and nothing
+  storing it, or a request-scoped one in JSON-response mode) fails the calling handler with
+  `CONNECTION_CLOSED` instead of parking it for an answer that cannot arrive.
+
 ### `MCPServer.get_context()` removed
 
 `MCPServer.get_context()` has been removed. Context is now injected by the framework and passed explicitly — there is no ambient ContextVar to read from.

src/mcp/server/lowlevel/server.py

@@ -704,8 +704,9 @@ async def run(
         Thin wrapper over `serve_dual_era_loop`: enters the server lifespan,
         then drives the loop, serving the legacy handshake era and the modern
         per-request-envelope era (the client's first request decides which).
-        Transports with their own lifespan owner (the streamable-HTTP manager)
-        call `serve_loop` directly instead.
+        Transports with their own lifespan owner call `serve_loop` directly
+        instead (or, without a stream pair - the streamable-HTTP manager -
+        dispatch each request themselves).
         """
         async with self.lifespan(self) as lifespan_context:
             await serve_dual_era_loop(

src/mcp/server/runner.py

@@ -230,7 +230,12 @@ async def _inner(ctx: ServerRequestContext[LifespanT, Any]) -> HandlerResult:
         result = _dump_result(await call(ctx))
         if method == "initialize":
             # Commit only on chain success, so a middleware veto leaves no state.
-            # Race-free: the read loop is parked until this call returns.
+            # Race-free for the session's first handshake: the transport runs no
+            # other request until it returns (a stream driver's read loop is
+            # parked here; streamable HTTP holds later requests behind the
+            # in-progress initialize, and the session id only ships with its
+            # response). A repeated initialize on an established session (a
+            # recorded divergence) recommits alongside whatever is running.
             # TODO: this re-reads the wire `params`, so a middleware that rewrote
             # `ctx.params` (or `ctx.method`, or short-circuited without `call_next`)
             # can leave `connection.protocol_version` out of step with the
@@ -477,9 +482,9 @@ async def serve_loop(
     """Drive ``server`` in handshake-only loop mode over a stream pair until the channel closes.
 
     Builds the loop-mode `JSONRPCDispatcher` + `Connection` and hands them to
-    `serve_connection`. The streamable-HTTP manager (which owns its lifespan
-    and serves the modern era on the single-exchange entry instead) calls
-    this; `Server.run` drives `serve_dual_era_loop`, which extends the same
+    `serve_connection`. For a transport that supplies a duplex message stream
+    pair but owns its own lifespan (so `Server.run`'s lifespan entry is not
+    wanted); `Server.run` drives `serve_dual_era_loop`, which extends the same
     dispatcher recipe (notably the `inline_methods={"initialize"}` rule) with
     era routing.
     """