feat: ServerRunner middleware (two-tier) + _on_notify

ContextMiddleware is a Protocol[L] (contravariant) so Server[L].middleware:
list[ContextMiddleware[L]] is properly typed. App-specific middleware sees
ctx.lifespan: L; reusable middleware typed ContextMiddleware[object] registers
on any Server via contravariance. Context's covariance (previous PR3 commit)
makes Context[L, ST] <: Context[L, TransportContext] so the chain composes
without casts.

dispatch_middleware (DispatchMiddleware list on ServerRunner) wraps the raw
_on_request and sees everything including initialize/METHOD_NOT_FOUND.
server.middleware (ContextMiddleware) runs inside _on_request after
validation/ctx-build and wraps registered handlers only.

_on_notify routes notifications/initialized (sets the flag), drops
before-init and unknown methods, otherwise builds Context and calls the
registered handler.

11 tests over DirectDispatcher + a real lowlevel Server.

Author: maxisbey

src/mcp/server/context.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from typing import Any, Generic
+from typing import Any, Generic, Protocol
 
+from pydantic import BaseModel
 from typing_extensions import TypeVar
 
 from mcp.server._typed_request import TypedServerRequestMixin
@@ -81,3 +83,35 @@ async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *
         if meta:
             params["_meta"] = meta
         await self.notify("notifications/message", params)
