[v2] ClientSession runs on JSONRPCDispatcher; BaseSession removed (#2838)

docs/migration.md

@@ -634,11 +634,9 @@ server = Server("my-server", on_call_tool=handle_call_tool)
 
 The `mcp.shared.context` module has been removed. `RequestContext` is now split into `ClientRequestContext` (in `mcp.client.context`) and `ServerRequestContext` (in `mcp.server.context`).
 
-The `RequestContext` class has been split to separate shared fields from server-specific fields. The shared `RequestContext` now only takes 1 type parameter (the session type) instead of 3.
-
 **`RequestContext` changes:**
 
-- Type parameters reduced from `RequestContext[SessionT, LifespanContextT, RequestT]` to `RequestContext[SessionT]`
+- The `RequestContext[SessionT, LifespanContextT, RequestT]` generic no longer exists; use `ClientRequestContext` or `ServerRequestContext[LifespanContextT, RequestT]`
 - Server-specific fields (`lifespan_context`, `request`, `close_sse_stream`, `close_standalone_sse_stream`) moved to new `ServerRequestContext` class in `mcp.server.context`
 
 **Before (v1):**
@@ -1122,9 +1120,9 @@ async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestPar
     )
 ```
 
-### `RequestContext`: request-specific fields are now optional
+### `ServerRequestContext`: request-specific fields are now optional
 
-The `RequestContext` class now uses optional fields for request-specific data (`request_id`, `meta`, etc.) so it can be used for both request and notification handlers. In notification handlers, these fields are `None`.
+`ServerRequestContext` now uses optional fields for request-specific data (`request_id`, `meta`, etc.) so it can be used for both request and notification handlers. In notification handlers, these fields are `None`.
 
 ```python
 from mcp.server import ServerRequestContext
