Contain notification-handler exceptions in the dispatcher

A raising notification handler ran as a bare task in the dispatcher's task
group, so its exception cancelled the read loop and every in-flight request.
Wrap spawned handlers in the same containment boundary progress callbacks
already have: log the failure and keep the connection serving.

Author: maxisbey

src/mcp/shared/jsonrpc_dispatcher.py

@@ -34,7 +34,7 @@
 
 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,
@@ -190,6 +190,18 @@ async def _wrapped(progress: float, total: float | None, message: str | None) ->
     return _wrapped
 
 
+def _contained_notify(fn: OnNotify) -> OnNotify:
+    """Wrap a notification handler so it can't crash the dispatcher (same boundary as `_shielded_progress`)."""
+
+    async def _wrapped(dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None) -> None:
+        try:
+            await fn(dctx, method, params)
+        except Exception:
+            logger.exception("notification handler for %r raised", method)
+
+    return _wrapped
+
+
 def _outbound_metadata(related_request_id: RequestId | None, opts: CallOptions | None) -> MessageMetadata:
     """Choose the `SessionMessage.metadata` for an outgoing request/notification.
 
@@ -619,7 +631,7 @@ def _dispatch_notification(
         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

tests/shared/test_jsonrpc_dispatcher.py

@@ -610,6 +610,56 @@ async def caller() -> None:
     assert scopes[0].cancelled_caught
 
 
+@pytest.mark.anyio
+async def test_notification_handler_exception_is_contained(caplog: pytest.LogCaptureFixture):
+    """A raising notification handler costs only that notification, never the connection.
+
+    The handler runs as a bare task in the dispatcher's task group; without containment its
+    exception would cancel the read loop and every in-flight request. The TypeScript, C#, and
+    Go engines all contain notification-handler failures the same way.
+    """
+
+    async def server_on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None:
+        raise RuntimeError("notify boom")
+
+    async with running_pair(jsonrpc_pair, server_on_notify=server_on_notify) as (client, *_):
+        with anyio.fail_after(5):
+            await client.notify("boom", None)
+            # The connection survived: a full round-trip still works.
+            result = await client.send_raw_request("ping", None)
+    assert result == {"echoed": "ping", "params": {}}
+    assert "notification handler for 'boom' raised" in caplog.text
+
+
+@pytest.mark.anyio
+async def test_spawned_notification_handlers_run_concurrently():
+    """Notification handlers are spawned, not serialized: a parked one does not block the next.
+
+    The first handler waits for the second to have started - serialized dispatch would deadlock
+    here. This matches the TypeScript and C# engines (fire-and-forget); handlers needing
+    mutual ordering must coordinate themselves.
+    """
+    second_started = anyio.Event()
+    completed: list[str] = []
+    done = anyio.Event()
+
+    async def server_on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None:
+        if method == "first":
+            await second_started.wait()
+        else:
+            second_started.set()
+        completed.append(method)
+        if len(completed) == 2:
+            done.set()
+
+    async with running_pair(jsonrpc_pair, server_on_notify=server_on_notify) as (client, *_):
+        with anyio.fail_after(5):
+            await client.notify("first", None)
+            await client.notify("second", None)
+            await done.wait()
+    assert completed == ["second", "first"]
+
+
 @pytest.mark.anyio
 async def test_ctx_message_metadata_carries_inbound_request_metadata():
     """Transport-attached metadata (HTTP request, SSE close hooks) is readable off the dispatch context."""