+
+
+HandlerResult = BaseModel | dict[str, Any] | None
+"""What a request handler (or middleware) may return. `ServerRunner` serializes
+all three to a result dict."""
+
+CallNext = Callable[[], Awaitable[HandlerResult]]
+
+_MwLifespanT = TypeVar("_MwLifespanT", contravariant=True)
+
+
+class ContextMiddleware(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`.
+
+    `Server[L].middleware` holds `ContextMiddleware[L]`, so an app-specific
+    middleware sees `ctx.lifespan: L`. A reusable middleware (no app-specific
+    types) can be typed `ContextMiddleware[object]` — `Context` is covariant in
+    `LifespanT`, so it registers on any `Server[L]`.
+    """
+
+    async def __call__(
+        self,
+        ctx: Context[_MwLifespanT, TransportContext],
+        method: str,
+        params: BaseModel,
+        call_next: CallNext,
+    ) -> HandlerResult: ...

src/mcp/server/lowlevel/server.py

@@ -58,7 +58,7 @@ async def main():
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
 from mcp.server.auth.routes import build_resource_metadata_url, create_auth_routes, create_protected_resource_routes
 from mcp.server.auth.settings import AuthSettings
-from mcp.server.context import ServerRequestContext
+from mcp.server.context import ContextMiddleware, ServerRequestContext
 from mcp.server.experimental.request_context import Experimental
 from mcp.server.lowlevel.experimental import ExperimentalHandlers
 from mcp.server.models import InitializationOptions
@@ -199,6 +199,9 @@ def __init__(
         ] = {}
         self._experimental_handlers: ExperimentalHandlers[LifespanResultT] | None = None
         self._session_manager: StreamableHTTPSessionManager | None = None
+        # Context-tier middleware consumed by `ServerRunner`. Additive; the
+        # existing `run()` path ignores it.
+        self.middleware: list[ContextMiddleware[LifespanResultT]] = []
         logger.debug("Initializing server %r", name)
 
         # Populate internal handler dicts from on_* kwargs
@@ -256,11 +259,6 @@ def get_notification_handler(self, method: str) -> Callable[..., Awaitable[Any]]
         """Return the handler for a notification method, or ``None``."""
         return self._notification_handlers.get(method)
 
-    @property
-    def middleware(self) -> list[Any]:
-        """Context-tier middleware. Empty until the registry refactor adds registration."""
-        return []
-
     @property
     def connection_lifespan(self) -> None:
         """Per-connection lifespan. ``None`` until the registry refactor adds it."""

src/mcp/server/runner.py

@@ -17,17 +17,18 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass, field
+from functools import partial, reduce
 from typing import Any, Generic, Protocol, cast
 
 from pydantic import BaseModel
 from typing_extensions import TypeVar
 
 from mcp.server.connection import Connection
-from mcp.server.context import Context
+from mcp.server.context import CallNext, Context, ContextMiddleware
 from mcp.server.lowlevel.server import NotificationOptions
-from mcp.shared.dispatcher import DispatchContext, Dispatcher
+from mcp.shared.dispatcher import DispatchContext, Dispatcher, DispatchMiddleware, OnRequest
 from mcp.shared.exceptions import MCPError
 from mcp.shared.transport_context import TransportContext
 from mcp.types import (
@@ -51,7 +52,7 @@
     UnsubscribeRequestParams,
 )
 
-__all__ = ["ServerRegistry", "ServerRunner"]
+__all__ = ["CallNext", "ContextMiddleware", "ServerRegistry", "ServerRunner"]
 
 logger = logging.getLogger(__name__)
 
@@ -64,6 +65,7 @@
 `Context`-based handlers both fit during the transition.
 """
 
+
 _INIT_EXEMPT: frozenset[str] = frozenset({"ping"})
 
 # TODO: remove this lookup once `Server` stores (params_type, handler) in its
@@ -105,6 +107,9 @@ def name(self) -> str: ...
     @property
     def version(self) -> str | None: ...
 
+    @property
+    def middleware(self) -> Sequence[ContextMiddleware[Any]]: ...
+
     def get_request_handler(self, method: str) -> Handler | None: ...
     def get_notification_handler(self, method: str) -> Handler | None: ...
     def get_capabilities(
@@ -131,6 +136,7 @@ class ServerRunner(Generic[LifespanT, ServerTransportT]):
     lifespan_state: LifespanT
     has_standalone_channel: bool
     stateless: bool = False
+    dispatch_middleware: list[DispatchMiddleware] = field(default_factory=list[DispatchMiddleware])
 
     connection: Connection = field(init=False)
     _initialized: bool = field(init=False)
@@ -139,6 +145,16 @@ def __post_init__(self) -> None:
         self._initialized = self.stateless
         self.connection = Connection(self.dispatcher, has_standalone_channel=self.has_standalone_channel)
 
+    def _compose_on_request(self) -> OnRequest:
+        """Wrap `_on_request` in `dispatch_middleware`, outermost-first.
+
+        Dispatch-tier middleware sees raw ``(dctx, method, params) -> dict``
+        and wraps everything — initialize, METHOD_NOT_FOUND, validation
+        failures included. `run()` calls this once and hands the result to
+        `dispatcher.run()`.
+        """
+        return reduce(lambda h, mw: mw(h), reversed(self.dispatch_middleware), self._on_request)
+
     async def _on_request(
         self,
         dctx: DispatchContext[TransportContext],
@@ -162,8 +178,10 @@ async def _on_request(
         # it to INVALID_PARAMS.
         typed_params = params_type.model_validate(params or {})
         ctx = self._make_context(dctx, typed_params)
-        result = await handler(ctx, typed_params)
-        return _dump_result(result)
+        call: CallNext = partial(handler, ctx, typed_params)
+        for mw in reversed(self.server.middleware):
+            call = partial(mw, ctx, method, typed_params, call)
+        return _dump_result(await call())
 
     async def _on_notify(
         self,

tests/server/test_runner.py

@@ -8,6 +8,7 @@
 from typing import Any
 
 import anyio
+import anyio.lowlevel
 import pytest
 
 from mcp.server.connection import Connection
@@ -134,6 +135,143 @@ async def test_runner_unknown_method_raises_method_not_found(server: SrvT):
         tg.cancel_scope.cancel()
 
 
+@pytest.mark.anyio
+async def test_runner_on_notify_initialized_sets_flag_and_connection_event(server: SrvT):
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(server=server, dispatcher=server_d, lifespan_state=None, has_standalone_channel=True)
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(server_d.run, runner._on_request, runner._on_notify)
+        with anyio.fail_after(5):
+            await client.notify("notifications/initialized", None)
+            await runner.connection.initialized.wait()
+        assert runner._initialized is True
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_runner_on_notify_routes_to_registered_handler(server: SrvT):
+    seen: list[tuple[Any, Any]] = []
+
+    async def on_roots_changed(ctx: Any, params: Any) -> None:
+        seen.append((ctx, params))
+
+    server._notification_handlers["notifications/roots/list_changed"] = on_roots_changed
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(server=server, dispatcher=server_d, lifespan_state=None, has_standalone_channel=True)
+    runner._initialized = True
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(server_d.run, runner._on_request, runner._on_notify)
+        with anyio.fail_after(5):
+            await client.notify("notifications/roots/list_changed", None)
+            # DirectDispatcher delivers synchronously; one yield is enough.
+            await anyio.lowlevel.checkpoint()
+        assert len(seen) == 1
+        assert isinstance(seen[0][0], Context)
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_runner_on_notify_drops_before_init_and_unknown_methods(server: SrvT):
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(server=server, dispatcher=server_d, lifespan_state=None, has_standalone_channel=True)
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(server_d.run, runner._on_request, runner._on_notify)
+        with anyio.fail_after(5):
+            await client.notify("notifications/roots/list_changed", None)  # before init: dropped
+            await client.notify("notifications/initialized", None)
+            await client.notify("notifications/unknown", None)  # no handler: dropped
+        # No exception raised; both drops are silent.
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_runner_dispatch_middleware_wraps_everything_including_initialize(server: SrvT):
+    seen_methods: list[str] = []
+
+    def trace_mw(next_on_request: Any) -> Any:
+        async def wrapped(dctx: Any, method: str, params: Any) -> Any:
+            seen_methods.append(method)
+            return await next_on_request(dctx, method, params)
+
+        return wrapped
+
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(
+        server=server,
+        dispatcher=server_d,
+        lifespan_state=None,
+        has_standalone_channel=True,
+        dispatch_middleware=[trace_mw],
+    )
+    c_req, c_notify = echo_handlers(Recorder())
+    on_req = runner._compose_on_request()
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(server_d.run, on_req, runner._on_notify)
+        with anyio.fail_after(5):
+            await client.send_raw_request("initialize", _initialize_params())
+            await client.send_raw_request("tools/list", None)
+        assert seen_methods == ["initialize", "tools/list"]
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_runner_server_middleware_wraps_handlers_but_not_initialize(server: SrvT):
+    seen_methods: list[str] = []
+
+    async def ctx_mw(ctx: Any, method: str, params: Any, call_next: Any) -> Any:
+        seen_methods.append(method)
+        return await call_next()
+
+    server.middleware.append(ctx_mw)
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(server=server, dispatcher=server_d, lifespan_state=None, has_standalone_channel=True)
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(server_d.run, runner._on_request, runner._on_notify)
+        with anyio.fail_after(5):
+            await client.send_raw_request("initialize", _initialize_params())
+            await client.send_raw_request("ping", None)
+            await client.send_raw_request("tools/list", None)
+        # initialize NOT wrapped; ping and tools/list ARE wrapped.
+        assert seen_methods == ["ping", "tools/list"]
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_runner_server_middleware_runs_outermost_first(server: SrvT):
+    order: list[str] = []
+
+    def make_mw(tag: str) -> Any:
+        async def mw(ctx: Any, method: str, params: Any, call_next: Any) -> Any:
+            order.append(f"{tag}-in")
+            result = await call_next()
+            order.append(f"{tag}-out")
+            return result
+
+        return mw
+
+    server.middleware.extend([make_mw("a"), make_mw("b")])
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(server=server, dispatcher=server_d, lifespan_state=None, has_standalone_channel=True)
+    runner._initialized = True
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(server_d.run, runner._on_request, runner._on_notify)
+        with anyio.fail_after(5):
+            await client.send_raw_request("tools/list", None)
+        assert order == ["a-in", "b-in", "b-out", "a-out"]
+        tg.cancel_scope.cancel()
+
+
 @pytest.mark.anyio
 async def test_runner_stateless_skips_init_gate(server: SrvT):
     client, server_d = create_direct_dispatcher_pair()