review: Server.middleware wraps every request and notification (t04/t18); t25 TODO

ServerMiddleware now wraps the entire _on_request and _on_notify body so it
observes initialize, notifications/initialized, METHOD_NOT_FOUND, validation
failures, and pre-init drops - everything the old incoming_messages stream
saw. params is the raw Mapping[str, Any] | None (not the typed model);
ctx.request_id is None distinguishes a notification; call_next() raises
MCPError for request-side failures so middleware can observe them, returns
None for notifications. _on_notify keeps its swallow-at-boundary behavior
but middleware sees the raise first.

ServerRunner._on_request/_on_notify restructured: build ctx from raw inputs
first (_extract_meta validates only _meta via RequestParams so ctx.meta
keeps the alias-converted shape), then compose Server.middleware around an
_inner() containing validation/init/gate/lookup/handler.
_compose_server_middleware helper shared by both paths.
_handle_initialize returns InitializeResult (outer _dump_result serializes)
so middleware can transform it. _make_context now takes meta directly.

t25: TODO at the duplicate-inbound-request-id site (parity with v1/TS;
spec puts uniqueness on the sender).

Author: maxisbey

docs/migration.md

@@ -1113,7 +1113,7 @@ In practice, replace direct `ServerSession` use with `Server.run(read_stream, wr
 
 - `InitializationState` enum and `ServerSession._initialization_state` — initialization tracking is now on `Connection` (`connection.initialized` is an `anyio.Event`, `connection.client_params` holds the init params).
 - `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 request.
+- `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).
 

src/mcp/server/context.py

@@ -117,12 +117,25 @@ async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *
 
 
 class ServerMiddleware(Protocol[_MwLifespanT]):
-    """Context-tier middleware: `(ctx, method, typed_params, call_next) -> result`.
-
-    Runs *inside* `ServerRunner._on_request` after params validation and
-    context construction. Wraps registered handlers (including `ping`) but
-    not `initialize`, `METHOD_NOT_FOUND`, or validation failures. Listed
-    outermost-first on `Server.middleware`.
+    """Context-tier middleware: `(ctx, method, params, call_next) -> result`.
+
+    Runs at the top of `ServerRunner._on_request` / `_on_notify` after `ctx`
+    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
+    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`.
+
+    `ctx.request_id is None` distinguishes a notification from a request. For
+    notifications `call_next()` returns `None` (a dropped or unhandled
+    notification also returns `None`) and the middleware's own return value is
+    discarded.
+
+    `params` is the raw inbound mapping (no model validation has happened
+    yet). For typed inspection, validate against the model the middleware
+    expects.
 
     `Server[L].middleware` holds `ServerMiddleware[L]`, so an app-specific
     middleware sees `ctx.lifespan_context: L`. While the context is the
@@ -140,6 +153,6 @@ async def __call__(
         self,
         ctx: ServerRequestContext[_MwLifespanT, Any],
         method: str,
-        params: BaseModel | None,
+        params: Mapping[str, Any] | None,
         call_next: CallNext,
     ) -> HandlerResult: ...

src/mcp/server/lowlevel/server.py

@@ -215,7 +215,8 @@ def __init__(
         self._request_handlers: dict[str, HandlerEntry[LifespanResultT]] = {}
         self._notification_handlers: dict[str, HandlerEntry[LifespanResultT]] = {}
         self._session_manager: StreamableHTTPSessionManager | None = None
-        # Context-tier middleware: wraps each request handler with
+        # Context-tier middleware: wraps every inbound request (including
+        # `initialize`, lookup, validation, handler) with
         # `(ctx, method, params, call_next)`. Applied in `ServerRunner._on_request`.
         # TODO(maxisbey): provisional - signature and semantics change with the
         # Context/middleware rework (covariant `Context[L]`, outbound seam) before

src/mcp/server/runner.py

@@ -27,7 +27,7 @@
 from typing_extensions import TypeVar
 
 from mcp.server.connection import Connection
-from mcp.server.context import CallNext, ServerMiddleware, ServerRequestContext
+from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
 from mcp.shared._otel import extract_trace_context, otel_span
@@ -47,6 +47,7 @@
     InitializeRequestParams,
     InitializeResult,
     RequestParams,
+    RequestParamsMeta,
     client_request_adapter,
 )
 
@@ -62,6 +63,24 @@
 
 _INIT_EXEMPT: frozenset[str] = frozenset({"ping"})
 
+
+def _extract_meta(params: Mapping[str, Any] | None) -> RequestParamsMeta | None:
+    """Lift `_meta` from raw params with the same key-aliasing pydantic applies.
+
+    `RequestParams` only declares `meta` (alias `_meta`) and `MCPModel` does
+    not forbid extras, so this validate ignores everything else and never
+    rejects on the caller's other fields. Returns `None` for absent or
+    malformed `_meta` so context construction is independent of params
+    validity (which `_on_request` checks separately).
+    """
+    if not params or "_meta" not in params:
+        return None
+    try:
+        return RequestParams.model_validate(params, by_name=False).meta
+    except ValidationError:
+        return None
+
+
 _SPEC_CLIENT_METHODS: frozenset[str] = frozenset(
     cast(type[BaseModel], arm).model_fields["method"].default for arm in get_args(ClientRequest)
 )
