fix: address second bughunter pass (8 findings)

Dispatcher:
- notify omits the params key entirely when params is None; the spec
  requires absent over null and the TS SDK client rejects null.
- transport_builder calls on the read-loop path are guarded: a raising
  builder answers the request with INTERNAL_ERROR (or drops the
  notification) instead of killing the dispatcher.
- bool no longer matches the int() structural patterns (progressToken,
  cancelled requestId, progress), so true cannot alias to request id 1.
  The cancelled arm and the _in_flight table now coerce ids the same way
  responses do, so string/int id forms correlate both ways.
- Dropped the redundant outer _in_flight pop in _handle_request; it
  could evict a newer request that reused the id. Resumption hints
  passed alongside related_request_id are dropped with a debug log and
  the precedence is documented.
- notifications/cancelled now flows through to on_notify after the
  dispatcher applies its cancellation, so Server.middleware observes it
  and servers can register custom handling.

Runner / peer:
- dump_params serializes meta through RequestParams by alias (the
  inverse of _extract_meta), so a handler passing meta=ctx.meta emits
  progressToken on the wire, not progress_token.
- A handler returning ErrorData now raises MCPError inside call_next()
  so middleware observes the failure; _dump_result keeps the conversion
  as a backstop for middleware that itself returns ErrorData.

docs/migration.md: middleware example takes the raw mapping; remove the
unreachable DispatchMiddleware pointer; fix the inverted
raise_exceptions claim.

Author: maxisbey

docs/migration.md

