Make the subscription authorization hook accept-or-refuse instead of narrowing

Rework the `subscriptions/listen` authorization seam to Go's shape: the hook
(now `authorize=` on ListenHandler / `authorize_subscriptions=` on MCPServer,
typed `AuthorizeSubscription`) either returns to accept the request as asked
or raises MCPError to refuse the whole request before the acknowledgment. It
no longer returns a narrowed filter, and the honored subset is again the plain
support-based echo of the request rather than an intersection with a grant.
Refusing on any unauthorized URI, with a uniform error, is now the documented
pattern; a non-MCPError from the hook still fails closed.

No-Verification-Needed: pushing early for review at maintainer request; verification to follow

Author: maxisbey

docs/handlers/subscriptions.md

@@ -43,34 +43,33 @@ 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.
 
-## Deciding what a caller may watch
+## 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, 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:
+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_types import SubscriptionFilter
+from mcp.shared.exceptions import MCPError
+from mcp_types import INVALID_REQUEST, SubscriptionFilter
 
 
-async def owned_only(ctx: ServerRequestContext, requested: SubscriptionFilter) -> 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 ""
-    watchable = [uri for uri in requested.resource_subscriptions or [] if prefix and uri.startswith(prefix)]
-    return requested.model_copy(update={"resource_subscriptions": watchable})
+    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", narrow_subscriptions=owned_only)
+mcp = MCPServer("Sprint Board", authorize_subscriptions=only_own_resources)
 ```
 
-* 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.
+* 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, narrow=owned_only)`.
+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.
 

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`, `narrow_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:**
 
@@ -2823,7 +2823,7 @@ The client's same `elicitation_callback` answers both; the resolver lets the ser
 
 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.
+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))
 

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), 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.
+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/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, NarrowSubscription, 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,7 +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,
+        authorize_subscriptions: AuthorizeSubscription | None = None,
     ):
         self._resource_security = resource_security
         self.settings = Settings(
@@ -194,9 +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. `narrow_subscriptions` is the
-        # per-request grant hook (see `NarrowSubscription`); without it every
-        # requested kind and resource URI is honored.
+        # 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",
@@ -214,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, narrow=narrow_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