@@ -1164,7 +1162,22 @@ In practice, replace direct `ServerSession` use with `Server.run(read_stream, wr
 
 `BaseSession._in_flight` and the `RequestResponder` members that supported it (`cancel()`, the `cancelled` and `in_flight` properties, the `on_complete` constructor argument, and the internal `CancelScope`) have been removed. These existed to let `ServerSession` cancel a handler when a `CancelledNotification` arrived; `ServerSession` no longer drives a receive loop, so they were dead code. Inbound-cancellation handling for the server now lives in `JSONRPCDispatcher`.
 
-`BaseSession` is still used by `ClientSession`, which never relied on these members. `RequestResponder.respond()` is unchanged.
+`BaseSession` itself has since been removed entirely; see the next section.
+
+### `ClientSession` now runs on `JSONRPCDispatcher`; `BaseSession` removed
+
+`ClientSession`'s public surface is unchanged — same constructor, typed methods, manual `initialize()`, and async context-manager lifecycle — but `BaseSession`, the v1 receive loop underneath it, is removed with no shim. The engine now lives in `JSONRPCDispatcher` (`mcp.shared.jsonrpc_dispatcher`). To customize client behavior, use the `ClientSession` constructor callbacks, or pass a pre-built dispatcher via the new keyword-only `dispatcher=` constructor argument (e.g. a `DirectDispatcher` for in-process embedding).
+
+Behavior changes:
+
+- **Callbacks and notifications now run concurrently.** In v1 the receive loop processed one inbound message at a time, so callbacks ran inline and in order. Now each delivery starts in arrival order but runs as its own task. Server-initiated request callbacks (`sampling`, `elicitation`, `roots`) no longer block other traffic, may themselves send requests without deadlocking, and are interrupted if the server sends `notifications/cancelled` (the request is then answered with an error). Notification callbacks (`logging_callback`, `progress_callback`, `message_handler`) may interleave, and a `progress_callback` may run after the request it reports on has returned; there is no built-in bound on concurrent deliveries. Transport-level errors reach `message_handler` the same way, and a `message_handler` that raises is logged rather than fatal to the session. Callbacks that need strict sequencing must coordinate themselves.
+- **Timeouts**: a timed-out or abandoned request is now followed by `notifications/cancelled`, so the server stops the handler instead of leaving it running.
+- **A raising request callback** is answered with `code=0` and the exception text; v1 flattened every callback exception to `INVALID_PARAMS`. For a specific error response, return `ErrorData` (unchanged) or raise `MCPError`. One carve-out: pydantic's `ValidationError` is still answered with `INVALID_PARAMS`, as in v1.
+- **`send_request` before entering the context manager** raises `RuntimeError` immediately; v1 wrote to the transport and hung until the timeout. After the connection has closed it raises `MCPError` (`CONNECTION_CLOSED`) instead. `send_notification` before entry still works.
+- **`send_notification` no longer takes `related_request_id`, and `send_request` no longer accepts `ServerMessageMetadata`.** No client transport ever serialized these hints; progress and response correlation via `progressToken` and the request id is unaffected.
+- **Client callbacks now receive `mcp.client.ClientRequestContext`** (its `request_id` is always populated); the private `mcp.shared._context.RequestContext` generic is deleted. Annotations spelled `RequestContext[ClientSession]` become `ClientRequestContext`.
+
+`mcp.shared.session` is now a compatibility module: `ProgressFnT` is re-exported (its home is `mcp.shared.dispatcher`), and `RequestResponder` remains as a typing-only stub so `MessageHandlerFnT` annotations keep importing. `RequestResponder.respond()` no longer exists.
 
 ### Experimental Tasks support removed
 

src/mcp/client/client.py

@@ -12,7 +12,7 @@ from mcp.client.session import ClientSession, ElicitationFnT, ListRootsFnT, Logg
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
-from mcp.shared.session import ProgressFnT
+from mcp.shared.dispatcher import ProgressFnT
 from mcp.types import (
     CallToolResult,
     CompleteResult,

src/mcp/client/context.py

@@ -1,16 +1,5 @@
 """Request context for MCP client handlers."""
 
-from mcp.client.session import ClientSession
+from mcp.client.session import ClientRequestContext
-from mcp.shared._context import RequestContext
 
-ClientRequestContext = RequestContext[ClientSession]
+__all__ = ["ClientRequestContext"]
-"""Context for handling incoming requests in a client session.
-
-This context is passed to client-side callbacks (sampling, elicitation, list_roots) when the server sends requests
-to the client.
-
-Attributes:
-    request_id: The unique identifier for this request.
-    meta: Optional metadata associated with the request.
-    session: The client session handling this request.
-"""

src/mcp/client/session.py

@@ -1,28 +1,49 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import TracebackType
 from typing import Any, Protocol, cast, get_args
 
+import anyio
+import anyio.abc
 import anyio.lowlevel
-from pydantic import BaseModel, TypeAdapter
+from pydantic import BaseModel, TypeAdapter, ValidationError
+from typing_extensions import Self, TypeVar
 
 from mcp import types
 from mcp.client._transport import ReadStream, WriteStream
-from mcp.shared._context import RequestContext
+from mcp.shared._compat import resync_tracer
-from mcp.shared.message import SessionMessage
+from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, ProgressFnT
-from mcp.shared.session import BaseSession, ProgressFnT, RequestResponder
+from mcp.shared.exceptions import MCPError
+from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
+from mcp.shared.message import ClientMessageMetadata, SessionMessage
+from mcp.shared.session import RequestResponder
+from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
-from mcp.types._types import RequestParamsMeta
+from mcp.types import RequestId, RequestParamsMeta
 
 DEFAULT_CLIENT_INFO = types.Implementation(name="mcp", version="0.1.0")
 
 logger = logging.getLogger("client")
 
+ReceiveResultT = TypeVar("ReceiveResultT", bound=BaseModel)
+
+
+@dataclass(kw_only=True)
+class ClientRequestContext:
+    """Context for a server-initiated request, passed to the sampling/elicitation/list-roots callbacks."""
+
+    session: ClientSession
+    request_id: RequestId
+    meta: RequestParamsMeta | None = None
+
 
 class SamplingFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData: ...  # pragma: no branch
 
@@ -30,14 +51,14 @@ class SamplingFnT(Protocol):
 class ElicitationFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.ElicitRequestParams,
     ) -> types.ElicitResult | types.ErrorData: ...  # pragma: no branch
 
 
 class ListRootsFnT(Protocol):
     async def __call__(
-        self, context: RequestContext[ClientSession]
+        self, context: ClientRequestContext
     ) -> types.ListRootsResult | types.ErrorData: ...  # pragma: no branch
 
 
@@ -59,7 +80,7 @@ async def _default_message_handler(
 
 
 async def _default_sampling_callback(
-    context: RequestContext[ClientSession],
+    context: ClientRequestContext,
     params: types.CreateMessageRequestParams,
 ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData:
     return types.ErrorData(
@@ -69,7 +90,7 @@ async def _default_sampling_callback(
 
 
 async def _default_elicitation_callback(
-    context: RequestContext[ClientSession],
+    context: ClientRequestContext,
     params: types.ElicitRequestParams,
 ) -> types.ElicitResult | types.ErrorData:
     return types.ErrorData(
@@ -79,7 +100,7 @@ async def _default_elicitation_callback(
 
 
 async def _default_list_roots_callback(
-    context: RequestContext[ClientSession],
+    context: ClientRequestContext,
 ) -> types.ListRootsResult | types.ErrorData:
     return types.ErrorData(
         code=types.INVALID_REQUEST,
@@ -104,19 +125,21 @@ spec methods this SDK deliberately doesn't model, like `tasks/*` — are
 answered with METHOD_NOT_FOUND instead of failing union validation."""
 
 
-class ClientSession(
+class ClientSession:
-    BaseSession[
+    """Client half of an MCP connection, running on a `Dispatcher`.
-        types.ClientRequest,
+
-        types.ClientNotification,
+    Construct it over a transport's stream pair (or pass a pre-built
-        types.ClientResult,
+    `dispatcher=`), enter as an async context manager, then call
-        types.ServerRequest,
+    `initialize()`. The dispatcher owns the receive loop and request
-        types.ServerNotification,
+    correlation; this class owns the typed MCP layer and the constructor
-    ]
+    callbacks. Transport `Exception` items reach `message_handler` only when
-):
+    the session builds its own dispatcher from a stream pair.
+    """
+
     def __init__(
         self,
-        read_stream: ReadStream[SessionMessage | Exception],
+        read_stream: ReadStream[SessionMessage | Exception] | None = None,
-        write_stream: WriteStream[SessionMessage],
+        write_stream: WriteStream[SessionMessage] | None = None,
         read_timeout_seconds: float | None = None,
         sampling_callback: SamplingFnT | None = None,
         elicitation_callback: ElicitationFnT | None = None,
@@ -126,8 +149,9 @@ class ClientSession(
         client_info: types.Implementation | None = None,
         *,
         sampling_capabilities: types.SamplingCapability | None = None,
+        dispatcher: Dispatcher[Any] | None = None,
     ) -> None:
-        super().__init__(read_stream, write_stream, read_timeout_seconds=read_timeout_seconds)
+        self._session_read_timeout_seconds = read_timeout_seconds
         self._client_info = client_info or DEFAULT_CLIENT_INFO
         self._sampling_callback = sampling_callback or _default_sampling_callback
         self._sampling_capabilities = sampling_capabilities
@@ -137,18 +161,99 @@ class ClientSession(
         self._message_handler = message_handler or _default_message_handler
         self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
         self._initialize_result: types.InitializeResult | None = None
+        self._task_group: anyio.abc.TaskGroup | None = None
+        if dispatcher is not None:
+            if read_stream is not None or write_stream is not None:
+                raise ValueError("pass read_stream/write_stream or dispatcher, not both")
+            self._dispatcher: Dispatcher[Any] = dispatcher
+        else:
+            if read_stream is None or write_stream is None:
+                raise ValueError("read_stream and write_stream are required when no dispatcher is given")
+            # Built eagerly so notifications can be sent before entering the context manager.
+            self._dispatcher = JSONRPCDispatcher(
+                read_stream, write_stream, on_stream_exception=self._on_stream_exception
+            )
 
-    @property
+    async def __aenter__(self) -> Self:
-    def _receive_request_adapter(self) -> TypeAdapter[types.ServerRequest]:
+        self._task_group = anyio.create_task_group()
-        return types.server_request_adapter
+        await self._task_group.__aenter__()
+        try:
+            await self._task_group.start(self._dispatcher.run, self._on_request, self._on_notify)
+        except BaseException:
+            # Unwind the entered task group before propagating: a cancellation
+            # landing here (e.g. `move_on_after` around connect) would abandon
+            # it and anyio would later raise "exited non-innermost cancel scope".
+            task_group = self._task_group
+            self._task_group = None
+            task_group.cancel_scope.cancel()
+            # Shield the group's own scope (a new one would break LIFO exit)
+            # so a pending outer cancellation cannot re-fire inside __aexit__.
+            task_group.cancel_scope.shield = True
+            await task_group.__aexit__(None, None, None)
+            raise
+        return self
 
-    @property
+    async def __aexit__(
-    def _receive_request_methods(self) -> frozenset[str]:
+        self,
-        return _SERVER_REQUEST_METHODS
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None:
+        # Exit must not block: cancel the dispatcher and in-flight callbacks.
+        assert self._task_group is not None
+        self._task_group.cancel_scope.cancel()
+        result = await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
+        await resync_tracer()
+        return result
 
-    @property
+    async def send_request(
-    def _receive_notification_adapter(self) -> TypeAdapter[types.ServerNotification]:
+        self,
-        return types.server_notification_adapter
+        request: types.ClientRequest,
+        result_type: type[ReceiveResultT],
+        request_read_timeout_seconds: float | None = None,
+        metadata: ClientMessageMetadata | None = None,
+        progress_callback: ProgressFnT | None = None,
+    ) -> ReceiveResultT:
+        """Send a request and wait for its typed result.
+
+        Args:
+            metadata: Streamable HTTP resumption hints.
+
+        Raises:
+            MCPError: Error response, read timeout, or connection closed.
+            RuntimeError: Called before entering the context manager.
+        """
+        data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
+        method: str = data["method"]
+        opts: CallOptions = {}
+        timeout = (
+            request_read_timeout_seconds
+            if request_read_timeout_seconds is not None
+            else self._session_read_timeout_seconds
+        )
+        if timeout is not None:
+            opts["timeout"] = timeout
+        if progress_callback is not None:
+            opts["on_progress"] = progress_callback
+        if metadata is not None:
+            if metadata.resumption_token is not None:
+                opts["resumption_token"] = metadata.resumption_token
+            if metadata.on_resumption_token_update is not None:
+                opts["on_resumption_token"] = metadata.on_resumption_token_update
+        if method == "initialize":
+            # The spec forbids cancelling initialize.
+            opts["cancel_on_abandon"] = False
+        raw = await self._dispatcher.send_raw_request(method, data.get("params"), opts)
+        return result_type.model_validate(raw, by_name=False)
+
+    async def send_notification(self, notification: types.ClientNotification) -> None:
+        """Send a one-way notification. Usable before entering the context manager.
+
+        Fire-and-forget: after the connection has closed, the notification is
+        dropped with a debug log instead of raising.
+        """
+        data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
+        await self._dispatcher.notify(data["method"], data.get("params"))
 
     async def initialize(self) -> types.InitializeResult:
         sampling = (
@@ -397,49 +502,75 @@ class ClientSession(
         """Send a roots/list_changed notification."""
         await self.send_notification(types.RootsListChangedNotification())
 
-    async def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]) -> None:
+    async def _on_request(
-        ctx = RequestContext[ClientSession](request_id=responder.request_id, meta=responder.request_meta, session=self)
+        self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
+    ) -> dict[str, Any]:
+        """Answer a server-initiated request via the registered callbacks."""
+        if method not in _SERVER_REQUEST_METHODS:
+            raise MCPError(code=types.METHOD_NOT_FOUND, message="Method not found", data=method)
+        payload: dict[str, Any] = {"method": method}
+        if params is not None:
+            payload["params"] = dict(params)
+        request = types.server_request_adapter.validate_python(payload, by_name=False)
 
-        match responder.request:
+        response: types.ClientResult | types.ErrorData
-            case types.CreateMessageRequest(params=params):
+        if isinstance(request, types.PingRequest):
-                with responder:
+            # Answered without a context: ping has no callback that would need one.
-                    response = await self._sampling_callback(ctx, params)
+            response = types.EmptyResult()
-                    client_response = ClientResponse.validate_python(response)
+        else:
-                    await responder.respond(client_response)
+            assert dctx.request_id is not None  # the callback-driving dispatchers always assign ids
-
+            ctx = ClientRequestContext(
-            case types.ElicitRequest(params=params):
+                session=self, request_id=dctx.request_id, meta=request.params.meta if request.params else None
-                with responder:
+            )
-                    response = await self._elicitation_callback(ctx, params)
+            match request:
-                    client_response = ClientResponse.validate_python(response)
+                case types.CreateMessageRequest(params=sampling_params):
-                    await responder.respond(client_response)
+                    response = await self._sampling_callback(ctx, sampling_params)
-
+                case types.ElicitRequest(params=elicit_params):
-            case types.ListRootsRequest():
+                    response = await self._elicitation_callback(ctx, elicit_params)
-                with responder:
+                case types.ListRootsRequest():  # pragma: no branch
                     response = await self._list_roots_callback(ctx)
-                    client_response = ClientResponse.validate_python(response)
+        client_response = ClientResponse.validate_python(response)
-                    await responder.respond(client_response)
+        if isinstance(client_response, types.ErrorData):
+            raise MCPError.from_error_data(client_response)
+        return client_response.model_dump(by_alias=True, mode="json", exclude_none=True)
 
-            case types.PingRequest():  # pragma: no branch
+    async def _on_notify(
-                with responder:
+        self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
-                    await responder.respond(types.EmptyResult())
-
-    async def _handle_incoming(
-        self,
-        req: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
     ) -> None:
-        """Handle incoming messages by forwarding to the message handler."""
+        """Route a server notification: validate, run the typed callback, tee to message_handler."""
-        await self._message_handler(req)
+        payload: dict[str, Any] = {"method": method}
+        if params is not None:
+            payload["params"] = dict(params)
+        try:
+            notification = types.server_notification_adapter.validate_python(payload, by_name=False)
+        except ValidationError:
+            logger.warning("Failed to validate notification: %s", payload, exc_info=True)
+            return
+        if isinstance(notification, types.CancelledNotification):
+            # The dispatcher already applied the cancellation; not surfaced to message_handler.
+            return
+        try:
+            if isinstance(notification, types.LoggingMessageNotification):
+                await self._logging_callback(notification.params)
+            await self._message_handler(notification)
+        except Exception:
+            # Contain here, not in the dispatcher: DirectDispatcher awaits this
+            # handler inline in the peer's notify() call, so a raising callback
+            # would otherwise fail the peer's send. A raising logging_callback
+            # skips the message_handler tee for that notification (v1 parity).
+            logger.exception("notification callback for %r raised", method)
 
-    async def _received_notification(self, notification: types.ServerNotification) -> None:
+    async def _on_stream_exception(self, exc: Exception) -> None:
-        """Handle notifications from the server."""
+        """Deliver a transport-level fault to message_handler via a spawned task.
-        # Process specific notification types
+
-        match notification:
+        Running the handler inline would park the dispatcher's read loop and
-            case types.LoggingMessageNotification(params=params):
+        deadlock handlers that await session I/O.
-                await self._logging_callback(params)
+        """
-            case types.ElicitCompleteNotification(params=params):
+        assert self._task_group is not None
-                # Handle elicitation completion notification
+        self._task_group.start_soon(self._deliver_stream_exception, exc)
-                # Clients MAY use this to retry requests or update UI
+
-                # The notification contains the elicitationId of the completed elicitation
+    async def _deliver_stream_exception(self, exc: Exception) -> None:
-                pass
+        try:
-            case _:
+            await self._message_handler(exc)
-                pass
+        except Exception:
+            logger.exception("message_handler raised on transport exception")

src/mcp/server/session.py

@@ -84,7 +84,7 @@ class ServerSession:
         """
         data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
         opts: CallOptions = {}
-        if request_read_timeout_seconds:
+        if request_read_timeout_seconds is not None:
             opts["timeout"] = request_read_timeout_seconds
         if progress_callback is not None:
             opts["on_progress"] = progress_callback

src/mcp/server/streamable_http.py

@@ -717,6 +717,9 @@ class StreamableHTTPServerTransport:
                         # Send the message via SSE
                         event_data = self._create_event_data(event_message)
                         await sse_stream_writer.send(event_data)
+            except anyio.ClosedResourceError:
+                # Session teardown can close the stream while the writer is between dequeues.
+                pass
             except Exception:
                 logger.exception("Error in standalone SSE writer")  # pragma: no cover
             finally:

src/mcp/shared/_context.py

@@ -1,24 +0,0 @@
-"""Request context for MCP handlers."""
-
-from dataclasses import dataclass
-from typing import Any, Generic
-
-from typing_extensions import TypeVar
-
-from mcp.shared.session import BaseSession
-from mcp.types import RequestId, RequestParamsMeta
-
-SessionT = TypeVar("SessionT", bound=BaseSession[Any, Any, Any, Any, Any])
-
-
-@dataclass(kw_only=True)
-class RequestContext(Generic[SessionT]):
-    """Common context for handling incoming requests.
-
-    For request handlers, request_id is always populated.
-    For notification handlers, request_id is None.
-    """
-
-    session: SessionT
-    request_id: RequestId | None = None
-    meta: RequestParamsMeta | None = None

src/mcp/shared/direct_dispatcher.py

@@ -15,6 +15,7 @@ to the caller - there is no exception-to-`ErrorData` boundary here.
 
 from __future__ import annotations
 
+import logging
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass, field
 from typing import Any
@@ -27,7 +28,9 @@ from mcp.shared.dispatcher import CallOptions, OnNotify, OnRequest, ProgressFnT
 from mcp.shared.exceptions import MCPError, NoBackChannelError
 from mcp.shared.message import MessageMetadata
 from mcp.shared.transport_context import TransportContext
-from mcp.types import INTERNAL_ERROR, INVALID_PARAMS, REQUEST_TIMEOUT, RequestId
+from mcp.types import CONNECTION_CLOSED, INTERNAL_ERROR, INVALID_PARAMS, REQUEST_TIMEOUT, RequestId
+
+logger = logging.getLogger(__name__)
 
 __all__ = ["DirectDispatcher", "create_direct_dispatcher_pair"]
 
@@ -50,7 +53,7 @@ class _DirectDispatchContext:
     _back_request: _Request
     _back_notify: _Notify
     request_id: RequestId | None = None
-    """Always `None`: direct dispatch has no wire-level request id."""
+    """A dispatcher-synthesized id for requests; `None` for notifications."""
     message_metadata: MessageMetadata = None  # TODO(maxisbey): remove for Context rework
     """Always `None`: in-memory dispatch attaches no transport metadata."""
     _on_progress: ProgressFnT | None = None
@@ -84,6 +87,13 @@ class DirectDispatcher:
     Two instances are wired together with `create_direct_dispatcher_pair`; each
     holds a reference to the other. `send_raw_request` on one awaits the peer's
     `on_request`. `run` parks until `close` is called.
+
+    Lifecycle mirrors `JSONRPCDispatcher`: `send_raw_request` requires `run()`
+    to have started, and once a side has closed - via `close()` or `run()`
+    ending - `send_raw_request` raises `MCPError` (`CONNECTION_CLOSED`) and
+    inbound requests fail the peer's call the same way instead of invoking the
+    handler. Notifications are fire-and-forget in both directions: after close
+    they are silently dropped.
     """
 
     def __init__(self, transport_ctx: TransportContext):
@@ -91,8 +101,11 @@ class DirectDispatcher:
         self._peer: DirectDispatcher | None = None
         self._on_request: OnRequest | None = None
         self._on_notify: OnNotify | None = None
+        self._next_id = 0
         self._ready = anyio.Event()
-        self._closed = anyio.Event()
+        self._close_event = anyio.Event()
+        self._running = False
+        self._closed = False
 
     def connect_to(self, peer: DirectDispatcher) -> None:
         self._peer = peer
@@ -103,13 +116,35 @@ class DirectDispatcher:
         params: Mapping[str, Any] | None,
         opts: CallOptions | None = None,
     ) -> dict[str, Any]:
+        """Send a request by invoking the peer's `on_request` directly.
+
+        Raises:
+            MCPError: The peer's handler raised; `REQUEST_TIMEOUT` if
+                `opts["timeout"]` elapsed; `CONNECTION_CLOSED` if either
+                side has closed.
+            RuntimeError: Called before `run()`.
+        """
         if self._peer is None:
             raise RuntimeError("DirectDispatcher has no peer; use create_direct_dispatcher_pair()")
+        # Post-close sends get the same CONNECTION_CLOSED contract as JSONRPCDispatcher.
+        if self._closed:
+            raise MCPError(code=CONNECTION_CLOSED, message="Connection closed")
+        if not self._running:
+            raise RuntimeError("DirectDispatcher.send_raw_request called before run()")
         return await self._peer._dispatch_request(method, params, opts)
 
     async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+        """Send a notification by invoking the peer's `on_notify` directly.
+
+        Fire-and-forget: usable before `run()` (delivery waits for the peer to
+        start), and after close it is silently dropped, matching
+        `JSONRPCDispatcher.notify`.
+        """
         if self._peer is None:
             raise RuntimeError("DirectDispatcher has no peer; use create_direct_dispatcher_pair()")
+        if self._closed:
+            logger.debug("dropped notification %r on closed DirectDispatcher", method)
+            return
         await self._peer._dispatch_notify(method, params)
 
     async def run(
@@ -119,37 +154,77 @@ class DirectDispatcher:
         *,
         task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
     ) -> None:
-        self._on_request = on_request
+        """Mark this side ready and park until `close()` is called.
-        self._on_notify = on_notify
+
-        self._ready.set()
+        Single-shot, like `JSONRPCDispatcher.run`: once it returns the
-        task_status.started()
+        dispatcher stays closed and cannot be restarted.
-        await self._closed.wait()
+        """
+        try:
+            self._on_request = on_request
+            self._on_notify = on_notify
+            self._running = True
+            self._ready.set()
+            task_status.started()
+            await self._close_event.wait()
+        finally:
+            self._running = False
+            self._closed = True
+            # run() may end via cancellation without close() ever being
+            # called; setting the event wakes `_wait_ready` waiters so they
+            # observe the closed state instead of parking forever.
+            self._close_event.set()
 
     def close(self) -> None:
-        self._closed.set()
+        self._closed = True
+        self._close_event.set()
 
-    def _make_context(self, on_progress: ProgressFnT | None = None) -> _DirectDispatchContext:
+    def _make_context(
+        self, on_progress: ProgressFnT | None = None, request_id: RequestId | None = None
+    ) -> _DirectDispatchContext:
         assert self._peer is not None
         peer = self._peer
         return _DirectDispatchContext(
             transport=self._transport_ctx,
             _back_request=lambda m, p, o: peer._dispatch_request(m, p, o),
             _back_notify=lambda m, p: peer._dispatch_notify(m, p),
+            request_id=request_id,
             _on_progress=on_progress,
         )
 
+    async def _wait_ready(self) -> None:
+        """Park until `run()` has started, waking early if this side closes.
+
+        Raises:
+            MCPError: `CONNECTION_CLOSED` if this side has closed.
+        """
+        if not self._ready.is_set() and not self._close_event.is_set():
+            async with anyio.create_task_group() as tg:
+
+                async def wake_on(event: anyio.Event) -> None:
+                    await event.wait()
+                    tg.cancel_scope.cancel()
+
+                tg.start_soon(wake_on, self._ready)
+                tg.start_soon(wake_on, self._close_event)
+        if self._closed:
+            raise MCPError(code=CONNECTION_CLOSED, message="Connection closed")
+
     async def _dispatch_request(
         self,
         method: str,
         params: Mapping[str, Any] | None,
         opts: CallOptions | None,
     ) -> dict[str, Any]:
-        await self._ready.wait()
-        assert self._on_request is not None
         opts = opts or {}
-        dctx = self._make_context(on_progress=opts.get("on_progress"))
         try:
             with anyio.fail_after(opts.get("timeout")):
+                # Inside the timeout scope, so a configured timeout also bounds
+                # waiting on a peer whose run() has not started yet.
+                await self._wait_ready()
+                assert self._on_request is not None
+                # Synthesize an id: the DispatchContext contract reserves None for notifications.
+                self._next_id += 1
+                dctx = self._make_context(on_progress=opts.get("on_progress"), request_id=self._next_id)
                 try:
                     return await self._on_request(dctx, method, params)
                 except MCPError:
@@ -167,7 +242,13 @@ class DirectDispatcher:
             ) from None
 
     async def _dispatch_notify(self, method: str, params: Mapping[str, Any] | None) -> None:
-        await self._ready.wait()
+        try:
+            await self._wait_ready()
+        except MCPError:
+            # Notifications are fire-and-forget: a notify to a closed peer is
+            # dropped, not raised back into the sender's call.
+            logger.debug("dropped notification %r to closed DirectDispatcher", method)
+            return
         assert self._on_notify is not None
         dctx = self._make_context()
         await self._on_notify(dctx, method, params)

src/mcp/shared/dispatcher.py

@@ -55,6 +55,13 @@ class CallOptions(TypedDict, total=False):
     timeout: float
     """Seconds to wait for a result before raising and sending `notifications/cancelled`."""
 
+    cancel_on_abandon: bool
+    """Whether abandoning this request (timeout or caller cancellation) sends `notifications/cancelled`.
+
+    Defaults to `True`. Set `False` for requests the protocol forbids cancelling, such as `initialize`.
+    Also suppressed when resumption hints reach the transport, or when the request was never written.
+    """
+
     on_progress: ProgressFnT
     """Receive `notifications/progress` updates for this request."""
 
@@ -97,9 +104,6 @@ class Outbound(Protocol):
     ) -> dict[str, Any]:
         """Send a request and await its raw result dict.
 
-        `opts` carries per-call `timeout` / `on_progress` / resumption hints;
-        see `CallOptions`.
-
         Raises:
             MCPError: If the peer responded with an error, or the handler
                 raised. Implementations normalize all handler exceptions to
@@ -187,6 +191,8 @@ class Dispatcher(Outbound, Protocol[TransportT_co]):
 
     Implementations own correlation of outbound requests to inbound results, the
     receive loop, per-request concurrency, and cancellation/progress wiring.
+
+    The lifecycle surface is provisional; `run()` may change before v2 stable.
     """
 
     async def run(

src/mcp/shared/jsonrpc_dispatcher.py

@@ -1,21 +1,8 @@
-"""JSON-RPC `Dispatcher` implementation.
+"""JSON-RPC `Dispatcher` over the `SessionMessage` stream contract all transports speak.
 
-Consumes the existing `SessionMessage`-based stream contract that all current
+Owns request-id correlation, the receive loop, per-request task isolation,
-transports (stdio, SSE, streamable HTTP) speak. Owns request-id correlation,
+cancellation/progress wiring, and the single exception-to-wire boundary;
-the receive loop, per-request task isolation, cancellation/progress wiring, and
+methods and params are otherwise opaque strings and dicts.
-the single exception-to-wire boundary.
-
-The MCP type layer (`ServerRunner`, `Context`, `Client`) sits above this and
-sees only `(ctx, method, params) -> dict`. Transports sit below and see only
-`SessionMessage` reads/writes.
-
-The dispatcher is *mostly* MCP-agnostic - methods/params are opaque strings and
-dicts - but it intercepts `notifications/cancelled` and
-`notifications/progress` because request correlation, cancellation and
-progress are exactly the wiring this layer exists to provide. Those few wire
-shapes are extracted with structural `match` patterns (no casts, no
-`mcp.types` model coupling); a malformed payload simply fails to match and
-the correlation is skipped.
 """
 
 from __future__ import annotations
@@ -24,17 +11,21 @@ import contextvars
 import logging
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass, field
-from typing import Any, Generic, Literal, TypeVar, cast, overload
+from functools import partial
+from typing import Any, Generic, Literal, cast
 
 import anyio
 import anyio.abc
+import anyio.lowlevel
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from opentelemetry.trace import SpanKind
 from pydantic import ValidationError
+from typing_extensions import TypeVar
 
+from mcp.shared._compat import resync_tracer
 from mcp.shared._otel import inject_trace_context, otel_span
 from mcp.shared._stream_protocols import ReadStream, WriteStream
-from mcp.shared.dispatcher import CallOptions, Dispatcher, OnNotify, OnRequest, ProgressFnT
+from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, OnNotify, OnRequest, ProgressFnT
 from mcp.shared.exceptions import MCPError, NoBackChannelError
 from mcp.shared.message import (
     ClientMessageMetadata,
@@ -47,7 +38,6 @@ from mcp.types import (
     CONNECTION_CLOSED,
     INTERNAL_ERROR,
     INVALID_PARAMS,
-    REQUEST_CANCELLED,
     REQUEST_TIMEOUT,
     ErrorData,
     JSONRPCError,
@@ -63,23 +53,22 @@ __all__ = ["JSONRPCDispatcher"]
 
 logger = logging.getLogger(__name__)
 
-TransportT = TypeVar("TransportT", bound=TransportContext)
+_ABANDON_WRITE_TIMEOUT: float = 5
+"""Bound for courtesy-cancel writes on the abandon paths; the caller-cancel
+arm shields its write, so a wedged transport would otherwise hang it uncancellably."""
+
+_SHUTDOWN_WRITE_TIMEOUT: float = 1
+"""Tighter bound for the shutdown-arm error write so a wedged transport can't hold session close."""
+
+TransportT = TypeVar("TransportT", bound=TransportContext, default=TransportContext)
 
 PeerCancelMode = Literal["interrupt", "signal"]
-"""How inbound `notifications/cancelled` is applied to a running handler.
+"""How `notifications/cancelled` is applied: `"interrupt"` (default) cancels
-
+the handler's scope; `"signal"` only sets `ctx.cancel_requested`."""
-`"interrupt"` (default) cancels the handler's scope. `"signal"` only sets
-`ctx.cancel_requested` and lets the handler observe it cooperatively.
-"""
 
 
 def _coerce_id(request_id: RequestId) -> RequestId:
-    """Coerce a string request ID to int when it's a valid int literal.
+    """Coerce a stringified int request ID back to int so a peer-echoed ID still correlates (matches the TS SDK)."""
-
-    `_allocate_id` only ever produces `int` keys for `_pending`, but a peer
-    may echo the ID back as a JSON string. The TypeScript SDK and `BaseSession`
-    both perform this coercion at lookup time so the response still correlates.
-    """
     if isinstance(request_id, str):
         try:
             return int(request_id)
@@ -113,12 +102,7 @@ class _JSONRPCDispatchContext(Generic[TransportT]):
     _dispatcher: JSONRPCDispatcher[TransportT]
     _request_id: RequestId | None
     message_metadata: MessageMetadata = None  # TODO(maxisbey): remove for Context rework
-    """The transport-attached `SessionMessage.metadata` for this inbound message.
+    """Transport-attached `SessionMessage.metadata` that the server lifts onto its request context."""
-
-    Carries `ServerMessageMetadata` (HTTP request, SSE stream-close callbacks)
-    that the server lifts onto its request context. `None` for transports
-    that attach nothing.
-    """
     _progress_token: ProgressToken | None = None
     _closed: bool = False
     cancel_requested: anyio.Event = field(default_factory=anyio.Event)
@@ -166,13 +150,7 @@ def _default_transport_builder(_meta: MessageMetadata) -> TransportContext:
 
 
 def _shielded_progress(fn: ProgressFnT) -> ProgressFnT:
-    """Wrap a user progress callback so it can't crash the dispatcher.
+    """Wrap a user progress callback so an exception can't cancel the dispatcher's task group."""
-
-    The callback runs as a bare task in the dispatcher's task group; an
-    uncaught exception would cancel every sibling (the read loop and all
-    in-flight requests). Swallow and log instead, matching the previous
-    receive-loop's behavior.
-    """
 
     async def _wrapped(progress: float, total: float | None, message: str | None) -> None:
         try:
@@ -183,61 +161,57 @@ def _shielded_progress(fn: ProgressFnT) -> ProgressFnT:
     return _wrapped
 
 
-def _outbound_metadata(related_request_id: RequestId | None, opts: CallOptions | None) -> MessageMetadata:
+def _contained_notify(fn: OnNotify) -> OnNotify:
-    """Choose the `SessionMessage.metadata` for an outgoing request/notification.
+    """Wrap a notification handler so it can't crash the dispatcher (same boundary as `_shielded_progress`)."""
 
-    `ServerMessageMetadata` tags a server-to-client message with the inbound
+    async def _wrapped(dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None) -> None:
-    request it belongs to (so streamable-HTTP can route it onto that request's
+        try:
-    SSE stream). `ClientMessageMetadata` carries resumption hints to the
+            await fn(dctx, method, params)
-    client transport. `None` is the common case.
+        except Exception:
+            logger.exception("notification handler for %r raised", method)
 
-    `SessionMessage.metadata` carries exactly one of these, so when
+    return _wrapped
-    `related_request_id` is set it takes precedence and any resumption hints
+
-    in `opts` are dropped (with a debug log): requests made from a dispatch
+
-    context are routed onto the inbound request's stream, not resumed.
+@dataclass(slots=True, frozen=True)
+class _OutboundPlan:
+    """Outbound metadata plus whether abandoning the request sends a courtesy `notifications/cancelled`."""
+
+    metadata: MessageMetadata
+    cancel_on_abandon: bool
+
+
+def _plan_outbound(related_request_id: RequestId | None, opts: CallOptions | None) -> _OutboundPlan:
+    """Choose the outbound `SessionMessage.metadata` and the abandon-cancellation policy.
+
+    `related_request_id` wins over resumption hints (they are dropped). Only
+    hints that actually reach the transport suppress the courtesy cancel - a
+    request that is neither resumable nor cancelled would leak the peer's work.
     """
+    opts = opts or {}
+    cancel_on_abandon = opts.get("cancel_on_abandon", True)
+    token = opts.get("resumption_token")
+    on_token = opts.get("on_resumption_token")
     if related_request_id is not None:
-        if opts and (opts.get("resumption_token") is not None or opts.get("on_resumption_token") is not None):
+        if token is not None or on_token is not None:
             logger.debug(
                 "dropping resumption hints: related_request_id %r takes precedence on metadata", related_request_id
             )
-        return ServerMessageMetadata(related_request_id=related_request_id)
+        return _OutboundPlan(ServerMessageMetadata(related_request_id=related_request_id), cancel_on_abandon)
-    if opts:
+    if token is not None or on_token is not None:
-        token = opts.get("resumption_token")
+        return _OutboundPlan(
-        on_token = opts.get("on_resumption_token")
+            ClientMessageMetadata(resumption_token=token, on_resumption_token_update=on_token),
-        if token is not None or on_token is not None:
+            cancel_on_abandon=False,
-            return ClientMessageMetadata(resumption_token=token, on_resumption_token_update=on_token)
+        )
-    return None
+    return _OutboundPlan(None, cancel_on_abandon)
 
 
 class JSONRPCDispatcher(Dispatcher[TransportT]):
-    """`Dispatcher` over the existing `SessionMessage` stream contract.
+    """`Dispatcher` over the `SessionMessage` stream contract.
 
-    Inherits the `Dispatcher` Protocol explicitly so pyright checks
+    Explicit Protocol base so pyright checks conformance at the class definition.
-    conformance at the class definition rather than at first use.
     """
 
-    @overload
-    def __init__(
-        self: JSONRPCDispatcher[TransportContext],
-        read_stream: ReadStream[SessionMessage | Exception],
-        write_stream: WriteStream[SessionMessage],
-        *,
-        peer_cancel_mode: PeerCancelMode = "interrupt",
-        raise_handler_exceptions: bool = False,
-        inline_methods: frozenset[str] = frozenset(),
-    ) -> None: ...
-    @overload
-    def __init__(
-        self,
-        read_stream: ReadStream[SessionMessage | Exception],
-        write_stream: WriteStream[SessionMessage],
-        *,
-        transport_builder: Callable[[MessageMetadata], TransportT],
-        peer_cancel_mode: PeerCancelMode = "interrupt",
-        raise_handler_exceptions: bool = False,
-        inline_methods: frozenset[str] = frozenset(),
-    ) -> None: ...
     def __init__(
         self,
         read_stream: ReadStream[SessionMessage | Exception],
@@ -247,33 +221,41 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
         peer_cancel_mode: PeerCancelMode = "interrupt",
         raise_handler_exceptions: bool = False,
         inline_methods: frozenset[str] = frozenset(),
+        on_stream_exception: Callable[[Exception], Awaitable[None]] | None = None,
     ) -> None:
+        """Wire a dispatcher over a transport's `SessionMessage` stream pair.
+
+        Args:
+            transport_builder: Builds each message's `TransportContext` from
+                its `SessionMessage.metadata`.
+            raise_handler_exceptions: Re-raise handler exceptions out of
+                `run()` after the error response is written.
+            inline_methods: Methods awaited in the read loop before the next
+                message is dequeued (e.g. `initialize`); an inline handler
+                that awaits the peer deadlocks the parked loop.
+            on_stream_exception: Observer for `Exception` items on the read
+                stream; without it they are debug-logged and dropped. Awaited
+                inline in the read loop, so a slow observer stalls dispatch.
+        """
         self._read_stream = read_stream
         self._write_stream = write_stream
-        # The overloads guarantee that when `transport_builder` is omitted,
+        # With transport_builder omitted, TransportT defaults to
-        # `TransportT` is `TransportContext`, so the default is type-correct;
+        # TransportContext; pyright can't connect the two, hence the cast.
-        # pyright can't see across overloads, hence the cast.
         self._transport_builder = cast(
             "Callable[[MessageMetadata], TransportT]",
             transport_builder or _default_transport_builder,
         )
         self._peer_cancel_mode: PeerCancelMode = peer_cancel_mode
         self._raise_handler_exceptions = raise_handler_exceptions
-        # Request methods handled inline in the read loop (awaited before the
-        # next message is dequeued) instead of spawned concurrently. Use for
-        # methods whose side effects must be observable to the next message,
-        # e.g. `initialize`, so a pipelined follow-up sees the initialized state.
-        # Only suitable for handlers that complete quickly, since inline handling
-        # blocks dequeuing; a handler that awaits the peer (`send_raw_request`)
-        # while inline will deadlock because the parked read loop cannot dequeue
-        # the response.
         self._inline_methods = inline_methods
+        self._on_stream_exception = on_stream_exception
 
         self._next_id = 0
         self._pending: dict[RequestId, _Pending] = {}
         self._in_flight: dict[RequestId, _InFlight[TransportT]] = {}
         self._tg: anyio.abc.TaskGroup | None = None
         self._running = False
+        self._closed = False
 
     async def send_raw_request(
         self,
@@ -285,82 +267,105 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
     ) -> dict[str, Any]:
         """Send a JSON-RPC request and await its response.
 
-        `_related_request_id` is set only by `_JSONRPCDispatchContext` when a
+        `_related_request_id` is set only by `_JSONRPCDispatchContext` so that
-        handler makes a server-to-client request mid-flight; it routes the
+        mid-handler requests route onto the inbound request's SSE stream.
-        outgoing message onto the correct per-request SSE stream (SHTTP) via
-        `ServerMessageMetadata`. Top-level callers leave it `None`.
 
         Raises:
-            MCPError: The peer responded with a JSON-RPC error; or
+            MCPError: Peer error response; `REQUEST_TIMEOUT` if
-                `REQUEST_TIMEOUT` if `opts["timeout"]` elapsed; or
+                `opts["timeout"]` elapsed; `CONNECTION_CLOSED` if the
-                `CONNECTION_CLOSED` if the dispatcher shut down while
+                transport closed or the dispatcher shut down.
-                awaiting the response.
+            RuntimeError: Called before `run()`.
-            RuntimeError: Called before `run()` has started or after it has
-                finished.
         """
+        # Post-close sends get the same CONNECTION_CLOSED contract as in-flight waiters.
+        if self._closed:
+            raise MCPError(code=CONNECTION_CLOSED, message="Connection closed")
         if not self._running:
-            raise RuntimeError("JSONRPCDispatcher.send_raw_request called before run() / after close")
+            raise RuntimeError("JSONRPCDispatcher.send_raw_request called before run()")
         opts = opts or {}
         request_id = self._allocate_id()
         out_params = dict(params) if params is not None else {}
         out_meta = dict(out_params.get("_meta") or {})
         on_progress = opts.get("on_progress")
         if on_progress is not None:
-            # The caller wants progress updates. The spec mechanism is: include
+            # The request id doubles as the progress token, so `_pending[token]` finds `on_progress` directly.
-            # `_meta.progressToken` on the request; the peer echoes that token on
-            # any `notifications/progress` it sends. We use the request id as the
-            # token so the receive loop can find this `_Pending.on_progress` by
-            # `_pending[token]` without a second lookup table.
             out_meta["progressToken"] = request_id
         out_params["_meta"] = out_meta
 
-        # buffer=1: at most one outcome is ever delivered. A `WouldBlock` from
+        # buffer=1: a close signal can arrive before the waiter parks in receive();
-        # `_resolve_pending`/`_fan_out_closed` means the waiter already has an
+        # a WouldBlock later just means the waiter already has its one outcome.
-        # outcome and dropping the late/redundant signal is correct. buffer=0
-        # is unsafe - there's a window between registering `_pending[id]` and
-        # parking in `receive()` where a close signal would be lost.
         send, receive = anyio.create_memory_object_stream[dict[str, Any] | ErrorData](1)
         pending = _Pending(send=send, receive=receive, on_progress=on_progress)
         self._pending[request_id] = pending
 
-        metadata = _outbound_metadata(_related_request_id, opts)
+        plan = _plan_outbound(_related_request_id, opts)
+        # Spec MUST: only previously-issued requests may be cancelled. A write
+        # interrupted by cancellation may still have delivered (a memory-stream
+        # send can hand its item to the receiver and still raise), so a started
+        # write counts as issued: the peer ignores a cancel for an id it never
+        # saw, while skipping it would leak a delivered request's handler.
+        request_write_started = False
+        timeout_armed = False
+
         target = out_params.get("name")
         span_name = f"MCP send {method}{f' {target}' if isinstance(target, str) else ''}"
-        # TODO(maxisbey): the otel span + inject below mirror
+        # TODO(maxisbey): move the otel span + inject into an outbound
-        # BaseSession.send_request for parity. They belong in an outbound
+        # middleware once that seam exists; the dispatcher should not own otel.
-        # middleware (symmetric with otel_middleware on the inbound side) once
-        # that seam exists; the dispatcher should not own otel.
         try:
             with otel_span(
                 span_name,
                 kind=SpanKind.CLIENT,
                 attributes={"mcp.method.name": method, "jsonrpc.request.id": str(request_id)},
             ):
-                # Inject W3C trace context into _meta (SEP-414). With a no-op
+                # SEP-414: inject W3C trace context; `_meta` stays on the wire even with a no-op tracer.
-                # tracer this writes nothing, but `_meta` itself is still
-                # present on the wire (and the interaction suite pins that).
                 inject_trace_context(out_meta)
                 msg = JSONRPCRequest(jsonrpc="2.0", id=request_id, method=method, params=out_params)
-                await self._write(msg, metadata)
+                # Surface a pre-existing cancellation while the request provably
+                # never started; past this point a cancelled write counts as issued.
+                await anyio.lowlevel.checkpoint_if_cancelled()
+                request_write_started = True
+                try:
+                    await self._write(msg, plan.metadata)
+                except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+                    # Transport tore down before run() noticed EOF; surface the documented contract.
+                    raise MCPError(code=CONNECTION_CLOSED, message="Connection closed") from None
                 with anyio.fail_after(opts.get("timeout")):
+                    timeout_armed = True
                     outcome = await receive.receive()
         except TimeoutError:
-            # Spec-recommended courtesy: tell the peer we've given up so it can
+            if not timeout_armed:
-            # stop work and free resources. v1's BaseSession.send_request does
+                # `fail_after` arms only after the write, so this TimeoutError is the
-            # NOT do this; it's new behaviour.
+                # transport's own bounded send() failing - a transport error, not
-            await self._cancel_outbound(request_id, f"timed out after {opts.get('timeout')}s", _related_request_id)
+                # `opts["timeout"]` elapsing. Propagate it raw (v1 kept the write
+                # outside the timeout-catching try and did the same).
+                raise
+            # Courtesy cancel (spec-recommended, new vs v1) so the peer stops work;
+            # unshielded so an outer caller cancellation can still interrupt the write.
+            if plan.cancel_on_abandon:
+                await self._final_write(
+                    partial(
+                        self._cancel_outbound,
+                        request_id,
+                        f"timed out after {opts.get('timeout')}s",
+                        _related_request_id,
+                    ),
+                    shield=False,
+                    timeout=_ABANDON_WRITE_TIMEOUT,
+                    describe=f"courtesy cancel for timed-out request {request_id!r}",
+                )
             raise MCPError(code=REQUEST_TIMEOUT, message=f"Request {method!r} timed out") from None
         except anyio.get_cancelled_exc_class():
-            # Our caller's scope was cancelled. We're already inside a cancelled
+            # Caller cancelled: bare awaits re-raise here, so the shielded helper
-            # scope, so any bare `await` here re-raises immediately - shield to
+            # lets the courtesy cancel go out before we propagate.
-            # let the courtesy cancel notification go out before we propagate.
+            if plan.cancel_on_abandon and request_write_started:
-            with anyio.CancelScope(shield=True):
+                await self._final_write(
-                await self._cancel_outbound(request_id, "caller cancelled", _related_request_id)
+                    partial(self._cancel_outbound, request_id, "caller cancelled", _related_request_id),
+                    shield=True,
+                    timeout=_ABANDON_WRITE_TIMEOUT,
+                    describe=f"courtesy cancel for caller-cancelled request {request_id!r}",
+                )
             raise
         finally:
-            # Always remove the waiter, even on cancel/timeout, so a late
+            # Remove the waiter on every path so a late response is dropped, not leaked.
-            # response from the peer (race) hits a closed stream and is dropped
-            # in `_dispatch` rather than leaking.
             self._pending.pop(request_id, None)
             send.close()
             receive.close()
@@ -376,15 +381,26 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
         *,
         _related_request_id: RequestId | None = None,
     ) -> None:
-        # Leave `params` unset (not explicitly None) when there are none:
+        """Send a fire-and-forget notification.
-        # transports serialize with `exclude_unset=True`, and an explicit None
+
-        # would survive as `"params": null`, which JSON-RPC 2.0 forbids and
+        Fire-and-forget all the way: a post-close send or a write onto a
-        # strict peers (e.g. the TypeScript SDK's zod schemas) reject.
+        torn-down transport drops the notification with a debug log instead
+        of raising (same policy as the response writes and `ctx.notify`).
+        """
+        if self._closed:
+            logger.debug("dropped %s: dispatcher closed", method)
+            return
+        # Leave `params` unset when None: with `exclude_unset=True` an explicit
+        # None would serialize as `"params": null`, which JSON-RPC 2.0 forbids.
         if params is not None:
             msg = JSONRPCNotification(jsonrpc="2.0", method=method, params=dict(params))
         else:
             msg = JSONRPCNotification(jsonrpc="2.0", method=method)
-        await self._write(msg, _outbound_metadata(_related_request_id, None))
+        try:
+            await self._write(msg, _plan_outbound(_related_request_id, None).metadata)
+        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+            # Transport tore down before run() noticed EOF.
+            logger.debug("dropped %s: write stream closed", method)
 
     async def run(
         self,
@@ -395,50 +411,44 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
     ) -> None:
         """Drive the receive loop until the read stream closes.
 
-        Each inbound request is handled in its own task in an internal task
+        `task_status.started()` fires once `send_raw_request` is usable.
-        group; `task_status.started()` fires once that group is open, so
+        Single-shot: once the loop ends the dispatcher stays closed and cannot be restarted.
-        `await tg.start(dispatcher.run, ...)` resumes when `send_raw_request`
-        is usable.
         """
         try:
-            async with anyio.create_task_group() as tg:
+            # LIFO exits: the write stream closes only after the task-group join, so teardown writes still land.
-                self._tg = tg
+            async with self._write_stream:
-                self._running = True
+                async with anyio.create_task_group() as tg:
-                task_status.started()
+                    self._tg = tg
-                try:
+                    self._running = True
-                    async with self._read_stream, self._write_stream:
+                    task_status.started()
-                        try:
+                    try:
-                            async for item in self._read_stream:
+                        async with self._read_stream:
-                                # Duck-typed: `_context_streams.ContextReceiveStream`
+                            try:
-                                # exposes `.last_context` (the sender's contextvars
+                                async for item in self._read_stream:
-                                # snapshot per message). Plain memory streams don't.
+                                    # Duck-typed: only `ContextReceiveStream` carries the
-                                sender_ctx: contextvars.Context | None = getattr(
+                                    # sender's per-message contextvars snapshot.
-                                    self._read_stream, "last_context", None
+                                    sender_ctx: contextvars.Context | None = getattr(
-                                )
+                                        self._read_stream, "last_context", None
-                                await self._dispatch(item, on_request, on_notify, sender_ctx)
+                                    )
-                        except anyio.ClosedResourceError:
+                                    await self._dispatch(item, on_request, on_notify, sender_ctx)
-                            # The transport closed our receive end and we looped
+                            except anyio.ClosedResourceError:
-                            # back to `__anext__` on the now-closed stream
+                                # Receive end closed under us (stateless SHTTP teardown); same as EOF.
-                            # (stateless SHTTP teardown). Same as EOF.
+                                logger.debug("read stream closed by transport; treating as EOF")
-                            logger.debug("read stream closed by transport; treating as EOF")
+                        # EOF: wake blocked `send_raw_request` waiters with CONNECTION_CLOSED.
-                    # Read stream EOF: wake any blocked `send_raw_request` waiters
+                        self._running = False
-                    # (callers outside this task group) with CONNECTION_CLOSED.
+                        self._closed = True
-                    self._running = False
+                        self._fan_out_closed()
-                    self._fan_out_closed()
+                    finally:
-                finally:
+                        # Cancel in-flight handlers; otherwise the task-group join
-                    # Transport closed: cancel in-flight handlers. Without this
+                        # waits on handlers whose callers are already gone.
-                    # the task-group join waits for them, and a handler that
+                        tg.cancel_scope.cancel()
-                    # outlives its caller (its request timed out client-side, or
-                    # the client disconnected mid-call) would keep `run()` from
-                    # returning forever. Same behaviour as `Server.run()` before
-                    # the dispatcher rework.
-                    tg.cancel_scope.cancel()
         finally:
-            # Covers the cancel/crash paths where the inline fan-out above is
+            # Covers cancel/crash paths that skip the inline fan-out; idempotent.
-            # never reached. Idempotent.
             self._running = False
+            self._closed = True
             self._tg = None
             self._fan_out_closed()
+            await resync_tracer()
 
     async def _dispatch(
         self,
@@ -449,13 +459,17 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
     ) -> None:
         """Route one inbound item.
 
-        Everything here is `send_nowait` or `_spawn`; the only `await` is for
+        Only `inline_methods` requests and the `on_stream_exception` observer
-        `inline_methods` requests, which deliberately block dequeuing until
+        are awaited; any other `await` would head-of-line block the read loop.
-        handled. Any other `await` would let one slow message head-of-line
-        block the entire read loop.
         """
         if isinstance(item, Exception):
-            logger.debug("transport yielded exception: %r", item)
+            if self._on_stream_exception is None:
+                logger.debug("transport yielded exception: %r", item)
+                return
+            try:
+                await self._on_stream_exception(item)
+            except Exception:
+                logger.exception("on_stream_exception observer raised")
             return
         metadata = item.metadata
         msg = item.message
@@ -467,9 +481,7 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
             case JSONRPCResponse():
                 self._resolve_pending(msg.id, msg.result)
             case JSONRPCError():  # pragma: no branch
-                # `id` may be None per JSON-RPC (parse error before id known).
+                # Exhaustive over JSONRPCMessage, so the no-match arc is unreachable.
-                # The match is exhaustive over JSONRPCMessage; the no-match arc
-                # on this final case is unreachable.
                 self._resolve_pending(msg.id, msg.error)
 
     async def _dispatch_request(
@@ -481,8 +493,7 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
     ) -> None:
         progress_token: ProgressToken | None
         match req.params:
-            # The bool guard matters: `int()` patterns match bool (a subclass),
+            # bool subclasses int: without the guard True would alias request id 1.
-            # and `True == 1` would alias dict lookups to request id 1.
             case {"_meta": {"progressToken": str() | int() as progress_token}} if not isinstance(progress_token, bool):
                 pass
             case _:
@@ -490,9 +501,7 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
         try:
             transport_ctx = self._transport_builder(metadata)
         except Exception:
-            # Containment boundary for the user-supplied builder: a raising
+            # A raising builder must cost only this message, not the connection.
-            # builder must cost only this message, not the whole connection
-            # (the exception would otherwise escape into run()'s read loop).
             logger.exception("transport_builder raised; rejecting request %r", req.id)
             self._spawn(
                 self._write_error,
@@ -509,20 +518,13 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
             _progress_token=progress_token,
         )
         scope = anyio.CancelScope()
-        # TODO(maxisbey): the spec puts request-id uniqueness on the sender;
+        # TODO(maxisbey): duplicate ids blind-overwrite (v1/TS parity); revisit
-        # neither v1 nor the TS SDK guards a duplicate id here, so for now we
+        # rejecting with INVALID_REQUEST. Key coerced so a stringified
-        # blind-overwrite (parity). Revisit rejecting with INVALID_REQUEST.
+        # `notifications/cancelled` id still correlates.
-        # Coerced key so `notifications/cancelled` correlates regardless of
-        # whether the peer stringifies the id between request and cancel
-        # (`_dispatch_notification` coerces at lookup; responses still echo
-        # `req.id` verbatim).
         self._in_flight[_coerce_id(req.id)] = _InFlight(scope=scope, dctx=dctx)
         if req.method in self._inline_methods:
-            # Spawn (so `sender_ctx` applies, matching the concurrent path) but
+            # Spawn so `sender_ctx` applies, but park the read loop until the
-            # park the read loop until the handler returns; that's the inline
+            # handler returns - that's the inline ordering guarantee.
-            # ordering guarantee. Because the read loop is parked, a handler
-            # that awaits the peer here (e.g. `dctx.send_raw_request`) will
-            # deadlock: the response can never be dequeued.
             done = anyio.Event()
 
             async def _run_inline() -> None:
@@ -546,17 +548,12 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
         """Route one inbound notification.
 
         `notifications/cancelled` and `notifications/progress` are intercepted
-        here because they correlate against JSON-RPC request IDs - the
+        here (they correlate against the `_in_flight`/`_pending` tables this
-        `_in_flight` / `_pending` tables this layer owns - so no higher layer
+        layer owns) and still teed to `on_notify` afterwards.
-        can act on them. Both are still teed to `on_notify` afterwards, so
-        middleware and registered notification handlers observe every inbound
-        notification. See the module docstring for the design rationale.
         """
         if msg.method == "notifications/cancelled":
             match msg.params:
-                # The bool guards here and below matter: `int()` patterns match
+                # bool subclasses int: the guards keep True from aliasing request id 1.
-                # bool (a subclass), and `True == 1` would alias the dict lookup
-                # to the entry keyed by request id 1.
                 case {"requestId": str() | int() as rid} if (
                     not isinstance(rid, bool) and (in_flight := self._in_flight.get(_coerce_id(rid))) is not None
                 ):
@@ -565,9 +562,6 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
                         in_flight.scope.cancel()
                 case _:
                     pass
-            # fall through: cancelled is also teed to on_notify so middleware
-            # and registered handlers can observe it (matches DirectDispatcher,
-            # which forwards every notification).
         elif msg.method == "notifications/progress":
             match msg.params:
                 case {"progressToken": str() | int() as token, "progress": int() | float() as progress} if (
@@ -587,18 +581,16 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
                     )
                 case _:
                     pass
-            # fall through: progress is also teed to on_notify
         try:
             transport_ctx = self._transport_builder(metadata)
         except Exception:
-            # Same containment boundary as `_dispatch_request`: a raising
+            # Same containment as `_dispatch_request`: drop the notification, keep the loop.
-            # builder drops this notification instead of killing the read loop.
             logger.exception("transport_builder raised; dropping notification %r", msg.method)
             return
         dctx = _JSONRPCDispatchContext(
             transport=transport_ctx, _dispatcher=self, _request_id=None, message_metadata=metadata
         )
-        self._spawn(on_notify, dctx, msg.method, msg.params, sender_ctx=sender_ctx)
+        self._spawn(_contained_notify(on_notify), dctx, msg.method, msg.params, sender_ctx=sender_ctx)
 
     def _resolve_pending(self, request_id: RequestId | None, outcome: dict[str, Any] | ErrorData) -> None:
         pending = self._pending.get(_coerce_id(request_id)) if request_id is not None else None
@@ -618,10 +610,8 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
     ) -> None:
         """Schedule `fn(*args)` in the run() task group, propagating the sender's contextvars.
 
-        ASGI middleware (auth, OTel) sets contextvars on the request task that
+        ASGI middleware (auth, OTel) sets contextvars on the task that wrote the
-        wrote into the read stream. `Context.run(tg.start_soon, ...)` makes
+        message; `Context.run` makes the spawned handler inherit that context.
-        the spawned handler inherit *that* context instead of the receive
-        loop's, so `auth_context_var` and OTel spans survive.
         """
         assert self._tg is not None
         if sender_ctx is not None:
@@ -632,10 +622,9 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
     def _fan_out_closed(self) -> None:
         """Wake every pending `send_raw_request` waiter with `CONNECTION_CLOSED`.
 
-        Synchronous (uses `send_nowait`) because it's called from `finally`
+        Synchronous: callers may be inside a cancelled scope. Idempotent.
-        which may be inside a cancelled scope. Idempotent.
         """
-        closed = ErrorData(code=CONNECTION_CLOSED, message="connection closed")
+        closed = ErrorData(code=CONNECTION_CLOSED, message="Connection closed")
         for pending in self._pending.values():
             try:
                 pending.send.send_nowait(closed)
@@ -652,61 +641,65 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
     ) -> None:
         """Run `on_request` for one inbound request and write its response.
 
-        This is the single exception-to-wire boundary: handler exceptions are
+        The single exception-to-wire boundary: handler exceptions become `JSONRPCError` here.
-        caught here and serialized to `JSONRPCError`. Nothing above this in
-        the stack constructs wire errors.
         """
+        answer_write_started = False
         try:
             with scope:
                 try:
                     result = await on_request(dctx, req.method, req.params)
                 finally:
-                    # Handler done: close the back-channel (detached work that
+                    # Close the back-channel and drop from `_in_flight`; no checkpoint
-                    # later calls `dctx.send_raw_request()` should see
+                    # since handler return, so a peer cancel can't interleave.
-                    # `NoBackChannelError`) and drop from `_in_flight` so a
+                    # Identity guard: don't evict a duplicate id's newer entry.
-                    # late `notifications/cancelled` is a no-op rather than
-                    # racing the result write below. No checkpoint between
-                    # handler return and the pop, so the cancel can't
-                    # interleave there.
                     dctx.close()
-                    self._in_flight.pop(_coerce_id(req.id), None)
+                    key = _coerce_id(req.id)
+                    if (entry := self._in_flight.get(key)) is not None and entry.dctx is dctx:
+                        del self._in_flight[key]
+                # A write interrupted by cancellation may still have delivered
+                # (a memory-stream send can hand its item to the receiver and
+                # still raise), so a started answer write counts as sent below:
+                # peers drop late responses, while a second answer for one id
+                # would break JSON-RPC.
+                answer_write_started = True
                 await self._write_result(req.id, result)
-            if scope.cancel_called:
+            if scope.cancelled_caught:
-                # Peer-cancel: `_dispatch_notification` cancelled this scope
+                # anyio absorbs the scope's own cancel at __exit__, and
-                # while the handler was running. anyio swallows a scope's *own*
+                # `cancelled_caught` (unlike `cancel_called`) guarantees the
-                # cancel at __exit__, so execution lands here rather than the
+                # result write above did not happen - no double response.
-                # `except cancelled` arm below.
+                # TODO(maxisbey): spec says SHOULD NOT respond after cancel;
-                # TODO(maxisbey): spec says SHOULD NOT respond after cancel.
+                # the existing server always has, so match that for now.
-                # The existing server always has, so match that for now.
+                answer_write_started = True
                 await self._write_error(req.id, ErrorData(code=0, message="Request cancelled"))
         except anyio.get_cancelled_exc_class():
-            # Outer-cancel: run()'s task group is shutting down. Any bare
+            # Shutdown: answer the request so the peer isn't left waiting - unless
-            # `await` here re-raises immediately, so shield the courtesy write.
+            # an answer write already started (it may have reached the transport;
-            with anyio.CancelScope(shield=True):
+            # prefer possibly-zero answers over possibly-two). The shielded helper
-                await self._write_error(req.id, ErrorData(code=REQUEST_CANCELLED, message="Request cancelled"))
+            # is needed because bare awaits re-raise here.
+            if not answer_write_started:
+                await self._final_write(
+                    partial(self._write_error, req.id, ErrorData(code=CONNECTION_CLOSED, message="Connection closed")),
+                    shield=True,
+                    timeout=_SHUTDOWN_WRITE_TIMEOUT,
+                    describe=f"shutdown error response for request {req.id!r}",
+                )
             raise
         except MCPError as e:
             await self._write_error(req.id, e.error)
         except ValidationError:
-            # TODO(maxisbey): data="" is pinned compat with the existing
+            # TODO(maxisbey): data="" pins existing-server compat (no pydantic
-            # server (which never leaked pydantic error text onto the wire).
+            # text on the wire); revisit per the suite's divergence entry.
-            # Consider putting the validation detail in `data` once the
-            # interaction suite's divergence entry is resolved.
             await self._write_error(
                 req.id, ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data="")
             )
         except Exception as e:
             logger.exception("handler for %r raised", req.method)
-            # TODO(maxisbey): code=0 is pinned compat with the existing
+            # TODO(maxisbey): code=0 pins existing-server compat; JSON-RPC says
-            # server's `_handle_request`. JSON-RPC says INTERNAL_ERROR
+            # INTERNAL_ERROR. Revisit per the suite's divergence entry.
-            # (-32603); revisit once the suite's divergence entry is resolved.
             await self._write_error(req.id, ErrorData(code=0, message=str(e)))
             if self._raise_handler_exceptions:
                 raise
-        # No outer `_in_flight` pop here: the inner `finally` above already
+        # No `_in_flight` pop here: the inner finally covers every path, and a late pop could evict a reused id.
-        # removes the entry on every path out of the handler, and a second
-        # pop after the awaited response writes could evict a newer request
-        # that reused the id during that window.
 
     def _allocate_id(self) -> int:
         self._next_id += 1
@@ -727,16 +720,31 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
         except (anyio.BrokenResourceError, anyio.ClosedResourceError):
             logger.debug("dropped error for %r: write stream closed", request_id)
 
+    async def _final_write(
+        self,
+        write: Callable[[], Awaitable[None]],
+        *,
+        shield: bool,
+        timeout: float,
+        describe: str,
+    ) -> None:
+        """Attempt one last write under the shared abandon/teardown policy.
+
+        `shield=True` is for arms already inside a cancelled scope (a bare
+        `await` would re-raise); the bound keeps a wedged transport write
+        from becoming an uncancellable hang.
+        """
+        with anyio.move_on_after(timeout, shield=shield) as scope:
+            await write()
+        if scope.cancelled_caught:
+            logger.warning("%s gave up: transport write blocked", describe)
+
     async def _cancel_outbound(self, request_id: RequestId, reason: str, related_request_id: RequestId | None) -> None:
-        # Thread `related_request_id` so streamable-HTTP routes the cancel onto
+        # Thread `related_request_id` so streamable HTTP routes the cancel onto
-        # the same per-request SSE stream as the request it cancels; without it
+        # the request's own SSE stream instead of a possibly-absent GET stream.
-        # the notification falls through to the standalone GET stream and is
+        # `notify` swallows connection-state errors itself, so no guard here.
-        # dropped when no GET stream is open.
+        await self.notify(
-        try:
+            "notifications/cancelled",
-            await self.notify(
+            {"requestId": request_id, "reason": reason},
-                "notifications/cancelled",
+            _related_request_id=related_request_id,
-                {"requestId": request_id, "reason": reason},
+        )
-                _related_request_id=related_request_id,
-            )
-        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
-            pass

src/mcp/shared/session.py

@@ -1,487 +1,21 @@
-from __future__ import annotations
+"""Compatibility names that outlived the removed v1 session layer (`BaseSession`)."""
 
-import contextvars
+from typing import Generic, TypeVar
-import logging
-from contextlib import AsyncExitStack
-from types import TracebackType
-from typing import Any, Generic, Protocol, TypeVar
 
-import anyio
+from mcp.shared.dispatcher import ProgressFnT as ProgressFnT
-from anyio.streams.memory import MemoryObjectSendStream
+from mcp.shared.message import MessageMetadata
-from opentelemetry.trace import SpanKind
+from mcp.types import RequestParamsMeta
-from pydantic import BaseModel, TypeAdapter
-from typing_extensions import Self
-
-from mcp.shared._compat import resync_tracer
-from mcp.shared._otel import inject_trace_context, otel_span
-from mcp.shared._stream_protocols import ReadStream, WriteStream
-from mcp.shared.exceptions import MCPError
-from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
-from mcp.types import (
-    CONNECTION_CLOSED,
-    INVALID_PARAMS,
-    METHOD_NOT_FOUND,
-    REQUEST_TIMEOUT,
-    CancelledNotification,
-    ClientNotification,
-    ClientRequest,
-    ClientResult,
-    ErrorData,
-    JSONRPCError,
-    JSONRPCNotification,
-    JSONRPCRequest,
-    JSONRPCResponse,
-    ProgressNotification,
-    ProgressToken,
-    RequestParamsMeta,
-    ServerNotification,
-    ServerRequest,
-    ServerResult,
-)
-
-SendRequestT = TypeVar("SendRequestT", ClientRequest, ServerRequest)
-SendResultT = TypeVar("SendResultT", ClientResult, ServerResult)
-SendNotificationT = TypeVar("SendNotificationT", ClientNotification, ServerNotification)
-ReceiveRequestT = TypeVar("ReceiveRequestT", ClientRequest, ServerRequest)
-ReceiveResultT = TypeVar("ReceiveResultT", bound=BaseModel)
-ReceiveNotificationT = TypeVar("ReceiveNotificationT", ClientNotification, ServerNotification)
 
 RequestId = str | int
 
-
+ReceiveRequestT = TypeVar("ReceiveRequestT")
-class ProgressFnT(Protocol):
+SendResultT = TypeVar("SendResultT")
-    """Protocol for progress notification callbacks."""
-
-    async def __call__(
-        self, progress: float, total: float | None, message: str | None
-    ) -> None: ...  # pragma: no branch
 
 
 class RequestResponder(Generic[ReceiveRequestT, SendResultT]):
-    """Handles responding to MCP requests and manages request lifecycle.
+    """Typing stub for the v1 responder; the SDK never instantiates it."""
 
-    This class MUST be used as a context manager to ensure proper cleanup and
+    request_id: RequestId
-    cancellation handling:
+    request_meta: RequestParamsMeta | None
-
+    request: ReceiveRequestT
-    Example:
+    message_metadata: MessageMetadata
-        ```python
-        with request_responder as resp:
-            await resp.respond(result)
-        ```
-
-    The context manager ensures:
-    1. Proper cancellation scope setup and cleanup
-    2. Request completion tracking
-    3. Cleanup of in-flight requests
-    """
-
-    def __init__(
-        self,
-        request_id: RequestId,
-        request_meta: RequestParamsMeta | None,
-        request: ReceiveRequestT,
-        session: BaseSession[SendRequestT, SendNotificationT, SendResultT, ReceiveRequestT, ReceiveNotificationT],
-        message_metadata: MessageMetadata = None,
-        context: contextvars.Context | None = None,
-    ) -> None:
-        self.request_id = request_id
-        self.request_meta = request_meta
-        self.request = request
-        self.message_metadata = message_metadata
-        self.context = context
-        self._session = session
-        self._completed = False
-        self._entered = False  # Track if we're in a context manager
-
-    def __enter__(self) -> RequestResponder[ReceiveRequestT, SendResultT]:
-        self._entered = True
-        return self
-
-    def __exit__(
-        self,
-        exc_type: type[BaseException] | None,
-        exc_val: BaseException | None,
-        exc_tb: TracebackType | None,
-    ) -> None:
-        self._entered = False
-
-    async def respond(self, response: SendResultT | ErrorData) -> None:
-        """Send a response for this request.
-
-        Must be called within a context manager block.
-
-        Raises:
-            RuntimeError: If not used within a context manager
-            AssertionError: If request was already responded to
-        """
-        if not self._entered:  # pragma: no cover
-            raise RuntimeError("RequestResponder must be used as a context manager")
-        assert not self._completed, "Request already responded to"
-        self._completed = True
-        await self._session._send_response(  # type: ignore[reportPrivateUsage]
-            request_id=self.request_id, response=response
-        )
-
-
-class BaseSession(
-    Generic[
-        SendRequestT,
-        SendNotificationT,
-        SendResultT,
-        ReceiveRequestT,
-        ReceiveNotificationT,
-    ],
-):
-    """Implements an MCP "session" on top of read/write streams, including features
-    like request/response linking, notifications, and progress.
-
-    This class is an async context manager that automatically starts processing
-    messages when entered.
-    """
-
-    _response_streams: dict[RequestId, MemoryObjectSendStream[JSONRPCResponse | JSONRPCError]]
-    _request_id: int
-    _progress_callbacks: dict[RequestId, ProgressFnT]
-
-    def __init__(
-        self,
-        read_stream: ReadStream[SessionMessage | Exception],
-        write_stream: WriteStream[SessionMessage],
-        # If none, reading will never time out
-        read_timeout_seconds: float | None = None,
-    ) -> None:
-        self._read_stream = read_stream
-        self._write_stream = write_stream
-        self._response_streams = {}
-        self._request_id = 0
-        self._session_read_timeout_seconds = read_timeout_seconds
-        self._progress_callbacks = {}
-        self._exit_stack = AsyncExitStack()
-
-    async def __aenter__(self) -> Self:
-        self._task_group = anyio.create_task_group()
-        await self._task_group.__aenter__()
-        self._task_group.start_soon(self._receive_loop)
-        return self
-
-    async def __aexit__(
-        self,
-        exc_type: type[BaseException] | None,
-        exc_val: BaseException | None,
-        exc_tb: TracebackType | None,
-    ) -> bool | None:
-        await self._exit_stack.aclose()
-        # Using BaseSession as a context manager should not block on exit (this
-        # would be very surprising behavior), so make sure to cancel the tasks
-        # in the task group.
-        self._task_group.cancel_scope.cancel()
-        result = await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
-        await resync_tracer()
-        return result
-
-    async def send_request(
-        self,
-        request: SendRequestT,
-        result_type: type[ReceiveResultT],
-        request_read_timeout_seconds: float | None = None,
-        metadata: MessageMetadata = None,
-        progress_callback: ProgressFnT | None = None,
-    ) -> ReceiveResultT:
-        """Sends a request and waits for a response.
-
-        Raises an MCPError if the response contains an error. If a request read timeout is provided, it will take
-        precedence over the session read timeout.
-
-        Do not use this method to emit notifications! Use send_notification() instead.
-        """
-        request_id = self._request_id
-        self._request_id = request_id + 1
-
-        response_stream, response_stream_reader = anyio.create_memory_object_stream[JSONRPCResponse | JSONRPCError](1)
-        self._response_streams[request_id] = response_stream
-
-        # Set up progress token if progress callback is provided
-        request_data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
-        if progress_callback is not None:
-            # Use request_id as progress token
-            if "params" not in request_data:  # pragma: lax no cover
-                request_data["params"] = {}
-            if "_meta" not in request_data["params"]:  # pragma: lax no cover
-                request_data["params"]["_meta"] = {}
-            request_data["params"]["_meta"]["progressToken"] = request_id
-            # Store the callback for this request
-            self._progress_callbacks[request_id] = progress_callback
-
-        try:
-            target = request_data.get("params", {}).get("name")
-            span_name = f"MCP send {request.method} {target}" if target else f"MCP send {request.method}"
-
-            with otel_span(
-                span_name,
-                kind=SpanKind.CLIENT,
-                attributes={"mcp.method.name": request.method, "jsonrpc.request.id": str(request_id)},
-            ):
-                # Inject W3C trace context into _meta (SEP-414).
-                meta: dict[str, Any] = request_data.setdefault("params", {}).setdefault("_meta", {})
-                inject_trace_context(meta)
-
-                jsonrpc_request = JSONRPCRequest(jsonrpc="2.0", id=request_id, **request_data)
-                await self._write_stream.send(SessionMessage(message=jsonrpc_request, metadata=metadata))
-
-                # request read timeout takes precedence over session read timeout
-                timeout = request_read_timeout_seconds or self._session_read_timeout_seconds
-
-                try:
-                    with anyio.fail_after(timeout):
-                        response_or_error = await response_stream_reader.receive()
-                except TimeoutError:
-                    class_name = request.__class__.__name__
-                    message = f"Timed out while waiting for response to {class_name}. Waited {timeout} seconds."
-                    raise MCPError(code=REQUEST_TIMEOUT, message=message)
-
-                if isinstance(response_or_error, JSONRPCError):
-                    raise MCPError.from_jsonrpc_error(response_or_error)
-                else:
-                    return result_type.model_validate(response_or_error.result, by_name=False)
-
-        finally:
-            self._response_streams.pop(request_id, None)
-            self._progress_callbacks.pop(request_id, None)
-            await response_stream.aclose()
-            await response_stream_reader.aclose()
-
-    async def send_notification(
-        self,
-        notification: SendNotificationT,
-        related_request_id: RequestId | None = None,
-    ) -> None:
-        """Emits a notification, which is a one-way message that does not expect a response."""
-        # Some transport implementations may need to set the related_request_id
-        # to attribute to the notifications to the request that triggered them.
-        jsonrpc_notification = JSONRPCNotification(
-            jsonrpc="2.0",
-            **notification.model_dump(by_alias=True, mode="json", exclude_none=True),
-        )
-        session_message = SessionMessage(
-            message=jsonrpc_notification,
-            metadata=ServerMessageMetadata(related_request_id=related_request_id) if related_request_id else None,
-        )
-        await self._write_stream.send(session_message)
-
-    async def _send_response(self, request_id: RequestId, response: SendResultT | ErrorData) -> None:
-        if isinstance(response, ErrorData):
-            jsonrpc_error = JSONRPCError(jsonrpc="2.0", id=request_id, error=response)
-            session_message = SessionMessage(message=jsonrpc_error)
-            await self._write_stream.send(session_message)
-        else:
-            jsonrpc_response = JSONRPCResponse(
-                jsonrpc="2.0",
-                id=request_id,
-                result=response.model_dump(by_alias=True, mode="json", exclude_none=True),
-            )
-            session_message = SessionMessage(message=jsonrpc_response)
-            await self._write_stream.send(session_message)
-
-    @property
-    def _receive_request_adapter(self) -> TypeAdapter[ReceiveRequestT]:
-        """Each subclass must provide its own request adapter."""
-        raise NotImplementedError
-
-    @property
-    def _receive_request_methods(self) -> frozenset[str]:
-        """Method names in the receive-request union; anything else is
-        answered with METHOD_NOT_FOUND before validation is attempted."""
-        raise NotImplementedError
-
-    @property
-    def _receive_notification_adapter(self) -> TypeAdapter[ReceiveNotificationT]:
-        raise NotImplementedError
-
-    async def _receive_loop(self) -> None:
-        async with self._read_stream, self._write_stream:
-            try:
-
-                async def _handle_session_message(message: SessionMessage) -> None:
-                    sender_context: contextvars.Context | None = getattr(self._read_stream, "last_context", None)
-                    if isinstance(message.message, JSONRPCRequest):
-                        if message.message.method not in self._receive_request_methods:
-                            # Unknown methods are METHOD_NOT_FOUND (-32601) per
-                            # JSON-RPC 2.0, not validation failures (-32602).
-                            error_response = JSONRPCError(
-                                jsonrpc="2.0",
-                                id=message.message.id,
-                                error=ErrorData(
-                                    code=METHOD_NOT_FOUND, message="Method not found", data=message.message.method
-                                ),
-                            )
-                            await self._write_stream.send(SessionMessage(message=error_response))
-                            return
-                        try:
-                            validated_request = self._receive_request_adapter.validate_python(
-                                message.message.model_dump(by_alias=True, mode="json", exclude_none=True),
-                                by_name=False,
-                            )
-                            responder = RequestResponder(
-                                request_id=message.message.id,
-                                request_meta=validated_request.params.meta if validated_request.params else None,
-                                request=validated_request,
-                                session=self,
-                                message_metadata=message.metadata,
-                                context=sender_context,
-                            )
-                            await self._received_request(responder)
-                        except Exception:
-                            # For request validation errors, send a proper JSON-RPC error
-                            # response instead of crashing the server
-                            logging.warning("Failed to validate request", exc_info=True)
-                            logging.debug(f"Message that failed validation: {message.message}")
-                            error_response = JSONRPCError(
-                                jsonrpc="2.0",
-                                id=message.message.id,
-                                error=ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data=""),
-                            )
-                            session_message = SessionMessage(message=error_response)
-                            await self._write_stream.send(session_message)
-
-                    elif isinstance(message.message, JSONRPCNotification):
-                        try:
-                            notification = self._receive_notification_adapter.validate_python(
-                                message.message.model_dump(by_alias=True, mode="json", exclude_none=True),
-                                by_name=False,
-                            )
-                            if isinstance(notification, CancelledNotification):
-                                # ClientSession runs server-initiated requests
-                                # inline in this loop, so by the time a peer
-                                # cancellation is read there is nothing left to
-                                # cancel. Consume it here so message_handler
-                                # keeps the contract it had before the
-                                # dispatcher swap removed _in_flight.
-                                return
-                            # Handle progress notifications callback
-                            if isinstance(notification, ProgressNotification):
-                                progress_token = notification.params.progress_token
-                                # If there is a progress callback for this token,
-                                # call it with the progress information
-                                if progress_token in self._progress_callbacks:
-                                    callback = self._progress_callbacks[progress_token]
-                                    try:
-                                        await callback(
-                                            notification.params.progress,
-                                            notification.params.total,
-                                            notification.params.message,
-                                        )
-                                    except Exception:
-                                        logging.exception("Progress callback raised an exception")
-                            await self._received_notification(notification)
-                            await self._handle_incoming(notification)
-                        except Exception:
-                            # For other validation errors, log and continue
-                            logging.warning(
-                                "Failed to validate notification: %s",
-                                message.message,
-                                exc_info=True,
-                            )
-                    else:  # Response or error
-                        await self._handle_response(message)
-
-                async for message in self._read_stream:
-                    if isinstance(message, Exception):
-                        await self._handle_incoming(message)
-                        continue
-
-                    await _handle_session_message(message)
-
-            except anyio.ClosedResourceError:
-                # This is expected when the client disconnects abruptly.
-                # Without this handler, the exception would propagate up and
-                # crash the server's task group.
-                logging.debug("Read stream closed by client")
-            except Exception as e:
-                # Other exceptions are not expected and should be logged. We purposefully
-                # catch all exceptions here to avoid crashing the server.
-                logging.exception(f"Unhandled exception in receive loop: {e}")  # pragma: no cover
-            finally:
-                # after the read stream is closed, we need to send errors
-                # to any pending requests
-                # Snapshot: stream.send() wakes the waiter, whose finally pops
-                # from _response_streams before the next __next__() call.
-                for id, stream in list(self._response_streams.items()):
-                    error = ErrorData(code=CONNECTION_CLOSED, message="Connection closed")
-                    try:
-                        await stream.send(JSONRPCError(jsonrpc="2.0", id=id, error=error))
-                        await stream.aclose()
-                    except Exception:  # pragma: lax no cover
-                        # Stream might already be closed
-                        pass
-                self._response_streams.clear()
-
-    def _normalize_request_id(self, response_id: RequestId) -> RequestId:
-        """Normalize a response ID to match how request IDs are stored.
-
-        Since the client always sends integer IDs, we normalize string IDs
-        to integers when possible. This matches the TypeScript SDK approach:
-        https://github.com/modelcontextprotocol/typescript-sdk/blob/a606fb17909ea454e83aab14c73f14ea45c04448/src/shared/protocol.ts#L861
-
-        Args:
-            response_id: The response ID from the incoming message.
-
-        Returns:
-            The normalized ID (int if possible, otherwise original value).
-        """
-        if isinstance(response_id, str):
-            try:
-                return int(response_id)
-            except ValueError:
-                logging.warning(f"Response ID {response_id!r} cannot be normalized to match pending requests")
-        return response_id
-
-    async def _handle_response(self, message: SessionMessage) -> None:
-        """Handle an incoming response or error message."""
-        # This check is always true at runtime: the caller (_receive_loop) only invokes
-        # this method in the else branch after checking for JSONRPCRequest and
-        # JSONRPCNotification. However, the type checker can't infer this from the
-        # method signature, so we need this guard for type narrowing.
-        if not isinstance(message.message, JSONRPCResponse | JSONRPCError):
-            return  # pragma: no cover
-
-        if message.message.id is None:
-            # Narrows to JSONRPCError since JSONRPCResponse.id is always RequestId
-            error = message.message.error
-            logging.warning(f"Received error with null ID: {error.message}")
-            await self._handle_incoming(MCPError(error.code, error.message, error.data))
-            return
-        # Normalize response ID to handle type mismatches (e.g., "0" vs 0)
-        response_id = self._normalize_request_id(message.message.id)
-
-        stream = self._response_streams.pop(response_id, None)
-        if stream:
-            await stream.send(message.message)
-        else:
-            await self._handle_incoming(RuntimeError(f"Received response with an unknown request ID: {message}"))
-
-    async def _received_request(self, responder: RequestResponder[ReceiveRequestT, SendResultT]) -> None:
-        """Can be overridden by subclasses to handle a request without needing to
-        listen on the message stream.
-
-        If the request is responded to within this method, it will not be
-        forwarded on to the message stream.
-        """
-
-    async def _received_notification(self, notification: ReceiveNotificationT) -> None:
-        """Can be overridden by subclasses to handle a notification without needing
-        to listen on the message stream.
-        """
-
-    async def send_progress_notification(
-        self,
-        progress_token: ProgressToken,
-        progress: float,
-        total: float | None = None,
-        message: str | None = None,
-    ) -> None:
-        """Sends a progress notification for a request that is currently being processed."""
-
-    async def _handle_incoming(
-        self, req: RequestResponder[ReceiveRequestT, SendResultT] | ReceiveNotificationT | Exception
-    ) -> None:
-        """A generic handler for incoming messages. Overridden by subclasses."""

src/mcp/types/__init__.py

@@ -152,7 +152,6 @@ from mcp.types.jsonrpc import (
     INVALID_REQUEST,
     METHOD_NOT_FOUND,
     PARSE_ERROR,
-    REQUEST_CANCELLED,
     REQUEST_TIMEOUT,
     URL_ELICITATION_REQUIRED,
     ErrorData,
@@ -320,7 +319,6 @@ __all__ = [
     "INVALID_REQUEST",
     "METHOD_NOT_FOUND",
     "PARSE_ERROR",
-    "REQUEST_CANCELLED",
     "REQUEST_TIMEOUT",
     "URL_ELICITATION_REQUIRED",
     "ErrorData",

src/mcp/types/jsonrpc.py

@@ -43,7 +43,6 @@ URL_ELICITATION_REQUIRED = -32042
 # SDK error codes
 CONNECTION_CLOSED = -32000
 REQUEST_TIMEOUT = -32001
-REQUEST_CANCELLED = -32002
 
 # Standard JSON-RPC error codes
 PARSE_ERROR = -32700

tests/client/test_list_roots_callback.py

@@ -2,9 +2,8 @@ import pytest
 from pydantic import FileUrl
 
 from mcp import Client
-from mcp.client.session import ClientSession
+from mcp.client import ClientRequestContext
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.shared._context import RequestContext
 from mcp.types import ListRootsResult, Root, TextContent
 
 
@@ -20,7 +19,7 @@ async def test_list_roots_callback():
     )
 
     async def list_roots_callback(
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
     ) -> ListRootsResult:
         return callback_return
 

tests/client/test_resource_cleanup.py

@@ -1,66 +0,0 @@
-from typing import Any
-from unittest.mock import patch
-
-import anyio
-import pytest
-from pydantic import TypeAdapter
-
-from mcp.shared.message import SessionMessage
-from mcp.shared.session import BaseSession, RequestId, SendResultT
-from mcp.types import ClientNotification, ClientRequest, ClientResult, EmptyResult, ErrorData, PingRequest
-
-
-@pytest.mark.anyio
-async def test_send_request_stream_cleanup():
-    """Test that send_request properly cleans up streams when an exception occurs.
-
-    This test mocks out most of the session functionality to focus on stream cleanup.
-    """
-
-    # Create a mock session with the minimal required functionality
-    class TestSession(BaseSession[ClientRequest, ClientNotification, ClientResult, Any, Any]):
-        async def _send_response(
-            self, request_id: RequestId, response: SendResultT | ErrorData
-        ) -> None:  # pragma: no cover
-            pass
-
-        @property
-        def _receive_request_adapter(self) -> TypeAdapter[Any]:
-            return TypeAdapter(object)  # pragma: no cover
-
-        @property
-        def _receive_notification_adapter(self) -> TypeAdapter[Any]:
-            return TypeAdapter(object)  # pragma: no cover
-
-    # Create streams
-    write_stream_send, write_stream_receive = anyio.create_memory_object_stream[SessionMessage](1)
-    read_stream_send, read_stream_receive = anyio.create_memory_object_stream[SessionMessage](1)
-
-    # Create the session
-    session = TestSession(read_stream_receive, write_stream_send)
-
-    # Create a test request
-    request = PingRequest()
-
-    # Patch the _write_stream.send method to raise an exception
-    async def mock_send(*args: Any, **kwargs: Any):
-        raise RuntimeError("Simulated network error")
-
-    # Record the response streams before the test
-    initial_stream_count = len(session._response_streams)
-
-    # Run the test with the patched method
-    with patch.object(session._write_stream, "send", mock_send):
-        with pytest.raises(RuntimeError):
-            await session.send_request(request, EmptyResult)
-
-    # Verify that no response streams were leaked
-    assert len(session._response_streams) == initial_stream_count, (
-        f"Expected {initial_stream_count} response streams after request, but found {len(session._response_streams)}"
-    )
-
-    # Clean up
-    await write_stream_send.aclose()
-    await write_stream_receive.aclose()
-    await read_stream_send.aclose()
-    await read_stream_receive.aclose()

tests/client/test_sampling_callback.py

@@ -1,9 +1,8 @@
 import pytest
 
 from mcp import Client
-from mcp.client.session import ClientSession
+from mcp.client import ClientRequestContext
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.shared._context import RequestContext
 from mcp.types import (
     CreateMessageRequestParams,
     CreateMessageResult,
@@ -26,7 +25,7 @@ async def test_sampling_callback():
     )
 
     async def sampling_callback(
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: CreateMessageRequestParams,
     ) -> CreateMessageResult:
         return callback_return
@@ -71,7 +70,7 @@ async def test_create_message_backwards_compat_single_content():
     )
 
     async def sampling_callback(
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: CreateMessageRequestParams,
     ) -> CreateMessageResult:
         return callback_return

tests/client/test_session.py

@@ -1,22 +1,30 @@
 from __future__ import annotations
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Mapping
-from contextlib import asynccontextmanager
+from contextlib import AsyncExitStack, asynccontextmanager
 from typing import Any
 
 import anyio
+import anyio.abc
 import anyio.streams.memory
 import pytest
+from pydantic import FileUrl
 
-from mcp import types
+from mcp import MCPError, types
+from mcp.client import ClientRequestContext
 from mcp.client.session import DEFAULT_CLIENT_INFO, ClientSession
-from mcp.shared._context import RequestContext
+from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
+from mcp.shared.dispatcher import CallOptions, DispatchContext, OnNotify, OnRequest
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import RequestResponder
+from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
 from mcp.types import (
+    CONNECTION_CLOSED,
     INVALID_PARAMS,
     LATEST_PROTOCOL_VERSION,
+    METHOD_NOT_FOUND,
+    REQUEST_TIMEOUT,
     CallToolResult,
     Implementation,
     InitializedNotification,
@@ -420,7 +428,7 @@ async def test_client_capabilities_with_custom_callbacks():
     received_capabilities = None
 
     async def custom_sampling_callback(  # pragma: no cover
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.ErrorData:
         return types.CreateMessageResult(
@@ -430,7 +438,7 @@ async def test_client_capabilities_with_custom_callbacks():
         )
 
     async def custom_list_roots_callback(  # pragma: no cover
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
     ) -> types.ListRootsResult | types.ErrorData:
         return types.ListRootsResult(roots=[])
 
@@ -504,7 +512,7 @@ async def test_client_capabilities_with_sampling_tools():
     received_capabilities = None
 
     async def custom_sampling_callback(  # pragma: no cover
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.ErrorData:
         return types.CreateMessageResult(
@@ -751,8 +759,32 @@ async def test_receive_loop_answers_malformed_inbound_request_with_invalid_param
 
 
 @pytest.mark.anyio
-async def test_receive_loop_answers_invalid_params_when_sampling_callback_raises():
+async def test_receive_loop_answers_unknown_request_method_with_method_not_found():
-    """Same boundary catches exceptions from the request handler itself."""
+    """An unknown request method is answered with METHOD_NOT_FOUND, not INVALID_PARAMS (spec-mandated)."""
+    async with raw_client_session() as (_session, to_client, from_client):
+        await to_client.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=7, method="x/unknown")))
+        out = await from_client.receive()
+    assert isinstance(out.message, JSONRPCError)
+    assert out.message.id == 7
+    assert out.message.error == types.ErrorData(code=METHOD_NOT_FOUND, message="Method not found", data="x/unknown")
+
+
+@pytest.mark.anyio
+async def test_receive_loop_drops_unknown_notification_method_without_response():
+    """An unknown notification method is dropped silently: JSON-RPC forbids responses to notifications."""
+    async with raw_client_session() as (_session, to_client, from_client):
+        await to_client.send(SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="x/unknown")))
+        # The answered follow-up ping proves no response was emitted and the loop survived.
+        await to_client.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")))
+        out = await from_client.receive()
+    assert isinstance(out.message, JSONRPCResponse)
+    assert out.message.id == 1
+
+
+@pytest.mark.anyio
+async def test_raising_sampling_callback_answers_with_code_zero():
+    """A raising sampling callback is answered with code 0 and `str(exc)` (SDK-defined).
+    Raw streams because the assertion is the outbound `JSONRPCError` envelope itself."""
 
     async def boom(ctx: object, params: object) -> types.CreateMessageResult:
         raise RuntimeError("sampling boom")
@@ -767,12 +799,13 @@ async def test_receive_loop_answers_invalid_params_when_sampling_callback_raises
         )
         out = await from_client.receive()
     assert isinstance(out.message, JSONRPCError)
-    assert out.message.error.code == INVALID_PARAMS
+    assert out.message.error == types.ErrorData(code=0, message="sampling boom")
 
 
 @pytest.mark.anyio
 async def test_receive_loop_logs_and_drops_malformed_notification(caplog: pytest.LogCaptureFixture):
-    """A notification that fails ServerNotification validation is logged and dropped."""
+    """A malformed notification is logged and dropped without reaching `message_handler` (SDK-defined).
+    Scripted peer: the typed API cannot emit a method outside the spec's notification union."""
     seen: list[object] = []
     delivered = anyio.Event()
 
@@ -792,19 +825,54 @@ async def test_receive_loop_logs_and_drops_malformed_notification(caplog: pytest
 
 
 @pytest.mark.anyio
-async def test_receive_loop_forwards_transport_exception_to_message_handler():
+async def test_raising_message_handler_on_transport_exception_costs_the_delivery_not_the_connection(
+    caplog: pytest.LogCaptureFixture,
+):
+    """A `message_handler` that raises on a transport-level `Exception` item is contained: the
+    failure is logged and the receive loop keeps serving (SDK-defined). Raw streams because
+    only a transport can put an `Exception` item on the read stream."""
     seen: list[object] = []
     delivered = anyio.Event()
 
     async def handler(msg: object) -> None:
         seen.append(msg)
         delivered.set()
+        # No checkpoint between set() and the containment log, so after wait() the log entry exists.
+        raise RuntimeError("handler boom")
 
-    async with raw_client_session(message_handler=handler) as (_session, to_client, _):
+    async with raw_client_session(message_handler=handler) as (_session, to_client, from_client):
         exc = ValueError("bad bytes")
         await to_client.send(exc)
         await delivered.wait()
+        await to_client.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=9, method="ping")))
+        out = await from_client.receive()
     assert seen == [exc]
+    assert isinstance(out.message, JSONRPCResponse)
+    assert out.message.id == 9
+    assert "message_handler raised on transport exception" in caplog.text
+
+
+@pytest.mark.anyio
+async def test_message_handler_awaiting_session_traffic_on_transport_exception_completes():
+    """A `message_handler` that awaits session traffic on a transport `Exception` item completes:
+    fault deliveries are spawned into the task group, not run inline in the read loop (SDK-defined).
+    Raw streams because only a transport can put an `Exception` item on the read stream."""
+    ponged = anyio.Event()
+
+    # `session` resolves at call time, after the `as` clause binds it.
+    async def handler(msg: object) -> None:
+        assert isinstance(msg, Exception)
+        await session.send_ping()
+        ponged.set()
+
+    async with raw_client_session(message_handler=handler) as (session, to_client, from_client):
+        await to_client.send(ValueError("bad bytes"))
+        # Serve the handler's ping like a transport would; inline delivery would deadlock here.
+        out = await from_client.receive()
+        assert isinstance(out.message, JSONRPCRequest)
+        assert out.message.method == "ping"
+        await to_client.send(SessionMessage(JSONRPCResponse(jsonrpc="2.0", id=out.message.id, result={})))
+        await ponged.wait()
 
 
 @pytest.mark.anyio
@@ -814,6 +882,7 @@ async def test_receive_loop_consumes_server_cancelled_without_reaching_message_h
     The server dispatcher now emits this on sampling/elicitation timeout, but
     ClientSession has no in-flight tracking to act on it, so surfacing it would
     only break user handlers that exhaustively match ServerNotification.
+    Scripted peer: the typed server API cannot emit a bare `notifications/cancelled`.
     """
     seen: list[object] = []
     delivered = anyio.Event()
@@ -841,23 +910,344 @@ async def test_receive_loop_consumes_server_cancelled_without_reaching_message_h
 
 
 @pytest.mark.anyio
-async def test_receive_loop_swallows_progress_callback_exception(caplog: pytest.LogCaptureFixture):
+async def test_request_timeout_zero_overrides_session_timeout():
+    """`request_read_timeout_seconds=0` is a real per-request timeout (fail at the
+    first checkpoint, `anyio.fail_after(0)` semantics), not a fall-through to the
+    session-level timeout. The request is never answered, so falling back to the
+    30s session timeout would trip the harness's 5s guard instead."""
+    async with raw_client_session(read_timeout_seconds=30) as (session, _to_client, _from_client):
+        with pytest.raises(MCPError) as exc_info:
+            await session.send_request(types.PingRequest(), types.EmptyResult, request_read_timeout_seconds=0.0)
+    assert exc_info.value.error.code == REQUEST_TIMEOUT
+
+
+@pytest.mark.anyio
+async def test_progress_notification_reaches_request_callback_and_message_handler():
+    """A `notifications/progress` for an in-flight request reaches both the `progress_callback` and
+    `message_handler` (SDK-defined). Scripted peer: the progress token must echo the wire request id."""
+    updates: list[tuple[float, float | None, str | None]] = []
+    teed: list[types.ProgressNotification] = []
+    request_id: types.RequestId | None = None
+    progressed = anyio.Event()
     delivered = anyio.Event()
 
-    async def boom(progress: float, total: float | None, message: str | None) -> None:
+    async def on_progress(progress: float, total: float | None, message: str | None) -> None:
-        raise RuntimeError("progress boom")
+        updates.append((progress, total, message))
+        progressed.set()
 
     async def handler(msg: object) -> None:
+        # Only the progress notification is teed to the message handler here.
+        assert isinstance(msg, types.ProgressNotification)
+        teed.append(msg)
         delivered.set()
 
-    async with raw_client_session(message_handler=handler) as (session, to_client, _):
+    async with raw_client_session(message_handler=handler) as (session, to_client, from_client):
-        # Register the callback under a known token without sending a request.
+        async with anyio.create_task_group() as tg:
-        session._progress_callbacks[42] = boom  # pyright: ignore[reportPrivateUsage]
+
-        params = {"progressToken": 42, "progress": 0.5}
+            async def call() -> None:
-        await to_client.send(
+                await session.send_request(types.PingRequest(), types.EmptyResult, progress_callback=on_progress)
-            SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/progress", params=params))
+
-        )
+            tg.start_soon(call)
-        # The progress notification also reaches the message handler after the
+            request = await from_client.receive()
-        # callback runs, so this fires once the callback's exception is handled.
+            assert isinstance(request.message, JSONRPCRequest)
-        await delivered.wait()
+            request_id = request.message.id
-    assert "Progress callback raised an exception" in caplog.text
+            # The request id doubles as the progress token.
+            params = {"progressToken": request_id, "progress": 0.5, "total": 1.0, "message": "halfway"}
+            await to_client.send(
+                SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/progress", params=params))
+            )
+            await progressed.wait()
+            await delivered.wait()
+            await to_client.send(SessionMessage(JSONRPCResponse(jsonrpc="2.0", id=request_id, result={})))
+    assert updates == [(0.5, 1.0, "halfway")]
+    assert request_id is not None
+    assert len(teed) == 1
+    assert teed[0].params == types.ProgressNotificationParams(
+        progress_token=request_id, progress=0.5, total=1.0, message="halfway"
+    )
+
+
+@pytest.mark.anyio
+async def test_dispatcher_keyword_runs_over_direct_dispatch():
+    """A session built with dispatcher= works without a stream pair (in-process embedding)."""
+    client_side, server_side = create_direct_dispatcher_pair()
+
+    async def server_on_request(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> dict[str, object]:
+        assert method == "ping"
+        return {}
+
+    notified: list[str] = []
+
+    async def server_on_notify(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> None:
+        notified.append(method)
+
+    session = ClientSession(dispatcher=client_side)
+    results: list[types.EmptyResult] = []
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg:
+            await tg.start(server_side.run, server_on_request, server_on_notify)
+            async with session:
+                results.append(await session.send_ping(meta=None))
+                # Server-to-client: direct dispatch delivers ping with no params member (no _meta injection).
+                assert await server_side.send_raw_request("ping", None) == {}
+                await session.send_notification(types.RootsListChangedNotification())
+            server_side.close()
+    assert results == [types.EmptyResult()]
+    assert notified == ["notifications/roots/list_changed"]
+
+
+@pytest.mark.anyio
+async def test_direct_dispatch_roots_list_reaches_callback_with_synthesized_request_id():
+    """A server-initiated roots/list over dispatcher= reaches the registered callback and round-trips
+    the result; the callback context carries an int request_id (SDK-defined: DirectDispatcher
+    synthesizes ids)."""
+    client_side, server_side = create_direct_dispatcher_pair()
+    contexts: list[ClientRequestContext] = []
+
+    async def list_roots(context: ClientRequestContext) -> types.ListRootsResult:
+        contexts.append(context)
+        return types.ListRootsResult(roots=[types.Root(uri=FileUrl("file:///workspace"))])
+
+    async def server_on_request(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> dict[str, object]:
+        raise NotImplementedError
+
+    async def server_on_notify(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> None:
+        raise NotImplementedError
+
+    session = ClientSession(dispatcher=client_side, list_roots_callback=list_roots)
+    result: dict[str, Any] | None = None
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg:
+            await tg.start(server_side.run, server_on_request, server_on_notify)
+            async with session:
+                result = await server_side.send_raw_request("roots/list", None)
+            server_side.close()
+    assert result == {"roots": [{"uri": "file:///workspace"}]}
+    assert len(contexts) == 1
+    assert isinstance(contexts[0].request_id, int)
+
+
+@pytest.mark.anyio
+async def test_raising_notification_callbacks_over_direct_dispatch_cost_only_that_delivery(
+    caplog: pytest.LogCaptureFixture,
+):
+    """A raising `logging_callback` or `message_handler` is contained in the session, so the
+    in-process peer's notify() returns normally and the session keeps serving requests
+    (SDK-defined: DirectDispatcher awaits notification handlers inline in the peer's call).
+    A raising `logging_callback` skips the `message_handler` tee for that notification."""
+    client_side, server_side = create_direct_dispatcher_pair()
+    teed: list[types.ServerNotification] = []
+
+    async def logging_callback(params: types.LoggingMessageNotificationParams) -> None:
+        raise ValueError("logging callback boom")
+
+    async def message_handler(
+        message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
+    ) -> None:
+        assert not isinstance(message, RequestResponder | Exception)
+        teed.append(message)
+        raise ValueError("message handler boom")
+
+    async def server_on_request(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> dict[str, object]:
+        assert method == "ping"
+        return {}
+
+    async def server_on_notify(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> None:
+        raise NotImplementedError
+
+    session = ClientSession(dispatcher=client_side, logging_callback=logging_callback, message_handler=message_handler)
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg:
+            await tg.start(server_side.run, server_on_request, server_on_notify)
+            async with session:
+                # logging_callback raises: notify() must return, and message_handler is skipped.
+                await server_side.notify("notifications/message", {"level": "info", "data": "hello"})
+                # message_handler raises: notify() must return.
+                await server_side.notify("notifications/tools/list_changed", None)
+                # The session still serves requests afterwards.
+                assert await session.send_ping() == types.EmptyResult()
+            server_side.close()
+    assert [type(n) for n in teed] == [types.ToolListChangedNotification]
+    assert caplog.text.count("notification callback for") == 2
+    assert "notification callback for 'notifications/message' raised" in caplog.text
+    assert "notification callback for 'notifications/tools/list_changed' raised" in caplog.text
+
+
+@pytest.mark.anyio
+async def test_dispatcher_keyword_send_request_before_enter_raises_runtimeerror():
+    """The documented pre-enter RuntimeError holds for dispatcher= sessions too."""
+    client_side, _server_side = create_direct_dispatcher_pair()
+    session = ClientSession(dispatcher=client_side)
+    with anyio.fail_after(5), pytest.raises(RuntimeError) as exc:
+        await session.send_ping()
+    assert str(exc.value) == "DirectDispatcher.send_raw_request called before run()"
+
+
+@pytest.mark.anyio
+async def test_dispatcher_keyword_send_request_after_exit_raises_connection_closed():
+    """After __aexit__ a dispatcher= session raises MCPError(CONNECTION_CLOSED), matching the JSONRPC path."""
+    client_side, server_side = create_direct_dispatcher_pair()
+
+    async def server_on_request(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> dict[str, object]:
+        assert method == "ping"
+        return {}
+
+    async def server_on_notify(
+        ctx: DispatchContext[TransportContext], method: str, params: dict[str, object] | None
+    ) -> None:
+        raise NotImplementedError
+
+    session = ClientSession(dispatcher=client_side)
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg:
+            await tg.start(server_side.run, server_on_request, server_on_notify)
+            async with session:
+                assert await session.send_ping() == types.EmptyResult()
+            with pytest.raises(MCPError) as exc:
+                await session.send_ping()
+            assert exc.value.error.code == CONNECTION_CLOSED
+            server_side.close()
+
+
+@pytest.mark.anyio
+async def test_dispatcher_keyword_request_timeout_bounds_wait_for_never_run_peer():
+    """request_read_timeout_seconds fires even when the peer dispatcher never started running."""
+    client_side, _server_side = create_direct_dispatcher_pair()
+    session = ClientSession(dispatcher=client_side)
+    with anyio.fail_after(5):
+        async with session:
+            with pytest.raises(MCPError) as exc:
+                await session.send_request(types.PingRequest(), types.EmptyResult, request_read_timeout_seconds=0.01)
+            assert exc.value.error.code == REQUEST_TIMEOUT
+
+
+@pytest.mark.anyio
+async def test_initialize_opts_out_of_cancel_on_abandon_while_other_requests_leave_it_unset():
+    """`send_request` passes `cancel_on_abandon=False` for `initialize` — the spec forbids
+    cancelling it — and leaves the option unset for every other method."""
+
+    class RecordingDispatcher:
+        """Records `send_raw_request` opts and answers with canned results."""
+
+        def __init__(self) -> None:
+            self.calls: list[tuple[str, CallOptions]] = []
+
+        async def run(
+            self,
+            on_request: OnRequest,
+            on_notify: OnNotify,
+            *,
+            task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
+        ) -> None:
+            task_status.started()
+            await anyio.sleep_forever()
+
+        async def send_raw_request(
+            self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None
+        ) -> dict[str, Any]:
+            self.calls.append((method, opts or {}))
+            if method == "initialize":
+                return InitializeResult(
+                    protocol_version=LATEST_PROTOCOL_VERSION,
+                    capabilities=ServerCapabilities(),
+                    server_info=Implementation(name="mock-server", version="0.1.0"),
+                ).model_dump(by_alias=True, mode="json", exclude_none=True)
+            return {}
+
+        async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+            pass
+
+    dispatcher = RecordingDispatcher()
+    with anyio.fail_after(5):
+        async with ClientSession(dispatcher=dispatcher) as session:
+            await session.initialize()
+            await session.send_ping()
+    opts_by_method = dict(dispatcher.calls)
+    assert opts_by_method["initialize"].get("cancel_on_abandon") is False
+    assert "cancel_on_abandon" not in opts_by_method["ping"]
+
+
+def test_constructor_rejects_streams_and_dispatcher_together():
+    client_side, _server_side = create_direct_dispatcher_pair()
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    with pytest.raises(ValueError, match="not both"):
+        ClientSession(s2c_recv, dispatcher=client_side)
+    s2c_send.close()
+    s2c_recv.close()
+
+
+def test_constructor_requires_both_streams_without_dispatcher():
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    with pytest.raises(ValueError, match="read_stream and write_stream are required"):
+        ClientSession(s2c_recv)
+    with pytest.raises(ValueError, match="read_stream and write_stream are required"):
+        ClientSession()
+    s2c_send.close()
+    s2c_recv.close()
+
+
+@pytest.mark.anyio
+async def test_aenter_cancelled_while_dispatcher_starts_unwinds_cleanly():
+    """Cancellation while `__aenter__` waits for the dispatcher to start unwinds the half-entered
+    task group cleanly, not via anyio's "exited non-innermost cancel scope" RuntimeError (SDK-defined)."""
+
+    class NeverStartsDispatcher:
+        """`run()` parks without ever signalling `task_status.started()`."""
+
+        async def run(
+            self,
+            on_request: OnRequest,
+            on_notify: OnNotify,
+            *,
+            task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
+        ) -> None:
+            await anyio.sleep_forever()
+
+        async def send_raw_request(
+            self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None
+        ) -> dict[str, Any]:
+            raise NotImplementedError
+
+        async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+            raise NotImplementedError
+
+    session = ClientSession(dispatcher=NeverStartsDispatcher())
+    async with AsyncExitStack() as stack:
+        # `start()` is parked forever, so the deadline only ends the wait — any duration is non-racy.
+        with anyio.move_on_after(0.01) as scope:
+            await stack.enter_async_context(session)
+    assert scope.cancelled_caught
+    # The failed enter must not leave the session half-entered.
+    assert session._task_group is None
+
+
+@pytest.mark.anyio
+async def test_send_notification_after_close_is_dropped_silently():
+    """Post-close `send_notification` is fire-and-forget: the notification is dropped,
+    not surfaced as a raw transport error (v1 leaked `anyio.ClosedResourceError`)."""
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](4)
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage](4)
+    try:
+        async with ClientSession(s2c_recv, c2s_send) as session:
+            pass
+        with anyio.fail_after(5):
+            await session.send_notification(types.RootsListChangedNotification())
+        with pytest.raises(anyio.EndOfStream):
+            c2s_recv.receive_nowait()  # nothing reached the wire
+    finally:
+        for s in (s2c_send, s2c_recv, c2s_send, c2s_recv):
+            s.close()

tests/client/test_session_concurrency.py

@@ -0,0 +1,141 @@
+"""Concurrency over a single client session: multiple requests in flight at once, in both directions."""
+
+import anyio
+import pytest
+from inline_snapshot import snapshot
+
+from mcp import Client
+from mcp.client import ClientRequestContext
+from mcp.server.mcpserver import Context, MCPServer
+from mcp.types import (
+    CallToolResult,
+    CreateMessageRequestParams,
+    CreateMessageResult,
+    SamplingMessage,
+    TextContent,
+)
+
+pytestmark = pytest.mark.anyio
+
+
+async def test_concurrent_tool_calls_resolve_out_of_order_to_their_own_callers() -> None:
+    """Three tool calls in flight at once on one session each receive their own result, even though
+    the responses come back in the reverse of the order the requests were sent.
+
+    SDK-defined contract: pins the client request machinery's support for concurrent in-flight
+    calls with out-of-order response correlation. Each handler parks on its own release event
+    after signalling it started; a session that serialized requests would never start the later
+    handlers and the test would time out instead.
+    """
+    send_order = ["a", "b", "c"]
+    started = {tag: anyio.Event() for tag in send_order}
+    release = {tag: anyio.Event() for tag in send_order}
+    done = {tag: anyio.Event() for tag in send_order}
+    completion_order: list[str] = []
+    results: dict[str, CallToolResult] = {}
+
+    server = MCPServer("parking")
+
+    @server.tool()
+    async def park(tag: str) -> str:
+        started[tag].set()
+        await release[tag].wait()
+        return f"result:{tag}"
+
+    async with Client(server) as client:
+
+        async def call_and_record(tag: str) -> None:
+            results[tag] = await client.call_tool("park", {"tag": tag})
+            completion_order.append(tag)
+            done[tag].set()
+
+        with anyio.fail_after(5):
+            async with anyio.create_task_group() as task_group:  # pragma: no branch
+                # Waiting for each handler to start before issuing the next call fixes the send
+                # order, and leaves all three parked in flight together once the loop finishes.
+                for tag in send_order:
+                    task_group.start_soon(call_and_record, tag)
+                    await started[tag].wait()
+
+                # Nothing completed yet: all three calls are genuinely concurrent.
+                assert completion_order == []
+
+                # Release in reverse, awaiting each completion so the finish order is forced.
+                for tag in reversed(send_order):
+                    release[tag].set()
+                    await done[tag].wait()
+
+    assert completion_order == ["c", "b", "a"]
+    assert results == snapshot(
+        {
+            "c": CallToolResult(content=[TextContent(text="result:c")], structured_content={"result": "result:c"}),
+            "b": CallToolResult(content=[TextContent(text="result:b")], structured_content={"result": "result:b"}),
+            "a": CallToolResult(content=[TextContent(text="result:a")], structured_content={"result": "result:a"}),
+        }
+    )
+
+
+async def test_overlapping_sampling_requests_are_serviced_concurrently_by_the_client() -> None:
+    """A server tool that fans out two sampling requests at once gets both echoes back: the client
+    runs overlapping inbound `create_message` requests concurrently instead of serializing them in
+    its receive loop.
+
+    Regression pin for https://github.com/modelcontextprotocol/python-sdk/issues/2489 -- v1's
+    `BaseSession` awaited each inbound request handler inline, so the second sampling callback
+    could not start until the first returned; here both rendezvous before either is released.
+    """
+    sampling_started = {"x": anyio.Event(), "y": anyio.Event()}
+    sampling_release = anyio.Event()
+    tool_results: list[CallToolResult] = []
+
+    server = MCPServer("fan_out_server")
+
+    @server.tool()
+    async def fan_out(ctx: Context) -> str:
+        echoes: dict[str, str] = {}
+
+        async def sample(tag: str) -> None:
+            result = await ctx.session.create_message(
+                messages=[SamplingMessage(role="user", content=TextContent(text=tag))],
+                max_tokens=10,
+            )
+            assert isinstance(result.content, TextContent)
+            echoes[tag] = result.content.text
+
+        async with anyio.create_task_group() as sampler_group:
+            sampler_group.start_soon(sample, "x")
+            sampler_group.start_soon(sample, "y")
+        return f"{echoes['x']} {echoes['y']}"
+
+    async def sampling_callback(
+        context: ClientRequestContext, params: CreateMessageRequestParams
+    ) -> CreateMessageResult:
+        content = params.messages[0].content
+        assert isinstance(content, TextContent)
+        sampling_started[content.text].set()
+        await sampling_release.wait()
+        return CreateMessageResult(
+            role="assistant",
+            content=TextContent(text=f"echo:{content.text}"),
+            model="test-model",
+            stop_reason="endTurn",
+        )
+
+    async with Client(server, sampling_callback=sampling_callback) as client:
+        with anyio.fail_after(5):
+            async with anyio.create_task_group() as task_group:  # pragma: no branch
+
+                async def invoke_fan_out() -> None:
+                    tool_results.append(await client.call_tool("fan_out", {}))
+
+                task_group.start_soon(invoke_fan_out)
+
+                # Both sampling callbacks are mid-flight before either may answer -- a client that
+                # serialized inbound requests would never start the second one.
+                await sampling_started["x"].wait()
+                await sampling_started["y"].wait()
+                sampling_release.set()
+
+    assert tool_results == snapshot(
+        [CallToolResult(content=[TextContent(text="echo:x echo:y")], structured_content={"result": "echo:x echo:y"})]
+    )

tests/interaction/README.md

@@ -193,11 +193,13 @@ many requirements at once; if the assertions would be separate, write separate t
 
 ### Notifications and concurrency
 
-The client's receive loop dispatches each incoming message to completion before reading the next,
+The client's dispatcher starts a task per incoming notification in arrival order but does not
-and the in-memory transport delivers everything on one ordered stream. Together these guarantee
+await it before reading the next message, so completion order is not structural. What still
-that every notification a server handler emits before its response reaches the client callback
+holds: the in-memory transport delivers everything on one ordered stream, and a callback that
-before the originating request returns — so tests collect notifications into a plain list and
+records synchronously (no `await` before the append) finishes its scheduling slice before the
-assert after the call, with no synchronisation. The exceptions:
+awaited request's waiter — woken strictly later — resumes. So tests whose callbacks are plain
+appends may still collect into a list and assert after the call. A callback that awaits before
+recording loses that ordering and must synchronise. The other exceptions:
 
 - a notification not triggered by a request the test is awaiting needs an `anyio.Event` set in
   the receiving handler and awaited under `anyio.fail_after(5)`;
@@ -220,9 +222,8 @@ but still inside an outer `async with`, and no restructure can avoid it.
 
 A handful of `# pragma: lax no cover` markers in `src/` cover teardown exception handlers whose
 execution is timing-dependent under the in-process HTTP bridge — the POST-stream and
-stateless-session `except Exception` handlers in `server/streamable_http*.py`, the `_terminated`
+stateless-session `except Exception` handlers in `server/streamable_http*.py` and the
-check in `message_router`, and the response-stream double-close guard in
+`_terminated` check in `message_router`. `strict-no-cover` does not check `lax` lines; do not
-`BaseSession._receive_loop`. `strict-no-cover` does not check `lax` lines; do not promote them to
+promote them to strict `no cover` without first making the teardown ordering deterministic. The
-strict `no cover` without first making the teardown ordering deterministic. The suite also relies
+suite also relies on a one-line `src/mcp/server/sse.py` fix (`sse_stream_reader.aclose()`) that
-on a one-line `src/mcp/server/sse.py` fix (`sse_stream_reader.aclose()`) that closes a stream the
+closes a stream the SSE leg would otherwise leak.
-SSE leg would otherwise leak.

tests/interaction/_helpers.py

@@ -67,8 +67,9 @@ class _RecordingWriteStream:
         self._log = log
 
     async def send(self, item: SessionMessage, /) -> None:
-        self._log.append(item)
+        # Record only after the inner send returns: a failed or cancelled send never reached the transport.
         await self._inner.send(item)
+        self._log.append(item)
 
     async def aclose(self) -> None:
         await self._inner.aclose()