@@ -206,45 +225,50 @@ async def _on_request(
         method: str,
         params: Mapping[str, Any] | None,
     ) -> dict[str, Any]:
-        # TODO(maxisbey): pinned compat. `BaseSession._receive_loop` validates
-        # every inbound request against the spec `ClientRequest` discriminated
-        # union *before* handler lookup, so a spec method with malformed params
-        # surfaces as INVALID_PARAMS via the dispatcher's ValidationError
-        # boundary even when no handler is registered. v2 wanted to decouple
-        # the runner from the spec union; revisit once the suite's divergence
-        # entry is resolved. Gated on spec methods so custom methods registered
-        # via `add_request_handler` still route (the existing server rejects
-        # those too, but nothing pins that and routing them is strictly better).
-        if method in _SPEC_CLIENT_METHODS:
-            payload: dict[str, Any] = {"method": method}
-            if params is not None:
-                payload["params"] = dict(params)
-            client_request_adapter.validate_python(payload, by_name=False)
-        if method == "initialize":
-            return self._handle_initialize(params)
-        if not self._initialized and method not in _INIT_EXEMPT:
-            # TODO(maxisbey): pinned compat. The existing server has no
-            # dedicated pre-init check; the request dies in ClientRequest
-            # validation, so the client sees the generic invalid-params shape.
-            raise MCPError(code=INVALID_PARAMS, message="Invalid request parameters", data="")
-        entry = self.server.get_request_handler(method)
-        if entry is None:
-            raise MCPError(code=METHOD_NOT_FOUND, message="Method not found")
-        # ValidationError propagates; the dispatcher's exception boundary maps
-        # it to INVALID_PARAMS. Absent wire params reach the handler as None
-        # (matches the existing `Server._handle_request`, where `req.params`
-        # is None for optional-params requests like tools/list); the empty-dict
-        # validate is a required-field check so a required-params model still
-        # surfaces as INVALID_PARAMS rather than reaching the handler as None.
-        if params is None:
-            entry.params_type.model_validate({}, by_name=False)
-            typed_params = None
-        else:
-            typed_params = entry.params_type.model_validate(params, by_name=False)
-        ctx = self._make_context(dctx, typed_params)
-        call: CallNext = partial(entry.handler, ctx, typed_params)
-        for mw in reversed(self.server.middleware):
-            call = partial(mw, ctx, method, typed_params, call)
+        ctx = self._make_context(dctx, _extract_meta(params))
+
+        async def _inner() -> HandlerResult:
+            # TODO(maxisbey): pinned compat. `BaseSession._receive_loop`
+            # validates every inbound request against the spec `ClientRequest`
+            # discriminated union *before* handler lookup, so a spec method
+            # with malformed params surfaces as INVALID_PARAMS via the
+            # dispatcher's ValidationError boundary even when no handler is
+            # registered. v2 wanted to decouple the runner from the spec union;
+            # revisit once the suite's divergence entry is resolved. Gated on
+            # spec methods so custom methods registered via
+            # `add_request_handler` still route (the existing server rejects
+            # those too, but nothing pins that and routing is strictly better).
+            if method in _SPEC_CLIENT_METHODS:
+                payload: dict[str, Any] = {"method": method}
+                if params is not None:
+                    payload["params"] = dict(params)
+                client_request_adapter.validate_python(payload, by_name=False)
+            if method == "initialize":
+                return self._handle_initialize(params)
+            if not self._initialized and method not in _INIT_EXEMPT:
+                # TODO(maxisbey): pinned compat. The existing server has no
+                # dedicated pre-init check; the request dies in ClientRequest
+                # validation, so the client sees the generic invalid-params
+                # shape.
+                raise MCPError(code=INVALID_PARAMS, message="Invalid request parameters", data="")
+            entry = self.server.get_request_handler(method)
+            if entry is None:
+                raise MCPError(code=METHOD_NOT_FOUND, message="Method not found")
+            # ValidationError propagates; the dispatcher's exception boundary
+            # maps it to INVALID_PARAMS. Absent wire params reach the handler
+            # as None (matches the existing `Server._handle_request`, where
+            # `req.params` is None for optional-params requests like
+            # tools/list); the empty-dict validate is a required-field check
+            # so a required-params model still surfaces as INVALID_PARAMS
+            # rather than reaching the handler as None.
+            if params is None:
+                entry.params_type.model_validate({}, by_name=False)
+                typed_params = None
+            else:
+                typed_params = entry.params_type.model_validate(params, by_name=False)
+            return await entry.handler(ctx, typed_params)
+
+        call = self._compose_server_middleware(ctx, method, params, _inner)
         return _dump_result(await call())
 
     async def _on_notify(
@@ -253,45 +277,68 @@ async def _on_notify(
         method: str,
         params: Mapping[str, Any] | None,
     ) -> None:
-        if method == "notifications/initialized":
-            self._initialized = True
-            self.connection.initialized.set()
-            return
-        if not self._initialized:
-            logger.debug("dropped %s: received before initialization", method)
-            return
-        entry = self.server.get_notification_handler(method)
-        if entry is None:
-            logger.debug("no handler for notification %s", method)
-            return
-        # Absent wire params reach the handler as None, not an empty model
-        # (matches the existing `Server._handle_notification`). The empty-dict
-        # validate is a required-field check: a required-params model (e.g.
-        # ProgressNotificationParams) takes the malformed-params drop path
-        # instead of reaching a non-Optional handler as None.
-        try:
-            if params is None:
-                entry.params_type.model_validate({}, by_name=False)
-                typed_params = None
-            else:
-                typed_params = entry.params_type.model_validate(params, by_name=False)
-        except ValidationError:
-            logger.warning("dropped %r: malformed params", method)
-            return
-        ctx = self._make_context(dctx, typed_params)
-        try:
+        ctx = self._make_context(dctx, _extract_meta(params))
+
+        async def _inner() -> None:
+            if method == "notifications/initialized":
+                self._initialized = True
+                self.connection.initialized.set()
+                return
+            if not self._initialized:
+                logger.debug("dropped %s: received before initialization", method)
+                return
+            entry = self.server.get_notification_handler(method)
+            if entry is None:
+                logger.debug("no handler for notification %s", method)
+                return
+            # Absent wire params reach the handler as None, not an empty model
+            # (matches the existing `Server._handle_notification`). The
+            # empty-dict validate is a required-field check: a required-params
+            # model (e.g. ProgressNotificationParams) takes the
+            # malformed-params drop path instead of reaching a non-Optional
+            # handler as None.
+            try:
+                if params is None:
+                    entry.params_type.model_validate({}, by_name=False)
+                    typed_params = None
+                else:
+                    typed_params = entry.params_type.model_validate(params, by_name=False)
+            except ValidationError:
+                logger.warning("dropped %r: malformed params", method)
+                return
             await entry.handler(ctx, typed_params)
+
+        call = self._compose_server_middleware(ctx, method, params, _inner)
+        try:
+            await call()
         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).
