Remove the dispatch-tier middleware hook (#2997)
This commit is contained in:
parent
3caa445c6c
commit
3945bdde11
4 changed files with 32 additions and 79 deletions
|
|
@ -14,9 +14,9 @@ the `Connection`, the driver tears it down.
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
from collections.abc import Awaitable, Mapping, Sequence
|
from collections.abc import Awaitable, Mapping
|
||||||
from dataclasses import KW_ONLY, dataclass
|
from dataclasses import KW_ONLY, dataclass
|
||||||
from functools import cached_property, partial, reduce
|
from functools import cached_property, partial
|
||||||
from typing import TYPE_CHECKING, Any, Generic, cast
|
from typing import TYPE_CHECKING, Any, Generic, cast
|
||||||
|
|
||||||
import anyio
|
import anyio
|
||||||
|
|
@ -45,7 +45,7 @@ from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, Server
|
||||||
from mcp.server.models import InitializationOptions
|
from mcp.server.models import InitializationOptions
|
||||||
from mcp.server.session import ServerSession
|
from mcp.server.session import ServerSession
|
||||||
from mcp.shared._stream_protocols import ReadStream, WriteStream
|
from mcp.shared._stream_protocols import ReadStream, WriteStream
|
||||||
from mcp.shared.dispatcher import DispatchContext, Dispatcher, DispatchMiddleware, OnNotify, OnRequest
|
from mcp.shared.dispatcher import DispatchContext, Dispatcher, OnNotify, OnRequest
|
||||||
from mcp.shared.exceptions import MCPError
|
from mcp.shared.exceptions import MCPError
|
||||||
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
|
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
|
||||||
from mcp.shared.message import ServerMessageMetadata, SessionMessage
|
from mcp.shared.message import ServerMessageMetadata, SessionMessage
|
||||||
|
|
@ -141,22 +141,10 @@ class ServerRunner(Generic[LifespanT]):
|
||||||
_: KW_ONLY
|
_: KW_ONLY
|
||||||
init_options: InitializationOptions | None = None
|
init_options: InitializationOptions | None = None
|
||||||
"""`InitializeResult` payload. Defaults to `server.create_initialization_options()`."""
|
"""`InitializeResult` payload. Defaults to `server.create_initialization_options()`."""
|
||||||
dispatch_middleware: Sequence[DispatchMiddleware] = ()
|
|
||||||
"""Raw dispatch-tier wrappers `(dctx, method, params) -> dict`, applied outermost-first
|
|
||||||
around `_on_request`. Empty by default; OpenTelemetry tracing lives at the context tier
|
|
||||||
(`OpenTelemetryMiddleware`, seeded into `Server.middleware`)."""
|
|
||||||
|
|
||||||
@cached_property
|
@cached_property
|
||||||
def on_request(self) -> OnRequest:
|
def on_request(self) -> OnRequest:
|
||||||
"""`_on_request` wrapped in `dispatch_middleware`, outermost-first.
|
return self._on_request
|
||||||
|
|
||||||
Dispatch-tier middleware sees raw `(dctx, method, params) -> dict` and
|
|
||||||
wraps everything - initialize, METHOD_NOT_FOUND, validation failures
|
|
||||||
included.
|
|
||||||
"""
|
|
||||||
return reduce(
|
|
||||||
lambda handler, middleware: middleware(handler), reversed(self.dispatch_middleware), self._on_request
|
|
||||||
)
|
|
||||||
|
|
||||||
@cached_property
|
@cached_property
|
||||||
def on_notify(self) -> OnNotify:
|
def on_notify(self) -> OnNotify:
|
||||||
|
|
|
||||||
|
|
@ -29,7 +29,6 @@ from mcp.shared.transport_context import TransportContext
|
||||||
__all__ = [
|
__all__ = [
|
||||||
"CallOptions",
|
"CallOptions",
|
||||||
"DispatchContext",
|
"DispatchContext",
|
||||||
"DispatchMiddleware",
|
|
||||||
"Dispatcher",
|
"Dispatcher",
|
||||||
"OnNotify",
|
"OnNotify",
|
||||||
"OnRequest",
|
"OnRequest",
|
||||||
|
|
@ -185,9 +184,6 @@ OnRequest = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any]
|
||||||
OnNotify = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any] | None], Awaitable[None]]
|
OnNotify = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any] | None], Awaitable[None]]
|
||||||
"""Handler for inbound notifications: `(ctx, method, params)`."""
|
"""Handler for inbound notifications: `(ctx, method, params)`."""
|
||||||
|
|
||||||
DispatchMiddleware = Callable[[OnRequest], OnRequest]
|
|
||||||
"""Wraps an `OnRequest` to produce another `OnRequest`. Applied outermost-first."""
|
|
||||||
|
|
||||||
|
|
||||||
class Dispatcher(Outbound, Protocol[TransportT_co]):
|
class Dispatcher(Outbound, Protocol[TransportT_co]):
|
||||||
"""A duplex request/notification channel with call-return semantics.
|
"""A duplex request/notification channel with call-return semantics.
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,7 @@ so these tests assert against the default-configured server rather than appendin
|
||||||
the middleware by hand.
|
the middleware by hand.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
from dataclasses import replace
|
from dataclasses import replace
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
|
@ -20,6 +21,7 @@ from mcp_types import (
|
||||||
ListToolsResult,
|
ListToolsResult,
|
||||||
NotificationParams,
|
NotificationParams,
|
||||||
PaginatedRequestParams,
|
PaginatedRequestParams,
|
||||||
|
RequestParamsMeta,
|
||||||
Tool,
|
Tool,
|
||||||
)
|
)
|
||||||
from opentelemetry.trace import SpanKind, StatusCode
|
from opentelemetry.trace import SpanKind, StatusCode
|
||||||
|
|
@ -166,16 +168,17 @@ async def test_notification_span_omits_request_id(server: SrvT, spans: SpanCaptu
|
||||||
assert "jsonrpc.request.id" not in span.attributes
|
assert "jsonrpc.request.id" not in span.attributes
|
||||||
|
|
||||||
|
|
||||||
def _ambient_span(call_next: Any) -> Any:
|
def _ambient(rewrite_meta: Callable[[RequestParamsMeta | None], RequestParamsMeta | None]) -> Any:
|
||||||
"""Dispatch-tier wrapper that opens an ambient SERVER span around the whole
|
"""A middleware placed outside `OpenTelemetryMiddleware` (head of
|
||||||
request, so the context-tier span has a current span to nest under when the
|
`Server.middleware`) that opens an ambient SERVER span around the request
|
||||||
inbound message carries no traceparent."""
|
and rewrites `ctx.meta` via `rewrite_meta`, so the context-tier span sees
|
||||||
|
the no-traceparent path yet has a current span to nest under."""
|
||||||
|
|
||||||
async def wrapped(dctx: Any, method: str, params: dict[str, Any] | None) -> Any:
|
async def middleware(ctx: Ctx, call_next: CallNext) -> Any:
|
||||||
with otel_span("ambient", kind=SpanKind.SERVER):
|
with otel_span("ambient", kind=SpanKind.SERVER):
|
||||||
return await call_next(dctx, method, params)
|
return await call_next(replace(ctx, meta=rewrite_meta(ctx.meta)))
|
||||||
|
|
||||||
return wrapped
|
return middleware
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
|
|
@ -184,16 +187,10 @@ async def test_nests_under_ambient_span_when_no_traceparent(server: SrvT, spans:
|
||||||
parent to the ambient current span rather than become an orphan root.
|
parent to the ambient current span rather than become an orphan root.
|
||||||
SDK-defined: SEP-414 only covers the traceparent-present case."""
|
SDK-defined: SEP-414 only covers the traceparent-present case."""
|
||||||
|
|
||||||
def strip_meta(call_next: Any) -> Any:
|
# The in-process client always injects `_meta.traceparent`; drop it so the
|
||||||
# The in-process client always injects `_meta.traceparent`; strip it so
|
# span sees the no-carrier path.
|
||||||
# the span sees the no-carrier path.
|
server.middleware.insert(0, _ambient(lambda _meta: None))
|
||||||
async def wrapped(dctx: Any, method: str, params: dict[str, Any] | None) -> Any:
|
async with connected_runner(server) as (client, _):
|
||||||
stripped = {k: v for k, v in (params or {}).items() if k != "_meta"}
|
|
||||||
return await call_next(dctx, method, stripped or None)
|
|
||||||
|
|
||||||
return wrapped
|
|
||||||
|
|
||||||
async with connected_runner(server, dispatch_middleware=[_ambient_span, strip_meta]) as (client, _):
|
|
||||||
spans.clear()
|
spans.clear()
|
||||||
await client.send_raw_request("tools/list", None)
|
await client.send_raw_request("tools/list", None)
|
||||||
server_spans = [s for s in spans.finished() if s.kind == SpanKind.SERVER]
|
server_spans = [s for s in spans.finished() if s.kind == SpanKind.SERVER]
|
||||||
|
|
@ -213,14 +210,8 @@ async def test_nests_under_ambient_span_when_meta_lacks_traceparent(server: SrvT
|
||||||
would orphan the span; the middleware must fall through to ambient
|
would orphan the span; the middleware must fall through to ambient
|
||||||
parenting just as if `_meta` were absent."""
|
parenting just as if `_meta` were absent."""
|
||||||
|
|
||||||
def replace_meta(call_next: Any) -> Any:
|
server.middleware.insert(0, _ambient(lambda _meta: {"progressToken": "tok"}))
|
||||||
async def wrapped(dctx: Any, method: str, params: dict[str, Any] | None) -> Any:
|
async with connected_runner(server) as (client, _):
|
||||||
rewritten = {**(params or {}), "_meta": {"progressToken": "tok"}}
|
|
||||||
return await call_next(dctx, method, rewritten)
|
|
||||||
|
|
||||||
return wrapped
|
|
||||||
|
|
||||||
async with connected_runner(server, dispatch_middleware=[_ambient_span, replace_meta]) as (client, _):
|
|
||||||
spans.clear()
|
spans.clear()
|
||||||
await client.send_raw_request("tools/list", None)
|
await client.send_raw_request("tools/list", None)
|
||||||
server_spans = [s for s in spans.finished() if s.kind == SpanKind.SERVER]
|
server_spans = [s for s in spans.finished() if s.kind == SpanKind.SERVER]
|
||||||
|
|
|
||||||
|
|
@ -9,7 +9,7 @@ behaviour under test. Driver tests (`serve_connection`, `serve_one`,
|
||||||
|
|
||||||
from collections.abc import AsyncIterator, Mapping
|
from collections.abc import AsyncIterator, Mapping
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field, replace
|
||||||
from functools import partial
|
from functools import partial
|
||||||
from typing import Any, cast
|
from typing import Any, cast
|
||||||
|
|
||||||
|
|
@ -48,7 +48,7 @@ from mcp.server.runner import (
|
||||||
serve_one,
|
serve_one,
|
||||||
)
|
)
|
||||||
from mcp.server.session import ServerSession
|
from mcp.server.session import ServerSession
|
||||||
from mcp.shared.dispatcher import CallOptions, DispatchContext, DispatchMiddleware, OnRequest
|
from mcp.shared.dispatcher import CallOptions
|
||||||
from mcp.shared.exceptions import MCPError
|
from mcp.shared.exceptions import MCPError
|
||||||
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
|
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
|
||||||
from mcp.shared.message import MessageMetadata
|
from mcp.shared.message import MessageMetadata
|
||||||
|
|
@ -91,7 +91,6 @@ async def connected_runner(
|
||||||
*,
|
*,
|
||||||
initialized: bool = True,
|
initialized: bool = True,
|
||||||
init_options: InitializationOptions | None = None,
|
init_options: InitializationOptions | None = None,
|
||||||
dispatch_middleware: list[DispatchMiddleware] | None = None,
|
|
||||||
connection: Connection | None = None,
|
connection: Connection | None = None,
|
||||||
) -> AsyncIterator[tuple[JSONRPCDispatcher[TransportContext], ServerRunner[dict[str, Any]]]]:
|
) -> AsyncIterator[tuple[JSONRPCDispatcher[TransportContext], ServerRunner[dict[str, Any]]]]:
|
||||||
"""Yield `(client, runner)` running over an in-memory JSON-RPC dispatcher pair.
|
"""Yield `(client, runner)` running over an in-memory JSON-RPC dispatcher pair.
|
||||||
|
|
@ -116,7 +115,6 @@ async def connected_runner(
|
||||||
connection=connection,
|
connection=connection,
|
||||||
lifespan_state={},
|
lifespan_state={},
|
||||||
init_options=init_options,
|
init_options=init_options,
|
||||||
dispatch_middleware=dispatch_middleware or [],
|
|
||||||
)
|
)
|
||||||
c_req, c_notify = echo_handlers(Recorder())
|
c_req, c_notify = echo_handlers(Recorder())
|
||||||
body_exc: BaseException | None = None
|
body_exc: BaseException | None = None
|
||||||
|
|
@ -511,8 +509,8 @@ async def test_runner_absent_wire_params_reaches_request_handler_as_defaults_mod
|
||||||
"""A request with no `params` member on the wire reaches the handler as
|
"""A request with no `params` member on the wire reaches the handler as
|
||||||
the params model with its defaults, never `None`.
|
the params model with its defaults, never `None`.
|
||||||
|
|
||||||
The in-SDK client always attaches `_meta`, so a dispatch middleware
|
The in-SDK client always attaches `_meta`, so a middleware rewrites
|
||||||
forwards `params=None` to model what an external client sends.
|
`ctx.params` to `None` to model what an external client sends.
|
||||||
"""
|
"""
|
||||||
seen: list[PaginatedRequestParams | None] = []
|
seen: list[PaginatedRequestParams | None] = []
|
||||||
|
|
||||||
|
|
@ -520,14 +518,12 @@ async def test_runner_absent_wire_params_reaches_request_handler_as_defaults_mod
|
||||||
seen.append(params)
|
seen.append(params)
|
||||||
return ListToolsResult(tools=[])
|
return ListToolsResult(tools=[])
|
||||||
|
|
||||||
def drop_params(next_on_request: OnRequest) -> OnRequest:
|
async def drop_params(ctx: Ctx, call_next: Any) -> Any:
|
||||||
async def wrapped(dctx: DispatchContext[Any], method: str, params: Any) -> dict[str, Any]:
|
return await call_next(replace(ctx, params=None) if ctx.method == "tools/list" else ctx)
|
||||||
return await next_on_request(dctx, method, None if method == "tools/list" else params)
|
|
||||||
|
|
||||||
return wrapped
|
|
||||||
|
|
||||||
server: SrvT = Server(name="s", on_list_tools=list_tools)
|
server: SrvT = Server(name="s", on_list_tools=list_tools)
|
||||||
async with connected_runner(server, dispatch_middleware=[drop_params]) as (client, _):
|
server.middleware.append(drop_params)
|
||||||
|
async with connected_runner(server) as (client, _):
|
||||||
await client.send_raw_request("tools/list", None)
|
await client.send_raw_request("tools/list", None)
|
||||||
assert seen == [PaginatedRequestParams()]
|
assert seen == [PaginatedRequestParams()]
|
||||||
|
|
||||||
|
|
@ -543,15 +539,13 @@ async def test_runner_absent_wire_params_for_required_params_custom_method_is_in
|
||||||
async def greet(ctx: Ctx, params: GreetParams) -> dict[str, Any]:
|
async def greet(ctx: Ctx, params: GreetParams) -> dict[str, Any]:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def drop_params(next_on_request: OnRequest) -> OnRequest:
|
async def drop_params(ctx: Ctx, call_next: Any) -> Any:
|
||||||
async def wrapped(dctx: DispatchContext[Any], method: str, params: Any) -> dict[str, Any]:
|
return await call_next(replace(ctx, params=None) if ctx.method == "custom/greet" else ctx)
|
||||||
return await next_on_request(dctx, method, None if method == "custom/greet" else params)
|
|
||||||
|
|
||||||
return wrapped
|
|
||||||
|
|
||||||
server: SrvT = Server(name="s")
|
server: SrvT = Server(name="s")
|
||||||
server.add_request_handler("custom/greet", GreetParams, greet)
|
server.add_request_handler("custom/greet", GreetParams, greet)
|
||||||
async with connected_runner(server, dispatch_middleware=[drop_params]) as (client, _):
|
server.middleware.append(drop_params)
|
||||||
|
async with connected_runner(server) as (client, _):
|
||||||
with pytest.raises(MCPError) as exc:
|
with pytest.raises(MCPError) as exc:
|
||||||
await client.send_raw_request("custom/greet", {"name": "x"})
|
await client.send_raw_request("custom/greet", {"name": "x"})
|
||||||
assert exc.value.error.code == INVALID_PARAMS
|
assert exc.value.error.code == INVALID_PARAMS
|
||||||
|
|
@ -574,22 +568,6 @@ async def test_runner_on_notify_drops_before_init_and_unknown_methods(server: Sr
|
||||||
assert seen == [NotificationParams()] # only the post-init one reached the handler
|
assert seen == [NotificationParams()] # only the post-init one reached the handler
|
||||||
|
|
||||||
|
|
||||||
@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
|
|
||||||
|
|
||||||
async with connected_runner(server, dispatch_middleware=[trace_mw]) as (client, _):
|
|
||||||
await client.send_raw_request("tools/list", None)
|
|
||||||
assert seen_methods == ["initialize", "tools/list"]
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_runner_server_middleware_wraps_every_request_including_initialize(server: SrvT):
|
async def test_runner_server_middleware_wraps_every_request_including_initialize(server: SrvT):
|
||||||
seen: list[tuple[str, Any]] = []
|
seen: list[tuple[str, Any]] = []
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue