Harden ClientSession/dispatcher lifecycle and error handling

DirectDispatcher now mirrors JSONRPCDispatcher's lifecycle so the
public ClientSession(dispatcher=...) path honors the documented
contracts: send_raw_request raises RuntimeError before run() and
MCPError(CONNECTION_CLOSED) after close, inbound dispatch to a closed
side fails the peer's request instead of serving the exited session's
callbacks, and the ready-wait sits inside fail_after so a request
timeout bounds waiting on a peer whose run() never started.

ClientSession._on_notify self-contains a raising logging_callback /
message_handler (matching ServerRunner._on_notify) so a callback that
throws over DirectDispatcher costs only that delivery instead of
failing the in-process peer's notify() call.

JSONRPCDispatcher.notify is now fire-and-forget all the way: a post-
close send or a write onto a torn-down transport drops the
notification with a debug log instead of leaking raw anyio
ClosedResourceError/BrokenResourceError. DirectDispatcher.notify
matches.

JSONRPCDispatcher.send_raw_request: a builtin TimeoutError raised by a
custom transport's bounded send() now propagates raw instead of being
mislabelled REQUEST_TIMEOUT (the timeout-conversion arm only fires
once the response fail_after has armed). request_write_started is set
before the write so a delivered-but-cancelled request still triggers
the courtesy notifications/cancelled, and _handle_request skips the
shutdown CONNECTION_CLOSED answer once an answer write has started,
preferring possibly-zero responses over possibly-two for one id.

request_read_timeout_seconds=0.0 is now honored as fail-immediately on
both ClientSession.send_request and ServerSession.send_request (was
silently treated as unset via or-fallback). None remains the only
sentinel that defers to the session-level timeout.

Author: maxisbey

src/mcp/client/session.py

@@ -226,7 +226,11 @@ async def send_request(
         data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
         method: str = data["method"]
         opts: CallOptions = {}
-        timeout = request_read_timeout_seconds or self._session_read_timeout_seconds
+        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:
@@ -243,7 +247,11 @@ async def send_request(
         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."""
+        """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"))
 
@@ -541,9 +549,16 @@ async def _on_notify(
         if isinstance(notification, types.CancelledNotification):
             # The dispatcher already applied the cancellation; not surfaced to message_handler.
             return
-        if isinstance(notification, types.LoggingMessageNotification):
-            await self._logging_callback(notification.params)
-        await self._message_handler(notification)
+        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 _on_stream_exception(self, exc: Exception) -> None:
         """Deliver a transport-level fault to message_handler via a spawned task.

src/mcp/server/session.py

@@ -84,7 +84,7 @@ async def send_request(
         """
         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/shared/direct_dispatcher.py

@@ -15,6 +15,7 @@
 
 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.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"]
 
@@ -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):
@@ -93,7 +103,9 @@ def __init__(self, transport_ctx: TransportContext):
         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
@@ -104,13 +116,35 @@ async def send_raw_request(
         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(
@@ -120,14 +154,29 @@ async def run(
         *,
         task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
     ) -> None:
-        self._on_request = on_request
-        self._on_notify = on_notify
-        self._ready.set()
-        task_status.started()
-        await self._closed.wait()
+        """Mark this side ready and park until `close()` is called.
+
+        Single-shot, like `JSONRPCDispatcher.run`: once it returns the
+        dispatcher stays closed and cannot be restarted.
+        """
+        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, request_id: RequestId | None = None
@@ -142,20 +191,40 @@ def _make_context(
             _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 {}
-        # 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:
             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:
@@ -173,7 +242,13 @@ async def _dispatch_request(
             ) 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/jsonrpc_dispatcher.py

@@ -16,6 +16,7 @@
 
 import anyio
 import anyio.abc
+import anyio.lowlevel
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from opentelemetry.trace import SpanKind
 from pydantic import ValidationError
@@ -296,9 +297,13 @@ async def send_raw_request(
         self._pending[request_id] = pending
 
         plan = _plan_outbound(_related_request_id, opts)
-        # Spec MUST: only previously-issued requests may be cancelled, so the
-        # courtesy cancel arms only once the request write completes.
-        request_written = False
+        # 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 ''}"
@@ -313,18 +318,28 @@ async def send_raw_request(
                 # SEP-414: inject W3C trace context; `_meta` stays on the wire even with a no-op tracer.
                 inject_trace_context(out_meta)
                 msg = JSONRPCRequest(jsonrpc="2.0", id=request_id, method=method, params=out_params)
+                # 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
-                request_written = True
                 with anyio.fail_after(opts.get("timeout")):
+                    timeout_armed = True
                     outcome = await receive.receive()
         except TimeoutError:
+            if not timeout_armed:
+                # `fail_after` arms only after the write, so this TimeoutError is the
+                # transport's own bounded send() failing - a transport error, not
+                # `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 and request_written:
+            if plan.cancel_on_abandon:
                 await self._final_write(
                     partial(
                         self._cancel_outbound,
@@ -340,7 +355,7 @@ async def send_raw_request(
         except anyio.get_cancelled_exc_class():
             # Caller cancelled: bare awaits re-raise here, so the shielded helper
             # lets the courtesy cancel go out before we propagate.
-            if plan.cancel_on_abandon and request_written:
+            if plan.cancel_on_abandon and request_write_started:
                 await self._final_write(
                     partial(self._cancel_outbound, request_id, "caller cancelled", _related_request_id),
                     shield=True,
@@ -365,13 +380,26 @@ async def notify(
         *,
         _related_request_id: RequestId | None = None,
     ) -> None:
+        """Send a fire-and-forget notification.
+
+        Fire-and-forget all the way: a post-close send or a write onto a
+        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, _plan_outbound(_related_request_id, None).metadata)
+        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,
@@ -613,6 +641,7 @@ async def _handle_request(
 
         The single exception-to-wire boundary: handler exceptions become `JSONRPCError` here.
         """
+        answer_write_started = False
         try:
             with scope:
                 try:
@@ -625,23 +654,33 @@ async def _handle_request(
                     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.cancelled_caught:
                 # anyio absorbs the scope's own cancel at __exit__, and
                 # `cancelled_caught` (unlike `cancel_called`) guarantees the
                 # result write above did not happen - no double response.
                 # TODO(maxisbey): spec says SHOULD NOT respond after cancel;
                 # 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():
-            # Shutdown: answer the request so the peer isn't left waiting; the
-            # shielded helper is needed because bare awaits re-raise here.
-            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}",
-            )
+            # Shutdown: answer the request so the peer isn't left waiting - unless
+            # an answer write already started (it may have reached the transport;
+            # prefer possibly-zero answers over possibly-two). The shielded helper
+            # 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)
@@ -701,11 +740,9 @@ async def _final_write(
     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
         # the request's own SSE stream instead of a possibly-absent GET stream.
-        try:
-            await self.notify(
-                "notifications/cancelled",
-                {"requestId": request_id, "reason": reason},
-                _related_request_id=related_request_id,
-            )
-        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
-            pass
+        # `notify` swallows connection-state errors itself, so no guard here.
+        await self.notify(
+            "notifications/cancelled",
+            {"requestId": request_id, "reason": reason},
+            _related_request_id=related_request_id,
+        )