Expose the middleware chain on MCPServer and stop sending unrequested change notifications (#3201)
This commit is contained in:
Max
GitHub
parent
528e366558
commit
c9c431b71a
13 changed files with 234 additions and 28 deletions
|
|
@ -5,12 +5,14 @@ A **middleware** is one async function that wraps every message your server rece
|
|||
You write it as `async (ctx, call_next)` and append it to `server.middleware`. That is the whole API.
|
||||
|
||||
!!! warning
|
||||
`Server.middleware` is marked **provisional** in the source. The signature and semantics are
|
||||
expected to change before v2 is final. Use it to *observe*: timing, logging, tracing.
|
||||
Do not make it the foundation your server stands on.
|
||||
The middleware list is marked **provisional** in the source. The signature and semantics are
|
||||
expected to change before v2 is final. Use it to *observe* (timing, logging, tracing) and to
|
||||
*refuse* messages; do not make it the foundation your server stands on.
|
||||
|
||||
This is a **low-level `Server`** feature. `MCPServer` does not expose a middleware list.
|
||||
If `Server(name, on_call_tool=...)` is new to you, read **[The low-level Server](low-level-server.md)** first.
|
||||
`MCPServer` takes the list at construction (`MCPServer(name, middleware=[...])`) and exposes it as
|
||||
`mcp.middleware`; the low-level `Server` exposes the same list as `server.middleware`. The example
|
||||
below uses the low-level `Server`; if `Server(name, on_call_tool=...)` is new to you, read
|
||||
**[The low-level Server](low-level-server.md)** first.
|
||||
|
||||
## A timing middleware
|
||||
|
||||
|
|
@ -57,7 +59,10 @@ In increasing order of how much you should hesitate:
|
|||
|
||||
* **Observe.** Time it, count it, log it. The example above.
|
||||
* **Refuse.** Raise an `MCPError` *instead of* calling `call_next(ctx)` and that one message is
|
||||
answered with a JSON-RPC error. The connection stays up; the next message goes through.
|
||||
answered with a JSON-RPC error. The connection stays up; the next message goes through. This is
|
||||
how a server gates `subscriptions/listen` per caller:
|
||||
**[Deciding who may watch](../handlers/subscriptions.md#deciding-who-may-watch)** on the
|
||||
Subscriptions page walks through it.
|
||||
* **Rewrite.** `ctx` is a dataclass: `await call_next(dataclasses.replace(ctx, params=...))`
|
||||
hands the rest of the chain different params than the client sent. Never do this to
|
||||
`initialize`: the result the client gets back is built from your rewritten params, but the
|
||||
|
|
@ -98,8 +103,8 @@ don't think about it. It is a no-op until you install an exporter, and it has it
|
|||
|
||||
## Recap
|
||||
|
||||
* A middleware is `async (ctx, call_next) -> result`, appended to `server.middleware` on the
|
||||
low-level `Server`.
|
||||
* A middleware is `async (ctx, call_next) -> result`, passed as `MCPServer(middleware=[...])` (or
|
||||
appended to `mcp.middleware`), and appended to `server.middleware` on the low-level `Server`.
|
||||
* It wraps **every** inbound message (`server/discover`, `initialize`, requests, notifications,
|
||||
unknown methods) and runs outermost-first.
|
||||
* `ctx.request_id is None` is how you tell a notification from a request.
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ Duplicate events waiting to be consumed collapse into one, and refetching still
|
|||
|
||||
Two more properties of the handle:
|
||||
|
||||
* `sub.honored` is the filter the server acknowledged: a `SubscriptionFilter` with the fields you passed, read as attributes (`sub.honored.prompts_list_changed`). `MCPServer` honors every kind you ask for, so it echoes your request back. A server that narrows the filter (see the [filter warning](../handlers/subscriptions.md#only-what-was-asked-for) on the server page) acknowledges less, and an honored kind may still never fire.
|
||||
* `sub.honored` is the filter the server acknowledged: a `SubscriptionFilter` with the fields you passed, read as attributes (`sub.honored.prompts_list_changed`). `MCPServer` honors every kind you ask for, so it echoes your request back. A server that supports fewer kinds acknowledges less, and an honored kind may still never fire. A server may also refuse the whole request rather than acknowledge it (see [Deciding who may watch](../handlers/subscriptions.md#deciding-who-may-watch) on the server page), which surfaces as the request's error.
|
||||
* `sub.subscription_id` is the listen request's id, the one stamped on every frame of this stream. Several subscriptions can be open at once, each demultiplexed by its own id.
|
||||
|
||||
## Watching without blocking
|
||||
|
|
|
|||
|
|
@ -43,14 +43,22 @@ Two things the stream is *not*:
|
|||
* **It is not a replay log.** A dropped stream is gone, and events published while nobody was connected are not queued. Clients re-listen and refetch.
|
||||
* **It is not the 2025 path.** Clients that called `resources/subscribe` are served by `ctx.session.send_resource_updated(uri)`. The `notify_*` methods reach `subscriptions/listen` streams only.
|
||||
|
||||
!!! warning
|
||||
Don't publish sensitive per-user URIs through `notify_resource_updated` on a multi-tenant
|
||||
server. Any client may name any URI in its filter, and `MCPServer` honors it. The exposure
|
||||
is narrow but real: a subscriber learns that a URI it can guess changed, and when. It never
|
||||
learns content, and it cannot probe what exists, because an unknown URI is honored too and
|
||||
simply never fires. To narrow the filter per client today, serve the method with your own
|
||||
handler on the low-level `Server` and acknowledge a smaller filter than the client asked
|
||||
for; the acknowledgment is how the client learns what it actually got.
|
||||
## Deciding who may watch
|
||||
|
||||
By default every requested kind and URI is honored: any caller may watch any URI you publish. Nothing consults your read handler, because nobody is reading — a caller your `files://{name}` handler would turn away can still open a stream on `files://payroll.csv` and learn that it changed, and when. It never learns content, and it cannot probe what exists, because an unknown URI is honored too and simply never fires. Narrow but real, so gate it before you publish per-user URIs from a multi-tenant server.
|
||||
|
||||
The gate is a middleware. It sees the `subscriptions/listen` request before the SDK acknowledges it and refuses when the caller asks for anything they may not read:
|
||||
|
||||
```python title="server.py" hl_lines="19-26 29"
|
||||
--8<-- "docs_src/subscriptions/tutorial006.py"
|
||||
```
|
||||
|
||||
* `ctx.params` is the raw request, so the middleware validates it into `SubscriptionsListenRequestParams` itself and reads the filter the client asked for.
|
||||
* Refusal is a raised `MCPError` before `call_next(ctx)`: the client gets that error and no stream, and the connection carries on. Keep the message uniform, naming no URI, so a refusal never confirms which URIs are protected.
|
||||
* One `can_access(user, uri)` answers both questions. The resource handler asks it on `resources/read`; the middleware asks it on `subscriptions/listen`. Swap the table for a database or your RBAC system and both stay in step.
|
||||
* The decision holds for the stream's lifetime. There is no per-event re-check, so if a caller's access can lapse mid-stream (an expiring token), end that caller's connection when it does.
|
||||
|
||||
The full middleware contract, including what else it wraps and why it is marked provisional, is on **[Middleware](../advanced/middleware.md)**.
|
||||
|
||||
## The client end
|
||||
|
||||
|
|
|
|||
|
|
@ -742,7 +742,7 @@ This parameter was redundant because the SSE transport already handles sub-path
|
|||
|
||||
### Transport-specific parameters moved from MCPServer constructor to run()/app methods
|
||||
|
||||
Transport-specific parameters have been moved off the `MCPServer` constructor and onto `run()`, `sse_app()`, and `streamable_http_app()`, so transport configuration is passed when starting or building the server. The rest of the constructor is unchanged: identity (`name`, `instructions`, `website_url`, `icons`, plus the newly added positional `title`, `description`, and `version` covered [above](#mcpserver-constructor-title-description-and-version-added-to-the-positional-parameters)), authentication (`auth`, `token_verifier`, `auth_server_provider`), `lifespan`, `dependencies`, `tools`, `debug`, `log_level`, and the `warn_on_duplicate_*` flags; the new keyword-only parameters (`resources`, `extensions`, `resource_security`, `request_state_security`, `cache_hints`, `subscriptions`) are additive.
|
||||
Transport-specific parameters have been moved off the `MCPServer` constructor and onto `run()`, `sse_app()`, and `streamable_http_app()`, so transport configuration is passed when starting or building the server. The rest of the constructor is unchanged: identity (`name`, `instructions`, `website_url`, `icons`, plus the newly added positional `title`, `description`, and `version` covered [above](#mcpserver-constructor-title-description-and-version-added-to-the-positional-parameters)), authentication (`auth`, `token_verifier`, `auth_server_provider`), `lifespan`, `dependencies`, `tools`, `debug`, `log_level`, and the `warn_on_duplicate_*` flags; the new keyword-only parameters (`resources`, `extensions`, `resource_security`, `request_state_security`, `cache_hints`, `subscriptions`, `middleware`) are additive.
|
||||
|
||||
**Parameters moved:**
|
||||
|
||||
|
|
@ -1718,7 +1718,7 @@ The helpers keep their v1 signatures, so calls through `ctx.session` are source-
|
|||
Behavior changes:
|
||||
|
||||
- **A new `ServerSession` proxy is built for every inbound message.** In v1 one `ServerSession` lived for the whole connection, and servers commonly keyed per-client state on `ctx.session` identity (a `WeakKeyDictionary[ServerSession, ...]`, `id(ctx.session)`, a set of captured sessions to notify later). In v2 each request and notification gets a fresh proxy over the same connection, so those idioms silently misbehave: a session-keyed dict never finds an earlier key, and a broadcast set grows by one entry per request, sending duplicates. Key on something connection-stable instead — on stateful streamable HTTP the `mcp-session-id` request header names the transport session (read it via `ctx.headers` on `MCPServer` or `ctx.request.headers` in a lowlevel handler); on stdio there is one connection per process. The per-connection object the proxies share is `mcp.server.connection.Connection` (`state`, `session_id`, `exit_stack`), which is not currently reachable from `ctx`.
|
||||
- **A captured `ctx.session` stays usable after the handler returns.** The proxy holds the connection, not the request, so a background task can keep calling `send_resource_updated()` / `send_tool_list_changed()` on it while the client stays connected; with `related_request_id` omitted these ride the standalone stream as in v1. A request-scoped send is only meaningful while that request is in flight — once the handler returns, that stream is closed and the message is dropped with a debug log.
|
||||
- **A captured `ctx.session` stays usable after the handler returns.** The proxy holds the connection, not the request, so a background task can keep calling `send_resource_updated()` / `send_tool_list_changed()` on it while the client stays connected; with `related_request_id` omitted these ride the standalone stream as in v1 — except on a 2026-era connection, where change notifications are dropped and belong on the subscription bus instead ([change notifications travel only on `subscriptions/listen` streams](#change-notifications-travel-only-on-subscriptionslisten-streams)). A request-scoped send is only meaningful while that request is in flight — once the handler returns, that stream is closed and the message is dropped with a debug log.
|
||||
- **Notifications after the connection has closed are dropped instead of raising.** In v1 the notification helpers raised `anyio.ClosedResourceError`/`anyio.BrokenResourceError` on a dead connection, and broadcast loops used that exception to prune sessions. In v2 the send returns normally (the drop is debug-logged), so probe with a request instead: `await session.send_ping()` raises `MCPError` once the connection has closed. On a 2026-07-28 connection, though, every server-initiated request raises `NoBackChannelError` (an `MCPError`) regardless, so a ping is a liveness probe only on connections negotiated at 2025-11-25 or earlier.
|
||||
|
||||
`ServerSession` is normally constructed for you by `Server.run()` and reached via `ctx.session` in handlers, so beyond the behavior changes above, most servers are unaffected. If you were constructing or subclassing it directly:
|
||||
|
|
@ -2804,7 +2804,7 @@ rest of this guide stays focused on the v1-to-v2 upgrade itself.
|
|||
|
||||
The 2026-07-28 protocol has no server-initiated requests, so a handler that reaches back to the client mid-request — `ctx.elicit()`, `ctx.elicit_url()`, `ctx.session.create_message()`, `ctx.session.list_roots()`, or any other `ServerSession` request helper — raises `NoBackChannelError` on such a connection instead of sending. An in-process `Client(server)` negotiates 2026-07-28 by default (see [`Client` defaults to `mode='auto'`](#client-defaults-to-modeauto)), so the first smoke test of an unchanged v1 sampling or elicitation tool fails, and setting `sampling_callback=` / `elicitation_callback=` on the client changes nothing because no request ever reaches the client.
|
||||
|
||||
`NoBackChannelError` lives in `mcp.shared.exceptions` and subclasses `MCPError` (code `-32600`, message `Cannot send '<method>': this transport context has no back-channel for server-initiated requests.`). Raised inside an `@mcp.tool()` it reaches the client as a top-level JSON-RPC error, not `CallToolResult(is_error=True)` — see [`MCPError` raised from an `@mcp.tool()` handler now surfaces as a JSON-RPC error](#mcperror-raised-from-an-mcptool-handler-now-surfaces-as-a-json-rpc-error) — and the [Troubleshooting](troubleshooting.md) page walks through the client-side traceback. The same exception is raised on a legacy session against a `stateless_http=True` server, and on the request-scoped channel of a stateful legacy session against a `json_response=True` server (a JSON body carries exactly one response, so a mid-request `ctx.elicit()` cannot ride it; the session's standalone `GET` stream still carries unrelated messages) — both places v1 dropped the message and stalled ([`Server.run()` no longer takes a `stateless` flag](#serverrun-no-longer-takes-a-stateless-flag)). Notifications never raise it: `send_log_message()`, `send_tool_list_changed()`, and the other notification helpers are dropped with a debug log where no channel exists, and `UrlElicitationRequiredError` from a tool is unaffected (it is an error response, not a request).
|
||||
`NoBackChannelError` lives in `mcp.shared.exceptions` and subclasses `MCPError` (code `-32600`, message `Cannot send '<method>': this transport context has no back-channel for server-initiated requests.`). Raised inside an `@mcp.tool()` it reaches the client as a top-level JSON-RPC error, not `CallToolResult(is_error=True)` — see [`MCPError` raised from an `@mcp.tool()` handler now surfaces as a JSON-RPC error](#mcperror-raised-from-an-mcptool-handler-now-surfaces-as-a-json-rpc-error) — and the [Troubleshooting](troubleshooting.md) page walks through the client-side traceback. The same exception is raised on a legacy session against a `stateless_http=True` server, and on the request-scoped channel of a stateful legacy session against a `json_response=True` server (a JSON body carries exactly one response, so a mid-request `ctx.elicit()` cannot ride it; the session's standalone `GET` stream still carries unrelated messages) — both places v1 dropped the message and stalled ([`Server.run()` no longer takes a `stateless` flag](#serverrun-no-longer-takes-a-stateless-flag)). Notifications never raise it: `send_log_message()`, `send_tool_list_changed()`, and the other notification helpers are dropped with a debug log where no channel exists (and the change-notification helpers are dropped on every 2026-era connection, channel or not — see [change notifications travel only on `subscriptions/listen` streams](#change-notifications-travel-only-on-subscriptionslisten-streams)), and `UrlElicitationRequiredError` from a tool is unaffected (it is an error response, not a request).
|
||||
|
||||
Two ways to migrate:
|
||||
|
||||
|
|
@ -2856,6 +2856,12 @@ async with Client(server, logging_callback=on_log, log_level="info") as client:
|
|||
|
||||
`log_level=None` (the default) means no opt-in — a `logging_callback` alone is not one — and a single request can override the client-wide default by supplying the key in its own `meta=` (e.g. `meta={LOG_LEVEL_META_KEY: "debug"}` from `mcp_types`). The opt-in is what the spec calls for on 2026-era servers generally, not just this SDK's. Because 2026 log delivery is request-scoped by construction, `related_request_id` on `send_log_message` no longer selects the standalone stream there: whatever is delivered rides the requesting stream.
|
||||
|
||||
### Change notifications travel only on `subscriptions/listen` streams
|
||||
|
||||
On a 2026-07-28 connection, `notifications/tools/list_changed`, `notifications/prompts/list_changed`, `notifications/resources/list_changed`, and `notifications/resources/updated` reach a client only through a `subscriptions/listen` stream it opened — the spec forbids sending a notification type a subscription did not request. The v1-style session helpers (`ctx.session.send_tool_list_changed()`, `send_prompt_list_changed()`, `send_resource_list_changed()`, `send_resource_updated(uri)`) push a bare copy onto the connection's standalone channel instead, so on such a connection they are dropped with a debug log: silently on streamable HTTP (there is no standalone channel), and on stdio, where earlier v2 releases wrote the bare notification to the shared pipe, it is now dropped too. On pre-2026 connections the helpers behave as in v1.
|
||||
|
||||
Migrate to publishing on the subscription bus, which stamps and filters per stream: `await ctx.notify_tools_changed()`, `notify_prompts_changed()`, `notify_resources_changed()`, and `notify_resource_updated(uri)` on `MCPServer`'s `Context`, or `await bus.publish(...)` on a low-level `Server`'s own `SubscriptionBus` — see [Subscriptions](handlers/subscriptions.md). A stream only ever receives the kinds and URIs the server acknowledged for it; to gate per caller which subscriptions may be opened, refuse `subscriptions/listen` in a middleware (`MCPServer(middleware=[...])`), covered on the same page.
|
||||
|
||||
### Servers validate `Mcp-Param-*` headers against the request body ([SEP-2243](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2243))
|
||||
|
||||
On the 2026-07-28 Streamable HTTP path, a `tools/call` whose tool declares `x-mcp-header` annotations is validated before dispatch — each annotated argument and its mirroring `Mcp-Param-*` header must be present together and agree (after base64-sentinel decoding; integers compare numerically), or absent together. A violation is rejected with HTTP 400 and JSON-RPC error `-32020` (`HeaderMismatch`), as the spec requires. A client that sends an annotated argument *without* its header — for example one that never listed the tool — is therefore rejected instead of silently served; the spec's recovery is to re-list and retry. On the client side, `ClientSession.call_tool` emits these headers automatically for annotated arguments of any tool it has listed; list the tool first, and note that pre-2026 connections and non-HTTP transports never emit them.
|
||||
|
|
|
|||
|
|
@ -107,7 +107,7 @@ Tools, resources, prompts, structured output, progress, errors: none of them car
|
|||
There is exactly one thing left, and it is **change notifications**, because the two eras listen on different pipes:
|
||||
|
||||
* A `2026-07-28` client opens a `subscriptions/listen` stream and reads the subscriptions bus. `ctx.notify_resource_updated()` (and `notify_tools_changed()`, `notify_prompts_changed()`, `notify_resources_changed()`) publish there, and *only* there. **[Subscriptions](../handlers/subscriptions.md)** is that page.
|
||||
* A legacy client reads the standalone stream its session keeps open. `ctx.session.send_resource_updated()` (and `send_tool_list_changed()` and friends) write to the *connection* that carried the request: for a legacy session, that is its standalone stream. For a modern HTTP request there is no such channel, and the notification is quietly dropped.
|
||||
* A legacy client reads the standalone stream its session keeps open. `ctx.session.send_resource_updated()` (and `send_tool_list_changed()` and friends) write to the *connection* that carried the request: for a legacy session, that is its standalone stream. A modern connection has no place for it: over HTTP there is no such channel, and over stdio the four change-notification kinds ride `subscriptions/listen` streams only, so on a modern connection the notification is quietly dropped.
|
||||
|
||||
Over HTTP, neither call reaches the other era's clients. To tell everyone, call both:
|
||||
|
||||
|
|
|
|||
|
|
@ -191,7 +191,7 @@ That file is the pitch in one place: one server, one `Resolve`-backed tool, and
|
|||
|
||||
### Change notifications become one stream
|
||||
|
||||
At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver.
|
||||
At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), a middleware can refuse a listen request per caller, and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver.
|
||||
|
||||
**[Subscriptions](handlers/subscriptions.md)** covers publishing and serving, **[its Clients twin](client/subscriptions.md)** the watching end, and **[Deploy & scale](run/deploy.md)** the bus.
|
||||
|
||||
|
|
|
|||
38
docs_src/subscriptions/tutorial006.py
Normal file
38
docs_src/subscriptions/tutorial006.py
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
from mcp_types import INVALID_REQUEST, SubscriptionsListenRequestParams
|
||||
|
||||
from mcp.server.auth.middleware.auth_context import get_access_token
|
||||
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
|
||||
from mcp.server.mcpserver import MCPServer
|
||||
from mcp.shared.exceptions import MCPError
|
||||
|
||||
# Who may see each file. Replace this table with a database or your RBAC system.
|
||||
ACCESS = {
|
||||
"files://report.pdf": {"alice", "bob"},
|
||||
"files://payroll.csv": {"carol"},
|
||||
}
|
||||
|
||||
|
||||
def can_access(user: str | None, uri: str) -> bool:
|
||||
return user is not None and user in ACCESS.get(uri, set())
|
||||
|
||||
|
||||
async def gate_subscriptions(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult:
|
||||
if ctx.method == "subscriptions/listen":
|
||||
params = SubscriptionsListenRequestParams.model_validate(ctx.params or {}, by_name=False)
|
||||
token = get_access_token()
|
||||
user = token.subject if token else None
|
||||
if not all(can_access(user, uri) for uri in params.notifications.resource_subscriptions or ()):
|
||||
raise MCPError(INVALID_REQUEST, "not permitted to watch the requested resources")
|
||||
return await call_next(ctx)
|
||||
|
||||
|
||||
mcp = MCPServer("Reports", middleware=[gate_subscriptions])
|
||||
|
||||
|
||||
@mcp.resource("files://{name}")
|
||||
def file(name: str) -> str:
|
||||
uri = f"files://{name}"
|
||||
token = get_access_token()
|
||||
if not can_access(token.subject if token else None, uri):
|
||||
raise MCPError(INVALID_REQUEST, f"Unknown resource: {uri}")
|
||||
return f"contents of {name}"
|
||||
|
|
@ -49,6 +49,7 @@ from typing_extensions import deprecated
|
|||
from mcp.shared.dispatcher import CallOptions, Outbound
|
||||
from mcp.shared.exceptions import MCPDeprecationWarning, NoBackChannelError
|
||||
from mcp.shared.peer import Meta, dump_params
|
||||
from mcp.shared.subscriptions import LISTEN_STREAM_METHODS
|
||||
|
||||
__all__ = ["Connection"]
|
||||
|
||||
|
|
@ -159,12 +160,25 @@ class NotifyOnlyOutbound(_NoChannelOutbound):
|
|||
over duplex stream transports: the pipe is real, so server notifications
|
||||
ride it, but the modern protocol forbids server-initiated JSON-RPC
|
||||
requests, so `send_raw_request` (inherited) refuses by construction.
|
||||
|
||||
Change notifications (`notifications/*/list_changed`,
|
||||
`notifications/resources/updated`) are dropped with a debug log: at this
|
||||
era they reach a client only through a `subscriptions/listen` stream it
|
||||
opened, so a bare copy on the shared channel would be an unrequested
|
||||
notification. Publish them on the server's `SubscriptionBus` instead.
|
||||
"""
|
||||
|
||||
def __init__(self, outbound: Outbound) -> None:
|
||||
self._outbound = outbound
|
||||
|
||||
async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
|
||||
# At the 2026-07-28 era these are `subscriptions/listen` stream goods
|
||||
# only: the spec forbids sending a change notification a subscription
|
||||
# did not request, and listen streams deliver them (stamped, filtered)
|
||||
# via the request-scoped outbound, never this connection-scoped channel.
|
||||
if method in LISTEN_STREAM_METHODS:
|
||||
logger.debug("dropped %s: delivered via subscriptions/listen at this era", method)
|
||||
return
|
||||
await self._outbound.notify(method, params, opts)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAut
|
|||
from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
|
||||
from mcp.server.auth.settings import AuthSettings
|
||||
from mcp.server.caching import CacheableMethod, CacheHint
|
||||
from mcp.server.context import HandlerResult, ServerRequestContext
|
||||
from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
|
||||
from mcp.server.extension import (
|
||||
Extension,
|
||||
MethodBinding,
|
||||
|
|
@ -172,6 +172,7 @@ class MCPServer(Generic[LifespanResultT]):
|
|||
request_state_security: RequestStateSecurity | None = None,
|
||||
cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
|
||||
subscriptions: SubscriptionBus | None = None,
|
||||
middleware: Sequence[ServerMiddleware[Any]] | None = None,
|
||||
):
|
||||
self._resource_security = resource_security
|
||||
self.settings = Settings(
|
||||
|
|
@ -228,6 +229,9 @@ class MCPServer(Generic[LifespanResultT]):
|
|||
raise ValueError(_MISSING_AUDIENCE)
|
||||
security = request_state_security
|
||||
self._lowlevel_server.middleware.append(RequestStateBoundary(security, default_audience=self.name))
|
||||
# User middleware runs inside the SDK's built-ins (OpenTelemetry, then the
|
||||
# request-state boundary), outermost-first in the order given.
|
||||
self._lowlevel_server.middleware.extend(middleware or ())
|
||||
# Validate auth configuration
|
||||
if self.settings.auth is not None:
|
||||
if auth_server_provider and token_verifier: # pragma: no cover
|
||||
|
|
@ -257,6 +261,17 @@ class MCPServer(Generic[LifespanResultT]):
|
|||
def name(self) -> str:
|
||||
return self._lowlevel_server.name
|
||||
|
||||
@property
|
||||
def middleware(self) -> list[ServerMiddleware[Any]]:
|
||||
"""The middleware chain wrapping every inbound message, outermost-first.
|
||||
|
||||
The same list as the low-level `Server.middleware`: append an
|
||||
`async (ctx, call_next)` callable to observe, refuse, or rewrite
|
||||
messages before they reach a handler. Provisional - the signature is
|
||||
expected to change before v2 is final; see the middleware guide.
|
||||
"""
|
||||
return self._lowlevel_server.middleware
|
||||
|
||||
@property
|
||||
def title(self) -> str | None:
|
||||
return self._lowlevel_server.title
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from mcp_types import (
|
|||
)
|
||||
|
||||
__all__ = [
|
||||
"LISTEN_STREAM_METHODS",
|
||||
"SUBSCRIPTION_ID_META_KEY",
|
||||
"PromptsListChanged",
|
||||
"ResourceUpdated",
|
||||
|
|
@ -79,6 +80,10 @@ _LIST_CHANGED_EVENTS: dict[str, ServerEvent] = {
|
|||
"notifications/resources/list_changed": ResourcesListChanged(),
|
||||
}
|
||||
|
||||
LISTEN_STREAM_METHODS: frozenset[str] = frozenset({*_LIST_CHANGED_EVENTS, "notifications/resources/updated"})
|
||||
"""The notification methods that ride `subscriptions/listen` streams at 2026-07-28
|
||||
(and, at that era, nowhere else): the change-notification vocabulary."""
|
||||
|
||||
|
||||
def event_from_wire(method: str, params: Mapping[str, Any] | None) -> ServerEvent | None:
|
||||
"""The event a raw listen-stream frame announces, or None if it carries none.
|
||||
|
|
|
|||
|
|
@ -16,11 +16,16 @@ from docs_src.subscriptions import (
|
|||
tutorial004_asyncio,
|
||||
tutorial004_trio,
|
||||
tutorial005,
|
||||
tutorial006,
|
||||
)
|
||||
from mcp import Client
|
||||
from mcp.server.auth.middleware.auth_context import auth_context_var
|
||||
from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
|
||||
from mcp.server.auth.provider import AccessToken
|
||||
from mcp.server.context import ServerRequestContext
|
||||
from mcp.server.lowlevel import Server
|
||||
from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, ListenHandler, ToolsListChanged
|
||||
from mcp.shared.exceptions import MCPError
|
||||
|
||||
_ReadResource = Callable[
|
||||
[ServerRequestContext[Any], types.ReadResourceRequestParams], Awaitable[types.ReadResourceResult]
|
||||
|
|
@ -301,3 +306,30 @@ async def test_the_follower_re_listens_after_the_stream_ends(capsys: pytest.Capt
|
|||
printed = capsys.readouterr().out
|
||||
assert "[x] design\n[ ] build" in printed # first stream, after design
|
||||
assert "[x] design\n[x] build" in printed # second stream, after build
|
||||
|
||||
|
||||
def _signed_in_as(subject: str) -> Any:
|
||||
"""Stand in for the auth middleware: put this user's token in the auth context."""
|
||||
token = AccessToken(token="demo", client_id="docs-client", scopes=[], subject=subject)
|
||||
return auth_context_var.set(AuthenticatedUser(token))
|
||||
|
||||
|
||||
async def test_the_middleware_refuses_a_listen_the_caller_could_not_read() -> None:
|
||||
"""tutorial006: one `can_access` gates both `resources/read` and `subscriptions/listen`.
|
||||
|
||||
Alice may read (and so watch) the report, and is refused the payroll file on both
|
||||
paths - the listen refusal is in-band, before any acknowledgment.
|
||||
"""
|
||||
reset = _signed_in_as("alice")
|
||||
try:
|
||||
async with Client(tutorial006.mcp, mode="2026-07-28") as client:
|
||||
async with client.listen(resource_subscriptions=["files://report.pdf"]) as sub:
|
||||
assert sub.honored.resource_subscriptions == ["files://report.pdf"]
|
||||
with pytest.raises(MCPError) as listen_error:
|
||||
async with client.listen(resource_subscriptions=["files://report.pdf", "files://payroll.csv"]):
|
||||
pass # pragma: no cover - the refusal precedes the stream
|
||||
assert listen_error.value.error.message == "not permitted to watch the requested resources"
|
||||
with pytest.raises(MCPError):
|
||||
await client.read_resource("files://payroll.csv")
|
||||
finally:
|
||||
auth_context_var.reset(reset)
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ from inline_snapshot import snapshot
|
|||
from mcp_types import (
|
||||
INTERNAL_ERROR,
|
||||
INVALID_PARAMS,
|
||||
INVALID_REQUEST,
|
||||
MISSING_REQUIRED_CLIENT_CAPABILITY,
|
||||
AudioContent,
|
||||
BlobResourceContents,
|
||||
|
|
@ -2345,3 +2346,53 @@ def test_remove_prompt_removes_and_unknown_name_raises() -> None:
|
|||
assert mcp._prompt_manager.list_prompts() == []
|
||||
with pytest.raises(ValueError, match="Unknown prompt: greeting"):
|
||||
mcp.remove_prompt("greeting")
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_middleware_kwarg_and_property_share_the_low_level_chain() -> None:
|
||||
"""SDK-defined: `MCPServer(middleware=[...])` appends to the low-level chain after
|
||||
the SDK's built-ins, and `mcp.middleware` is that same live list, so a
|
||||
middleware appended later still wraps requests."""
|
||||
seen: list[str] = []
|
||||
|
||||
async def from_ctor(ctx: ServerRequestContext[Any, Any], call_next: Any) -> Any:
|
||||
seen.append(f"ctor:{ctx.method}")
|
||||
return await call_next(ctx)
|
||||
|
||||
async def appended(ctx: ServerRequestContext[Any, Any], call_next: Any) -> Any:
|
||||
seen.append(f"appended:{ctx.method}")
|
||||
return await call_next(ctx)
|
||||
|
||||
mcp = MCPServer("mw", middleware=[from_ctor])
|
||||
assert mcp.middleware is mcp._lowlevel_server.middleware
|
||||
assert mcp.middleware[-1] is from_ctor # after the built-ins, outermost-first
|
||||
mcp.middleware.append(appended)
|
||||
|
||||
@mcp.tool()
|
||||
def ping() -> str:
|
||||
return "pong"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
await client.call_tool("ping", {})
|
||||
assert "ctor:tools/call" in seen
|
||||
assert seen.index("ctor:tools/call") < seen.index("appended:tools/call")
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_middleware_can_refuse_subscriptions_listen_before_the_ack() -> None:
|
||||
"""Spec-adjacent: a middleware that raises on `subscriptions/listen` refuses the
|
||||
request in-band - the client gets the error and no stream is opened."""
|
||||
|
||||
async def refuse_listen(ctx: ServerRequestContext[Any, Any], call_next: Any) -> Any:
|
||||
if ctx.method == "subscriptions/listen":
|
||||
raise MCPError(INVALID_REQUEST, "not permitted to watch the requested resources")
|
||||
return await call_next(ctx)
|
||||
|
||||
mcp = MCPServer("mw", middleware=[refuse_listen])
|
||||
|
||||
async with Client(mcp) as client:
|
||||
with pytest.raises(MCPError) as exc_info:
|
||||
async with client.listen(resource_subscriptions=["files://payroll.csv"]):
|
||||
pass # pragma: no cover - the refusal precedes the stream
|
||||
assert exc_info.value.error.code == INVALID_REQUEST
|
||||
assert exc_info.value.error.message == "not permitted to watch the requested resources"
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ behaviour under test. Driver tests (`serve_connection`, `serve_one`,
|
|||
"""
|
||||
|
||||
import contextvars
|
||||
import logging
|
||||
from collections.abc import AsyncIterator, Mapping
|
||||
from contextlib import asynccontextmanager
|
||||
from dataclasses import dataclass, field, replace
|
||||
|
|
@ -39,6 +40,8 @@ from mcp_types import (
|
|||
InitializeRequestParams,
|
||||
JSONRPCRequest,
|
||||
ListToolsResult,
|
||||
LoggingMessageNotification,
|
||||
LoggingMessageNotificationParams,
|
||||
NotificationParams,
|
||||
PaginatedRequestParams,
|
||||
ProgressNotificationParams,
|
||||
|
|
@ -1783,17 +1786,23 @@ async def test_dual_era_loop_a_notifications_meta_never_opens_the_log_gate(serve
|
|||
@pytest.mark.anyio
|
||||
async def test_dual_era_loop_modern_server_notifications_ride_the_pipe(server: SrvT):
|
||||
"""A modern handler's standalone notification reaches the client over the
|
||||
duplex stream - the notify-only outbound forwards it."""
|
||||
duplex stream - the notify-only outbound forwards it. Change notifications
|
||||
are the exception: at this era they reach clients only via
|
||||
`subscriptions/listen` streams, so a bare one is dropped rather than sent
|
||||
unrequested."""
|
||||
|
||||
async def emit(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]:
|
||||
await ctx.session.send_tool_list_changed()
|
||||
await ctx.session.send_tool_list_changed() # dropped: an unrequested change notification
|
||||
await ctx.session.send_notification(
|
||||
LoggingMessageNotification(params=LoggingMessageNotificationParams(level="info", data="hi"))
|
||||
)
|
||||
return {}
|
||||
|
||||
server.add_request_handler("x/emit", RequestParams, emit)
|
||||
async with dual_era_client(server) as (client, recorder):
|
||||
await client.send_raw_request("x/emit", _modern_params())
|
||||
await recorder.notified.wait()
|
||||
assert recorder.notifications[0][0] == "notifications/tools/list_changed"
|
||||
assert [method for method, _ in recorder.notifications] == ["notifications/message"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
|
|
@ -1992,12 +2001,35 @@ def test_no_server_requests_dispatch_context_passes_an_already_denying_transport
|
|||
async def test_notify_only_outbound_forwards_notifications_and_refuses_requests():
|
||||
inner = _RecordingInnerDctx()
|
||||
outbound = NotifyOnlyOutbound(inner)
|
||||
await outbound.notify("notifications/tools/list_changed", None)
|
||||
assert inner.notifies == ["notifications/tools/list_changed"]
|
||||
await outbound.notify("notifications/message", None)
|
||||
assert inner.notifies == ["notifications/message"]
|
||||
with pytest.raises(NoBackChannelError):
|
||||
await outbound.send_raw_request("ping", None)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize(
|
||||
"method",
|
||||
[
|
||||
"notifications/tools/list_changed",
|
||||
"notifications/prompts/list_changed",
|
||||
"notifications/resources/list_changed",
|
||||
"notifications/resources/updated",
|
||||
],
|
||||
)
|
||||
async def test_notify_only_outbound_drops_change_notifications(method: str, caplog: pytest.LogCaptureFixture):
|
||||
"""Spec: the server never sends a notification type a subscription did not
|
||||
request. At the modern era change notifications reach a client only through
|
||||
a `subscriptions/listen` stream, so a bare copy on the shared channel would
|
||||
be an unrequested notification - the standalone channel drops it."""
|
||||
inner = _RecordingInnerDctx()
|
||||
outbound = NotifyOnlyOutbound(inner)
|
||||
with caplog.at_level(logging.DEBUG, logger="mcp.server.connection"):
|
||||
await outbound.notify(method, None)
|
||||
assert inner.notifies == []
|
||||
assert f"dropped {method}: delivered via subscriptions/listen at this era" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_dual_era_loop_carries_the_sender_context_through_the_replay():
|
||||
"""A context-aware read stream's per-message sender context still reaches
|
||||
|
|
|
|||
Loading…
Reference in a new issue