fix: notification handlers and progress callbacks cannot crash the connection

Both run as bare tasks in the dispatcher's task group; an uncaught
exception cancels every sibling (read loop + in-flight requests) and
tears down run(). The previous receive-loop swallowed and logged both.

ServerRunner._on_notify: ValidationError -> warning + drop; handler
Exception -> logger.exception + swallow.

JSONRPCDispatcher: user on_progress callbacks are wrapped so a raise is
logged and swallowed instead of cascading.

Regression tests confirmed to fail before the fix.

Author: maxisbey

src/mcp/server/runner.py

@@ -23,7 +23,7 @@
 
 import anyio.abc
 from opentelemetry.trace import SpanKind, StatusCode
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from typing_extensions import TypeVar
 
 from mcp.server.connection import Connection
@@ -237,9 +237,20 @@ async def _on_notify(
             return
         # Absent wire params reach the handler as `None`, not an empty model
         # (matches the existing `Server._handle_notification`).
-        typed_params = entry.params_type.model_validate(params) if params is not None else None
+        try:
+            typed_params = entry.params_type.model_validate(params) if params is not None else None
+        except ValidationError:
+            logger.warning("dropped %r: malformed params", method)
+            return
         ctx = self._make_context(dctx, typed_params)
-        await entry.handler(ctx, typed_params)
+        try:
+            await entry.handler(ctx, typed_params)
+        except Exception:
+            # Top-level boundary: a notification handler crashing must not
+            # tear down the connection (it runs as a bare task in the
+            # dispatcher's task group; an uncaught exception would cancel
+            # every sibling, including the read loop and in-flight requests).
+            logger.exception("notification handler for %r raised", method)
 
     def _make_context(
         self, dctx: DispatchContext[TransportContext], typed_params: BaseModel | None

src/mcp/shared/jsonrpc_dispatcher.py

@@ -162,6 +162,24 @@ def _default_transport_builder(_meta: MessageMetadata) -> TransportContext:
     return TransportContext(kind="jsonrpc", can_send_request=True)
 
 
+def _shielded_progress(fn: ProgressFnT) -> ProgressFnT:
+    """Wrap a user progress callback so it can't crash the dispatcher.
+
+    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:
+            await fn(progress, total, message)
+        except Exception:
+            logger.exception("progress callback raised")
+
+    return _wrapped
+
+
 def _outbound_metadata(related_request_id: RequestId | None, opts: CallOptions | None) -> MessageMetadata:
     """Choose the `SessionMessage.metadata` for an outgoing request/notification.
 
@@ -471,7 +489,7 @@ def _dispatch_notification(
                     total = msg.params.get("total")
                     message = msg.params.get("message")
                     self._spawn(
-                        pending.on_progress,
+                        _shielded_progress(pending.on_progress),
                         float(progress),
                         float(total) if isinstance(total, int | float) else None,
                         message if isinstance(message, str) else None,

tests/server/conftest.py

@@ -44,5 +44,3 @@ def spans(capfire: CaptureLogfire) -> Iterator[SpanCapture]:
     capture.clear()
     yield capture
     capture.clear()
-
-

tests/server/test_runner.py

@@ -220,6 +220,39 @@ async def on_roots_changed(ctx: Ctx, params: NotificationParams | None) -> None:
     assert isinstance(seen[1][1], NotificationParams)
 
 
+@pytest.mark.anyio
+async def test_runner_on_notify_handler_exception_is_swallowed_and_logged(
+    server: SrvT, caplog: pytest.LogCaptureFixture
+):
+    """A notification handler crashing must not tear down the connection."""
+
+    async def boom(ctx: Ctx, params: NotificationParams | None) -> None:
+        raise RuntimeError("notification handler boom")
+
+    server.add_notification_handler("notifications/roots/list_changed", NotificationParams, boom)
+    async with connected_runner(server) as (client, _):
+        await client.notify("notifications/roots/list_changed", None)
+        # Connection still alive: a request after the crashing handler succeeds.
+        result = await client.send_raw_request("tools/list", None)
+    assert result["tools"][0]["name"] == "t"
+    assert "notification handler for 'notifications/roots/list_changed' raised" in caplog.text
+
+
+@pytest.mark.anyio
+async def test_runner_on_notify_drops_malformed_params(server: SrvT, caplog: pytest.LogCaptureFixture):
+    """Malformed notification params are logged and dropped, not raised."""
+
+    async def on_level(ctx: Ctx, params: SetLevelRequestParams) -> None:
+        raise NotImplementedError
+
+    server.add_notification_handler("notifications/roots/list_changed", SetLevelRequestParams, on_level)
+    async with connected_runner(server) as (client, _):
+        await client.notify("notifications/roots/list_changed", {"level": "not-a-level"})
+        result = await client.send_raw_request("tools/list", None)
+    assert result["tools"][0]["name"] == "t"
+    assert "dropped 'notifications/roots/list_changed': malformed params" in caplog.text
+
+
 @pytest.mark.anyio
 async def test_runner_on_notify_drops_before_init_and_unknown_methods(server: SrvT):
     async with connected_runner(server, initialized=False) as (client, _):

tests/shared/test_jsonrpc_dispatcher.py

@@ -441,6 +441,26 @@ async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] |
     assert received == [(0.25, None, None)]
 
 
+@pytest.mark.anyio
+async def test_progress_callback_exception_is_swallowed_and_logged(caplog: pytest.LogCaptureFixture):
+    """A user progress callback raising must not crash the dispatcher."""
+
+    async def boom(progress: float, total: float | None, message: str | None) -> None:
+        raise RuntimeError("progress callback boom")
+
+    async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
+        await ctx.progress(0.5)
+        return {"ok": True}
+
+    opts: CallOptions = {"on_progress": boom}
+    async with running_pair(jsonrpc_pair, server_on_request=server_on_request) as (client, *_):
+        with anyio.fail_after(5):
+            result = await client.send_raw_request("t", None, opts)
+    # Request still completes; the callback's crash was swallowed.
+    assert result == {"ok": True}
+    assert "progress callback raised" in caplog.text
+
+
 @pytest.mark.anyio
 async def test_send_raw_request_always_carries_meta_on_the_wire():
     """Outbound requests always include `params._meta` (otel injection per SEP-414).