Add a subscription authorization hook and stop sending unrequested change notifications

docs/handlers/subscriptions.md

@@ -43,14 +43,35 @@ 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. On a multi-tenant server, gate that with an authorization hook. It runs once per `subscriptions/listen`, before the acknowledgment, with the request context and the filter the client asked for, and either returns (accept the request as asked) or raises `MCPError` (refuse it):
+
+```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.shared.exceptions import MCPError
+from mcp_types import INVALID_REQUEST, SubscriptionFilter
+
+
+async def only_own_resources(ctx: ServerRequestContext, requested: SubscriptionFilter) -> None:
+    token = get_access_token()
+    prefix = f"user://{token.subject}/" if token and token.subject else ""
+    if any(not (prefix and uri.startswith(prefix)) for uri in requested.resource_subscriptions or []):
+        raise MCPError(INVALID_REQUEST, "not permitted to watch the requested resources")
+
+
+mcp = MCPServer("Sprint Board", authorize_subscriptions=only_own_resources)
+```
+
+* The hook decides on the whole request: accept it as asked, or refuse it. It does not narrow the filter - the acknowledgment reflects what the server supports, not an authorization verdict.
+* A refused request gets the error and no stream. Keep the error uniform (name no URI) so refusals do not reveal which URIs are protected. Any other exception refuses too: a broken policy never grants.
+* The verdict 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, authorize=only_own_resources)`.
+
+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`, `authorize_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 accept or refuse a listen request per caller, pass an `authorize_subscriptions` hook (`MCPServer(authorize_subscriptions=...)`, or `ListenHandler(bus, authorize=...)` 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), an `authorize_subscriptions` hook accepts or refuses each 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.
 

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 AuthorizeSubscription, InMemorySubscriptionBus, ListenHandler, 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,
+        authorize_subscriptions: AuthorizeSubscription | 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. `authorize_subscriptions` is
+        # the per-request accept-or-refuse hook (see `AuthorizeSubscription`);
+        # 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, authorize=authorize_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

src/mcp/server/subscriptions.py

@@ -20,12 +20,18 @@
 `_meta["io.modelcontextprotocol/subscriptionId"]`, and never delivers an
 event kind the client did not request. Delivery is fire-and-forget with no
 replay: a dropped stream is not resumable - clients re-listen and refetch.
+
+An `AuthorizeSubscription` hook, when supplied, decides per request whether
+the caller may open the subscription it asked for: it accepts the request as
+requested or refuses it outright before the acknowledgment. It does not narrow
+the filter - the honored subset is a support contract, not an authorization
+verdict.
 """
 
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from typing import Any, Protocol
 
 import anyio
@@ -56,6 +62,7 @@
 
 __all__ = [
     "SUBSCRIPTION_ID_META_KEY",
+    "AuthorizeSubscription",
     "InMemorySubscriptionBus",
     "ListenHandler",
     "PromptsListChanged",
@@ -68,6 +75,23 @@
 
 logger = logging.getLogger(__name__)
 
+AuthorizeSubscription = Callable[[ServerRequestContext[Any, Any], SubscriptionFilter], Awaitable[None]]
+"""Per-request accept-or-refuse hook for `subscriptions/listen`.
+
+Called once, before the acknowledgment, with the request context and the
+filter the client asked for. Return to accept the request as requested; raise
+`MCPError` to refuse it, so the client gets an in-band error and no stream is
+opened. Any other exception also refuses: a broken policy never grants access.
+
+The hook decides on the whole request - it does not narrow the filter. When
+a request names anything the caller may not watch, refuse it (a uniform error
+regardless of which URI failed avoids confirming which URIs are protected).
+The verdict holds for the stream's lifetime; there is no per-event re-check, so
+revoking one caller's access mid-stream (an expired credential, say) means
+ending that caller's connection - `ListenHandler.close` ends every open stream
+at once and is meant for server shutdown.
+"""
+
 
 class SubscriptionBus(Protocol):
     """Fan-out seam between event publishers and open listen streams.
@@ -167,16 +191,27 @@ class ListenHandler:
     Served on any transport that can carry the request's response stream:
     streamable HTTP's SSE mode, or a duplex stream pair such as stdio.
 
-    `max_subscriptions` bounds concurrent streams (further listen requests are
-    rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events`
+    Without an `authorize` hook every requested kind and resource URI is
+    honored: any caller may watch any URI the server publishes. Pass one to
+    accept or refuse each request - see `AuthorizeSubscription`.
+    `max_subscriptions` bounds concurrent streams (further listen requests
+    are rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events`
     bounds each stream's event backlog: a stream whose client has stopped
     reading is ended at the cap (the client re-listens and refetches - there
     is no replay, so ending the stream loses nothing the backlog wasn't
     already losing).
     """
 
-    def __init__(self, bus: SubscriptionBus, *, max_subscriptions: int = 1024, max_buffered_events: int = 1024) -> None:
+    def __init__(
+        self,
+        bus: SubscriptionBus,
+        *,
+        authorize: AuthorizeSubscription | None = None,
+        max_subscriptions: int = 1024,
+        max_buffered_events: int = 1024,
+    ) -> None:
         self._bus = bus
+        self._authorize = authorize
         self._max_subscriptions = max_subscriptions
         self._max_buffered_events = max_buffered_events
         self._streams: set[anyio.streams.memory.MemoryObjectSendStream[ServerEvent]] = set()
@@ -190,6 +225,7 @@ async def __call__(
         subscription_id = ctx.request_id
         if subscription_id is None:
             raise MCPError(INVALID_REQUEST, "subscriptions/listen requires a request id")
+        await self._authorize_request(ctx, params.notifications)
         if len(self._streams) >= self._max_subscriptions:
             raise MCPError(INTERNAL_ERROR, "Subscription limit reached")
         honored = _honored_subset(params.notifications)
@@ -240,6 +276,23 @@ def deliver(event: ServerEvent) -> None:
             recv.close()
         return SubscriptionsListenResult(_meta=meta)
 
+    async def _authorize_request(self, ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> None:
+        """Run the `authorize` hook, if any, before the acknowledgment.
+
+        The hook's `MCPError` is its explicit refusal and propagates as the
+        pre-ack error. Any other exception fails closed - the subscription is
+        refused, never opened - so a broken policy cannot hand out access.
+        """
+        if self._authorize is None:
+            return
+        try:
+            await self._authorize(ctx, requested)
+        except MCPError:
+            raise
+        except Exception as exc:  # deny-on-error: a raising policy must never grant
+            logger.exception("subscription authorize hook raised; refusing subscriptions/listen")
+            raise MCPError(INTERNAL_ERROR, "Subscription authorization failed") from exc
+
     def close(self) -> None:
         """Initiate graceful closure of every open listen stream.