@@ -835,16 +835,15 @@ server.add_notification_handler("notifications/custom", MyNotifyParams, my_notif
 These were private, but some users subclassed `Server` and overrode them to intercept requests. Use middleware instead:
 
 ```python
+from collections.abc import Mapping
 from typing import Any
 
-from pydantic import BaseModel
-
 from mcp.server import Server, ServerRequestContext
 from mcp.server.context import CallNext, HandlerResult
 
 
 async def logging_middleware(
-    ctx: ServerRequestContext[Any, Any], method: str, params: BaseModel, call_next: CallNext
+    ctx: ServerRequestContext[Any, Any], method: str, params: Mapping[str, Any] | None, call_next: CallNext
 ) -> HandlerResult:
     print(f"handling {method}")
     result = await call_next()
@@ -856,11 +855,11 @@ server = Server("my-server", on_call_tool=...)
 server.middleware.append(logging_middleware)
 ```
 
-For lower-level interception (raw method/params before validation, including unknown methods), use `DispatchMiddleware` from `mcp.shared.dispatcher`.
+Middleware runs before params validation, so `params` is the raw inbound mapping (or `None`), and it also wraps unknown methods.
 
 ### Lowlevel `Server.run(raise_exceptions=True)`: transport errors no longer re-raised
 
-`raise_exceptions=True` now only governs handler exceptions: an exception raised by an `on_*` handler propagates out of `run()` instead of being converted to a JSON-RPC error response.
+`raise_exceptions=True` now only governs handler exceptions: an exception raised by an `on_*` handler propagates out of `run()`. The JSON-RPC error response is still written to the client first, regardless of the flag.
 
 Previously it also re-raised exceptions yielded by the transport onto the read stream (e.g. JSON parse errors). Those are now debug-logged and dropped regardless of `raise_exceptions`. If you relied on `run()` exiting on a transport-level parse error, that no longer happens.
 
@@ -1115,7 +1114,7 @@ In practice, replace direct `ServerSession` use with `Server.run(read_stream, wr
 - `ServerRequestResponder` type alias.
 - `ServerSession.incoming_messages` stream — there is no longer a public stream of inbound messages to iterate. Register handlers via the `on_*` constructor params (or `add_request_handler`) and use `Server.middleware` to observe every inbound request and notification (`initialize`, unknown methods, validation failures, and `notifications/initialized` included).
 - `ServerSession.__aenter__` / `__aexit__` — `ServerSession` is no longer an async context manager.
-- The private `_receive_loop`, `_received_request`, `_received_notification`, and `_handle_incoming` overrides — there is nothing to override on `ServerSession` anymore. To intercept inbound messages, use `Server.middleware` or `DispatchMiddleware` (see the `_handle_*` removal section above).
+- The private `_receive_loop`, `_received_request`, `_received_notification`, and `_handle_incoming` overrides — there is nothing to override on `ServerSession` anymore. To intercept inbound messages, use `Server.middleware` (see the `_handle_*` removal section above).
 
 ### `BaseSession` / `RequestResponder`: server-side cancellation tracking removed
 

src/mcp/server/context.py

@@ -123,7 +123,9 @@ class ServerMiddleware(Protocol[_MwLifespanT]):
     is built but before any validation, lookup, or handshake. Wraps every
     inbound request and notification: `initialize`, the pre-init gate,
     `METHOD_NOT_FOUND`, params validation, the handler call, and
-    `notifications/initialized` all run inside `call_next()`. A request-side
+    `notifications/initialized` all run inside `call_next()`.
+    `notifications/cancelled` is observed too; the dispatcher applies the
+    cancellation itself, then forwards the notification. A request-side
     failure reaches the middleware as a raised `MCPError` (or
     `ValidationError` for malformed params) so observation/logging middleware
     can record it. Listed outermost-first on `Server.middleware`.

src/mcp/server/runner.py

@@ -141,10 +141,12 @@ def _dump_result(result: Any) -> dict[str, Any]:
     if result is None:
         return {}
     if isinstance(result, ErrorData):
-        # The existing `BaseSession._send_response` treats a handler-returned
-        # `ErrorData` as a JSON-RPC error, not a success result. Re-raise as
-        # `MCPError` so the dispatcher's exception boundary emits `JSONRPCError`.
-        raise MCPError(code=result.code, message=result.message, data=result.data)
+        # The existing `BaseSession._send_response` treats a returned
+        # `ErrorData` as a JSON-RPC error, not a success result. Handler
+        # returns are converted inside `_on_request`'s `_inner` (so
+        # `Server.middleware` observes the raise); this branch is the boundary
+        # check for middleware that itself returns `ErrorData`.
+        raise MCPError.from_error_data(result)
     if isinstance(result, BaseModel):
         return result.model_dump(by_alias=True, mode="json", exclude_none=True)
     if isinstance(result, dict):
@@ -266,7 +268,16 @@ async def _inner() -> HandlerResult:
                 typed_params = None
             else:
                 typed_params = entry.params_type.model_validate(params, by_name=False)
-            return await entry.handler(ctx, typed_params)
+            result = await entry.handler(ctx, typed_params)
+            if isinstance(result, ErrorData):
+                # A handler-returned `ErrorData` is a JSON-RPC error, not a
+                # success result (matches `BaseSession._send_response`). Raise
+                # here, inside the middleware chain, so `Server.middleware`
+                # observes the failure as a raised `MCPError` out of
+                # `call_next()` per the `ServerMiddleware` contract instead of
+                # a successful-looking `ErrorData` return.
+                raise MCPError.from_error_data(result)
+            return result
 
         call = self._compose_server_middleware(ctx, method, params, _inner)
         return _dump_result(await call())

src/mcp/shared/dispatcher.py

@@ -62,15 +62,19 @@ class CallOptions(TypedDict, total=False):
     """Opaque token to resume a previously interrupted request.
 
     Client-side, streamable-HTTP only. Ignored by server dispatchers and other
-    transports. Supports protocol version 2025-11-25 and earlier; SSE-stream
+    transports, and also ignored (with a debug log) for requests sent from a
+    `DispatchContext`, where routing onto the inbound request's stream takes
+    precedence. Supports protocol version 2025-11-25 and earlier; SSE-stream
     resumption is removed in the next protocol revision.
     """
 
     on_resumption_token: Callable[[str], Awaitable[None]]
     """Receive a resumption token when the transport issues one for this request.
 
     Client-side, streamable-HTTP only. Ignored by server dispatchers and other
-    transports. Supports protocol version 2025-11-25 and earlier; SSE-stream
+    transports, and also ignored (with a debug log) for requests sent from a
+    `DispatchContext`, where routing onto the inbound request's stream takes
+    precedence. Supports protocol version 2025-11-25 and earlier; SSE-stream
     resumption is removed in the next protocol revision.
     """
 

src/mcp/shared/jsonrpc_dispatcher.py

@@ -45,6 +45,7 @@
 from mcp.shared.transport_context import TransportContext
 from mcp.types import (
     CONNECTION_CLOSED,
+    INTERNAL_ERROR,
     INVALID_PARAMS,
     REQUEST_CANCELLED,
     REQUEST_TIMEOUT,
@@ -189,8 +190,17 @@ def _outbound_metadata(related_request_id: RequestId | None, opts: CallOptions |
     request it belongs to (so streamable-HTTP can route it onto that request's
     SSE stream). `ClientMessageMetadata` carries resumption hints to the
     client transport. `None` is the common case.
+
+    `SessionMessage.metadata` carries exactly one of these, so when
+    `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.
     """
     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):
+            logger.debug(
+                "dropping resumption hints: related_request_id %r takes precedence on metadata", related_request_id
+            )
         return ServerMessageMetadata(related_request_id=related_request_id)
     if opts:
         token = opts.get("resumption_token")
@@ -366,7 +376,14 @@ async def notify(
         *,
         _related_request_id: RequestId | None = None,
     ) -> None:
-        msg = JSONRPCNotification(jsonrpc="2.0", method=method, params=dict(params) if params is not None else None)
+        # Leave `params` unset (not explicitly None) when there are none:
+        # transports serialize with `exclude_unset=True`, and an explicit None
+        # would survive as `"params": null`, which JSON-RPC 2.0 forbids and
+        # strict peers (e.g. the TypeScript SDK's zod schemas) reject.
+        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))
 
     async def run(
@@ -464,11 +481,26 @@ async def _dispatch_request(
     ) -> None:
         progress_token: ProgressToken | None
         match req.params:
-            case {"_meta": {"progressToken": str() | int() as progress_token}}:
+            # The bool guard matters: `int()` patterns match bool (a subclass),
+            # 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 _:
                 progress_token = None
-        transport_ctx = self._transport_builder(metadata)
+        try:
+            transport_ctx = self._transport_builder(metadata)
+        except Exception:
+            # Containment boundary for the user-supplied builder: a raising
+            # 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,
+                req.id,
+                ErrorData(code=INTERNAL_ERROR, message="transport context unavailable"),
+                sender_ctx=sender_ctx,
+            )
+            return
         dctx = _JSONRPCDispatchContext(
             transport=transport_ctx,
             _dispatcher=self,
@@ -480,7 +512,11 @@ async def _dispatch_request(
         # TODO(maxisbey): the spec puts request-id uniqueness on the sender;
         # neither v1 nor the TS SDK guards a duplicate id here, so for now we
         # blind-overwrite (parity). Revisit rejecting with INVALID_REQUEST.
-        self._in_flight[req.id] = _InFlight(scope=scope, dctx=dctx)
+        # 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
             # park the read loop until the handler returns; that's the inline
@@ -512,22 +548,34 @@ def _dispatch_notification(
         `notifications/cancelled` and `notifications/progress` are intercepted
         here because they correlate against JSON-RPC request IDs - the
         `_in_flight` / `_pending` tables this layer owns - so no higher layer
-        can act on them. See the module docstring for the design rationale.
+        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:
-                case {"requestId": str() | int() as rid} if (in_flight := self._in_flight.get(rid)) is not None:
+                # The bool guards here and below matter: `int()` patterns match
+                # 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
+                ):
                     in_flight.dctx.cancel_requested.set()
                     if self._peer_cancel_mode == "interrupt":
                         in_flight.scope.cancel()
                 case _:
                     pass
-            return
-        if msg.method == "notifications/progress":
+            # 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 (
-                    pending := self._pending.get(_coerce_id(token))
-                ) is not None and pending.on_progress is not None:
+                    not isinstance(token, bool)
+                    and not isinstance(progress, bool)
+                    and (pending := self._pending.get(_coerce_id(token))) is not None
+                    and pending.on_progress is not None
+                ):
                     total = msg.params.get("total")
                     message = msg.params.get("message")
                     self._spawn(
@@ -540,7 +588,13 @@ def _dispatch_notification(
                 case _:
                     pass
             # fall through: progress is also teed to on_notify
-        transport_ctx = self._transport_builder(metadata)
+        try:
+            transport_ctx = self._transport_builder(metadata)
+        except Exception:
+            # Same containment boundary as `_dispatch_request`: a raising
+            # 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
         )
@@ -615,7 +669,7 @@ async def _handle_request(
                     # handler return and the pop, so the cancel can't
                     # interleave there.
                     dctx.close()
-                    self._in_flight.pop(req.id, None)
+                    self._in_flight.pop(_coerce_id(req.id), None)
                 await self._write_result(req.id, result)
             if scope.cancel_called:
                 # Peer-cancel: `_dispatch_notification` cancelled this scope
@@ -649,8 +703,10 @@ async def _handle_request(
             await self._write_error(req.id, ErrorData(code=0, message=str(e)))
             if self._raise_handler_exceptions:
                 raise
-        finally:
-            self._in_flight.pop(req.id, None)
+        # No outer `_in_flight` pop here: the inner `finally` above already
+        # 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

src/mcp/shared/peer.py

@@ -11,7 +11,7 @@
 """
 
 from collections.abc import Mapping
-from typing import Any, overload
+from typing import Any, cast, overload
 
 from pydantic import BaseModel
 
@@ -27,6 +27,8 @@
     IncludeContext,
     ListRootsResult,
     ModelPreferences,
+    RequestParams,
+    RequestParamsMeta,
     SamplingMessage,
     Tool,
     ToolChoice,
@@ -44,11 +46,18 @@ def dump_params(model: BaseModel | None, meta: Meta | None = None) -> dict[str,
     Shared by `ClientPeerMixin`, `Connection`, and `TypedServerRequestMixin` so every
     typed convenience method gets the same `_meta` handling. `meta` keys take
     precedence over any `_meta` already present on the model.
+
+    `meta` is serialized through `RequestParams` so Python field names emit
+    their wire aliases: an inbound `ctx.meta` carries `progress_token` (the
+    key `_extract_meta` validation produces), and forwarding it outbound via
+    `meta=ctx.meta` must put `progressToken` back on the wire. Keys not
+    declared on `RequestParamsMeta` pass through unchanged.
     """
     out = model.model_dump(by_alias=True, mode="json", exclude_none=True) if model is not None else None
     if meta:
+        wire_meta = RequestParams(_meta=cast(RequestParamsMeta, meta)).model_dump(by_alias=True, mode="json")["_meta"]
         out = dict(out or {})
-        out["_meta"] = {**out.get("_meta", {}), **meta}
+        out["_meta"] = {**out.get("_meta", {}), **wire_meta}
     return out