+            # Top-level boundary: a notification handler (or middleware)
+            # 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). Middleware sees the raise out of `call_next()` first.
             logger.exception("notification handler for %r raised", method)
 
+    def _compose_server_middleware(
+        self,
+        ctx: ServerRequestContext[LifespanT, Any],
+        method: str,
+        params: Mapping[str, Any] | None,
+        inner: CallNext,
+    ) -> CallNext:
+        """Wrap `inner` in `Server.middleware`, outermost-first.
+
+        Shared by `_on_request` and `_on_notify` so the same middleware chain
+        observes every inbound message.
+        """
+        call = inner
+        for mw in reversed(self.server.middleware):
+            call = partial(mw, ctx, method, params, call)
+        return call
+
     def _make_context(
-        self, dctx: DispatchContext[TransportContext], typed_params: BaseModel | None
+        self, dctx: DispatchContext[TransportContext], meta: RequestParamsMeta | None
     ) -> ServerRequestContext[LifespanT, Any]:
-        meta = typed_params.meta if isinstance(typed_params, RequestParams) else None
         # TODO(maxisbey): remove for Context rework. Reads the SHTTP per-request
         # data off the raw `dctx.message_metadata` carrier; replace with the
         # per-transport context once that lands.
@@ -312,7 +359,7 @@ def _make_context(
             close_standalone_sse_stream=close_standalone_sse_stream,
         )
 
-    def _handle_initialize(self, params: Mapping[str, Any] | None) -> dict[str, Any]:
+    def _handle_initialize(self, params: Mapping[str, Any] | None) -> InitializeResult:
         init = InitializeRequestParams.model_validate(params or {}, by_name=False)
         self.connection.client_params = init
         requested = init.protocol_version
@@ -335,4 +382,4 @@ def _handle_initialize(self, params: Mapping[str, Any] | None) -> dict[str, Any]
             ),
             instructions=opts.instructions,
         )
-        return _dump_result(result)
+        return result

src/mcp/shared/jsonrpc_dispatcher.py

@@ -477,6 +477,9 @@ async def _dispatch_request(
             _progress_token=progress_token,
         )
         scope = anyio.CancelScope()
+        # 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)
         if req.method in self._inline_methods:
             # Spawn (so `sender_ctx` applies, matching the concurrent path) but