Add a subscription authorization hook and stop sending unrequested change notifications

docs/handlers/subscriptions.md

@@ -43,14 +43,36 @@ 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 what a caller may watch
+
+By default every requested kind and URI is honored: any caller may watch any URI you publish. On a multi-tenant server, decide per caller with a narrowing hook. It runs once per `subscriptions/listen`, before the acknowledgment, with the request context and the filter the client asked for, and returns the filter to grant:
+
+```python
+from mcp.server.auth.middleware.auth_context import get_access_token
+from mcp.server.context import ServerRequestContext
+from mcp.server.mcpserver import MCPServer
+from mcp_types import SubscriptionFilter
+
+
+async def owned_only(ctx: ServerRequestContext, requested: SubscriptionFilter) -> SubscriptionFilter:
+    token = get_access_token()
+    prefix = f"user://{token.subject}/" if token and token.subject else ""
+    watchable = [uri for uri in requested.resource_subscriptions or [] if prefix and uri.startswith(prefix)]
+    return requested.model_copy(update={"resource_subscriptions": watchable})
+
+
+mcp = MCPServer("Sprint Board", narrow_subscriptions=owned_only)
+```
+
+* The grant is intersected with the request, so a hook can drop kinds and URIs but never add them.
+* The acknowledgment reports the grant; that is how the client learns it got less than it asked for.
+* Decide by pattern, not by lookup. `owned_only` prunes everything outside the caller's own prefix without checking whether a URI exists, so the acknowledgment reveals the shape of the policy and nothing about which URIs are real.
+* To refuse the whole subscription, raise `MCPError` from the hook: the client gets the error and no stream. Any other exception refuses too - a broken policy never grants.
+* 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. `ListenHandler.close()` is not the tool for that: it ends every open stream at once, which you want at shutdown, not for one caller.
+
+On the low-level `Server` the same hook is `ListenHandler(bus, narrow=owned_only)`.
+
+Without a hook the exposure is narrow but real, so weigh it before publishing per-user URIs from a multi-tenant server: 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.
 
 ## The client end
 

docs/migration.md

@@ -721,7 +721,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`, `narrow_subscriptions`) are additive.
 
 **Parameters moved:**
 
@@ -2819,6 +2819,12 @@ async def book_table(date: str, answer: Annotated[Confirmation, Resolve(ask_to_c
 
 The client's same `elicitation_callback` answers both; the resolver lets the server *return* the question instead of pushing it.
 
+### 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 decide per caller what that is, pass a `narrow_subscriptions` hook (`MCPServer(narrow_subscriptions=...)`, or `ListenHandler(bus, narrow=...)` on the low-level `Server`), 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.

docs/whats-new.md

@@ -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 `narrow_subscriptions` hook decides per caller what a stream may watch, 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.
 

src/mcp/server/connection.py

@@ -116,6 +116,19 @@ async def notify(self, method: str, params: Mapping[str, Any] | None, opts: Call
 
 _NO_CHANNEL = _NoChannelOutbound()
 
+# On 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 and filtered) via the request-scoped
+# outbound, never this connection-scoped channel.
+_SUBSCRIPTION_STREAM_METHODS = frozenset(
+    {
+        "notifications/tools/list_changed",
+        "notifications/prompts/list_changed",
+        "notifications/resources/list_changed",
+        "notifications/resources/updated",
+    }
+)
+
 
 class NotifyOnlyOutbound(_NoChannelOutbound):
     """Connection-scoped `Outbound` that forwards notifications and refuses requests.
@@ -124,12 +137,21 @@ 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:
+        if method in _SUBSCRIPTION_STREAM_METHODS:
+            logger.debug("dropped %s: delivered via subscriptions/listen at this era", method)
+            return
         await self._outbound.notify(method, params, opts)
 
 

src/mcp/server/mcpserver/server.py

@@ -88,7 +88,7 @@
 from mcp.server.stdio import stdio_server
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
-from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
+from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, NarrowSubscription, SubscriptionBus
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
 from mcp.shared.uri_template import UriTemplate
@@ -172,6 +172,7 @@ def __init__(
         request_state_security: RequestStateSecurity | None = None,
         cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
         subscriptions: SubscriptionBus | None = None,
+        narrow_subscriptions: NarrowSubscription | None = None,
     ):
         self._resource_security = resource_security
         self.settings = Settings(
@@ -193,7 +194,9 @@ def __init__(
         self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
         # The subscriptions/listen fan-out seam (2026-07-28). The default bus is
         # in-process; pass an `SubscriptionBus` implementation over an external pub/sub
-        # backend to fan events out across replicas.
+        # backend to fan events out across replicas. `narrow_subscriptions` is the
+        # per-request grant hook (see `NarrowSubscription`); without it every
+        # requested kind and resource URI is honored.
         self._subscriptions: SubscriptionBus = subscriptions if subscriptions is not None else InMemorySubscriptionBus()
         self._lowlevel_server = Server(
             name=name or "mcp-server",
@@ -211,7 +214,7 @@ def __init__(
             on_list_resource_templates=self._handle_list_resource_templates,
             on_list_prompts=self._handle_list_prompts,
             on_get_prompt=self._handle_get_prompt,
-            on_subscriptions_listen=ListenHandler(self._subscriptions),
+            on_subscriptions_listen=ListenHandler(self._subscriptions, narrow=narrow_subscriptions),
             # TODO(Marcelo): It seems there's a type mismatch between the lifespan type from an MCPServer and Server.
             # We need to create a Lifespan type that is a generic on the server type, like Starlette does.
             lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),  # type: ignore