From b07b2b331f29343caf3e776b8e1052456736ecf5 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Mon, 27 Jul 2026 23:07:37 +0000
Subject: [PATCH 1/2] Add a subscription narrowing hook and stop sending
 unrequested change notifications

Two server-side gaps on the 2026-07-28 subscriptions/listen path, closed together
because either alone leaves the other's leak open.

Nothing let a server decide per caller what a subscription may watch: any client
could name any resource URI in its filter and the server honored it verbatim.
ListenHandler now takes an optional `narrow` hook (`MCPServer(narrow_subscriptions=...)`)
that runs once before the acknowledgment and returns the filter to grant, or
raises MCPError to refuse the request. The grant is intersected with the request
so a hook can narrow but never widen, and any other exception fails closed rather
than granting.

Separately, the modern-era standalone channel forwarded every notification, so
`ctx.session.send_tool_list_changed()` on a 2026-07-28 stdio connection wrote a
bare, unstamped list_changed frame that no subscription had requested. That
channel now drops the four change-notification methods; they reach clients only
through listen streams, which stamp and filter them.
---
 docs/handlers/subscriptions.md     |  38 +++++++--
 docs/migration.md                  |   8 +-
 docs/whats-new.md                  |   2 +-
 src/mcp/server/connection.py       |  22 +++++
 src/mcp/server/mcpserver/server.py |   9 +-
 src/mcp/server/subscriptions.py    |  92 ++++++++++++++++----
 tests/server/test_runner.py        |  42 +++++++--
 tests/server/test_subscriptions.py | 133 +++++++++++++++++++++++++++++
 8 files changed, 311 insertions(+), 35 deletions(-)

diff --git a/docs/handlers/subscriptions.md b/docs/handlers/subscriptions.md
index 99b1bc98ad..917c08129b 100644
--- a/docs/handlers/subscriptions.md
+++ b/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
 
diff --git a/docs/migration.md b/docs/migration.md
index 7b5950e7cf..b1ab4762e0 100644
--- a/docs/migration.md
+++ b/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.
diff --git a/docs/whats-new.md b/docs/whats-new.md
index 9826fa5909..c4dc9c3e40 100644
--- a/docs/whats-new.md
+++ b/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.
 
diff --git a/src/mcp/server/connection.py b/src/mcp/server/connection.py
index ca05928dff..254fcbd242 100644
--- a/src/mcp/server/connection.py
+++ b/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)
 
 
diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py
index 33a0b08c84..8360d60eed 100644
--- a/src/mcp/server/mcpserver/server.py
+++ b/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
diff --git a/src/mcp/server/subscriptions.py b/src/mcp/server/subscriptions.py
index f5d3b5b0a4..dc54027706 100644
--- a/src/mcp/server/subscriptions.py
+++ b/src/mcp/server/subscriptions.py
@@ -20,12 +20,17 @@
 `_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.
+
+A `NarrowSubscription` hook, when supplied, decides per request what the
+server actually grants: it can narrow the requested filter (fewer kinds, a
+subset of the URIs) or refuse the subscription outright, and the acknowledged
+filter tells the client what it got.
 """
 
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from typing import Any, Protocol
 
 import anyio
@@ -58,6 +63,7 @@
     "SUBSCRIPTION_ID_META_KEY",
     "InMemorySubscriptionBus",
     "ListenHandler",
+    "NarrowSubscription",
     "PromptsListChanged",
     "ResourceUpdated",
     "ResourcesListChanged",
@@ -68,6 +74,24 @@
 
 logger = logging.getLogger(__name__)
 
+NarrowSubscription = Callable[[ServerRequestContext[Any, Any], SubscriptionFilter], Awaitable[SubscriptionFilter]]
+"""Per-request hook deciding what a `subscriptions/listen` request is granted.
+
+Called once, before the acknowledgment, with the request context and the
+filter the client asked for. Return the filter to grant - typically the
+request itself, possibly with kinds cleared or `resource_subscriptions` cut
+down to the URIs this caller may watch - or raise `MCPError` to refuse the
+subscription (the client gets an in-band error and no stream). The grant is
+intersected with the request, so the hook can narrow but never widen, and the
+acknowledgment reports the result so the client can see what it got. Any
+other exception refuses the subscription: a raising policy never grants.
+
+The decision 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` is not the
+tool for it: it ends every open stream at once (server shutdown).
+"""
+
 
 class SubscriptionBus(Protocol):
     """Fan-out seam between event publishers and open listen streams.
@@ -138,19 +162,24 @@ def _safe_unsubscribe(unsubscribe: Callable[[], None]) -> None:
         logger.exception("bus unsubscribe raised; continuing stream cleanup")
 
 
-def _honored_subset(requested: SubscriptionFilter) -> SubscriptionFilter:
+def _honored_subset(requested: SubscriptionFilter, granted: SubscriptionFilter) -> SubscriptionFilter:
     """The subset of `requested` the server will deliver, for the ack.
 
-    Every requested kind is honored - whether an event kind ever fires
-    depends on what the server publishes, exactly as a subscription to a
-    nonexistent resource URI is honored and never fires. Non-true flags and
+    A kind is honored only when both the request and the grant carry it, and
+    the URI list keeps the requested URIs the grant also names, so a grant can
+    narrow the request but never widen it. With no narrowing policy the grant
+    is the request itself and nothing is dropped. Whether an honored kind ever
+    fires depends on what the server publishes, exactly as a subscription to
+    a nonexistent resource URI is honored and never fires. Non-true flags and
     an empty URI list are dropped rather than echoed as falsy values.
     """
+    granted_uris = frozenset(granted.resource_subscriptions or ())
+    uris = [uri for uri in requested.resource_subscriptions or () if uri in granted_uris]
     return SubscriptionFilter(
-        tools_list_changed=True if requested.tools_list_changed else None,
-        prompts_list_changed=True if requested.prompts_list_changed else None,
-        resources_list_changed=True if requested.resources_list_changed else None,
-        resource_subscriptions=list(requested.resource_subscriptions) if requested.resource_subscriptions else None,
+        tools_list_changed=True if requested.tools_list_changed and granted.tools_list_changed else None,
+        prompts_list_changed=True if requested.prompts_list_changed and granted.prompts_list_changed else None,
+        resources_list_changed=True if requested.resources_list_changed and granted.resources_list_changed else None,
+        resource_subscriptions=uris or None,
     )
 
 
@@ -167,16 +196,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`
-    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).
+    Without a `narrow` hook every requested kind and resource URI is
+    honored: any caller may watch any URI the server publishes. Pass one to
+    grant per request - see `NarrowSubscription`. `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,
+        *,
+        narrow: NarrowSubscription | None = None,
+        max_subscriptions: int = 1024,
+        max_buffered_events: int = 1024,
+    ) -> None:
         self._bus = bus
+        self._narrow = narrow
         self._max_subscriptions = max_subscriptions
         self._max_buffered_events = max_buffered_events
         self._streams: set[anyio.streams.memory.MemoryObjectSendStream[ServerEvent]] = set()
@@ -190,9 +230,10 @@ async def __call__(
         subscription_id = ctx.request_id
         if subscription_id is None:
             raise MCPError(INVALID_REQUEST, "subscriptions/listen requires a request id")
+        granted = await self._grant(ctx, params.notifications)
         if len(self._streams) >= self._max_subscriptions:
             raise MCPError(INTERNAL_ERROR, "Subscription limit reached")
-        honored = _honored_subset(params.notifications)
+        honored = _honored_subset(params.notifications, granted)
         honored_uris = frozenset(honored.resource_subscriptions or ())
         meta: dict[str, Any] = {SUBSCRIPTION_ID_META_KEY: subscription_id}
 
@@ -240,6 +281,23 @@ def deliver(event: ServerEvent) -> None:
             recv.close()
         return SubscriptionsListenResult(_meta=meta)
 
+    async def _grant(self, ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+        """The filter to grant this request: the request itself unless a `narrow` hook decides.
+
+        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 granted - so a broken policy cannot hand out access.
+        """
+        if self._narrow is None:
+            return requested
+        try:
+            return await self._narrow(ctx, requested)
+        except MCPError:
+            raise
+        except Exception as exc:  # deny-on-error: a raising policy must never grant
+            logger.exception("subscription narrow 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.
 
diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py
index eb212dafdb..f22979f464 100644
--- a/tests/server/test_runner.py
+++ b/tests/server/test_runner.py
@@ -8,6 +8,7 @@
 """
 
 import contextvars
+import logging
 from collections.abc import AsyncIterator, Mapping
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field, replace
@@ -38,6 +39,8 @@
     InitializeRequestParams,
     JSONRPCRequest,
     ListToolsResult,
+    LoggingMessageNotification,
+    LoggingMessageNotificationParams,
     NotificationParams,
     PaginatedRequestParams,
     ProgressNotificationParams,
@@ -1762,17 +1765,23 @@ async def on_custom(ctx: Ctx, params: NotificationParams | None) -> None:
 @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
@@ -1971,12 +1980,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
diff --git a/tests/server/test_subscriptions.py b/tests/server/test_subscriptions.py
index 579e3522fe..217b36b93d 100644
--- a/tests/server/test_subscriptions.py
+++ b/tests/server/test_subscriptions.py
@@ -6,6 +6,7 @@
 import anyio
 import pytest
 from mcp_types import (
+    INTERNAL_ERROR,
     INVALID_REQUEST,
     PromptListChangedNotification,
     RequestId,
@@ -282,6 +283,138 @@ async def run() -> None:
     assert isinstance(session.sent[1][0], ToolListChangedNotification)
 
 
+@pytest.mark.anyio
+async def test_narrow_hook_narrows_the_ack_and_the_delivered_events() -> None:
+    """SDK-defined: a `narrow` hook grants a subset - it can drop a kind and
+    prune the URI list - and the ack and the stream both reflect the grant."""
+    bus = InMemorySubscriptionBus()
+    seen: list[SubscriptionFilter] = []
+
+    async def narrow(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+        seen.append(requested)
+        return SubscriptionFilter(resources_list_changed=True, resource_subscriptions=["r://mine"])
+
+    handler = ListenHandler(bus, narrow=narrow)
+    session = _RecordingSession()
+
+    async with anyio.create_task_group() as tg:
+
+        async def run() -> None:
+            await handler(
+                _ctx(session),
+                _params(
+                    tools_list_changed=True,
+                    resources_list_changed=True,
+                    resource_subscriptions=["r://mine", "r://theirs"],
+                ),
+            )
+
+        tg.start_soon(run)
+        await session.wait_for(1)
+        assert seen == [
+            SubscriptionFilter(
+                tools_list_changed=True, resources_list_changed=True, resource_subscriptions=["r://mine", "r://theirs"]
+            )
+        ]
+
+        ack, _ = session.sent[0]
+        assert isinstance(ack, SubscriptionsAcknowledgedNotification)
+        # tools_list_changed was requested but not granted, and r://theirs was pruned.
+        assert ack.params.notifications == SubscriptionFilter(
+            resources_list_changed=True, resource_subscriptions=["r://mine"]
+        )
+
+        await bus.publish(ToolsListChanged())  # requested but not granted
+        await bus.publish(ResourceUpdated(uri="r://theirs"))  # pruned URI
+        await bus.publish(ResourcesListChanged())
+        await bus.publish(ResourceUpdated(uri="r://mine"))
+        await session.wait_for(3)
+        handler.close()
+
+    delivered = [notification for notification, _ in session.sent[1:]]
+    assert isinstance(delivered[0], ResourceListChangedNotification)
+    assert isinstance(delivered[1], ResourceUpdatedNotification)
+    assert delivered[1].params.uri == "r://mine"
+    assert len(delivered) == 2
+
+
+@pytest.mark.anyio
+async def test_narrow_hook_can_only_narrow_never_widen() -> None:
+    """SDK-defined: the grant is intersected with the request, so a hook returning
+    kinds or URIs the client never asked for cannot cause them to be delivered -
+    the spec's never-send-unrequested-types rule holds by construction."""
+    bus = InMemorySubscriptionBus()
+
+    async def widen(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+        return SubscriptionFilter(
+            tools_list_changed=True, prompts_list_changed=True, resource_subscriptions=["r://a", "r://extra"]
+        )
+
+    handler = ListenHandler(bus, narrow=widen)
+    session = _RecordingSession()
+
+    async with anyio.create_task_group() as tg:
+
+        async def run() -> None:
+            await handler(_ctx(session), _params(tools_list_changed=True, resource_subscriptions=["r://a"]))
+
+        tg.start_soon(run)
+        await session.wait_for(1)
+
+        ack, _ = session.sent[0]
+        assert isinstance(ack, SubscriptionsAcknowledgedNotification)
+        assert ack.params.notifications == SubscriptionFilter(tools_list_changed=True, resource_subscriptions=["r://a"])
+
+        await bus.publish(PromptsListChanged())  # granted but never requested
+        await bus.publish(ResourceUpdated(uri="r://extra"))  # granted but never requested
+        await bus.publish(ToolsListChanged())
+        await session.wait_for(2)
+        handler.close()
+
+    delivered = [notification for notification, _ in session.sent[1:]]
+    assert isinstance(delivered[0], ToolListChangedNotification)
+    assert len(delivered) == 1
+
+
+@pytest.mark.anyio
+async def test_narrow_hook_mcp_error_refuses_the_subscription_pre_ack() -> None:
+    """SDK-defined: a hook that raises `MCPError` refuses the whole request -
+    the error reaches the client and no ack or stream is opened."""
+    bus = _SpyBus()
+
+    async def refuse(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+        raise MCPError(INVALID_REQUEST, "not allowed to watch these resources")
+
+    handler = ListenHandler(bus, narrow=refuse)
+    session = _RecordingSession()
+
+    with pytest.raises(MCPError) as exc_info:
+        await handler(_ctx(session), _params(resource_subscriptions=["r://secret"]))
+    assert exc_info.value.error.message == "not allowed to watch these resources"
+    assert session.sent == []  # no ack: the refusal precedes the first frame
+    assert bus.unsubscribed == 0  # never subscribed
+
+
+@pytest.mark.anyio
+async def test_a_broken_narrow_hook_fails_closed(caplog: pytest.LogCaptureFixture) -> None:
+    """SDK-defined: a hook that raises anything other than `MCPError` is a
+    policy error and must never grant - the subscription is refused with an
+    internal error rather than falling back to honoring the request."""
+
+    async def broken(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+        raise RuntimeError("policy database unavailable")
+
+    handler = ListenHandler(InMemorySubscriptionBus(), narrow=broken)
+    session = _RecordingSession()
+
+    with pytest.raises(MCPError) as exc_info:
+        await handler(_ctx(session), _params(tools_list_changed=True))
+    assert exc_info.value.error.code == INTERNAL_ERROR
+    assert exc_info.value.error.message == "Subscription authorization failed"
+    assert session.sent == []
+    assert "subscription narrow hook raised" in caplog.text
+
+
 @pytest.mark.anyio
 async def test_listen_requires_a_request_id() -> None:
     """SDK-defined: a context without a request id cannot open a stream."""

From 6c11684ebea3af3e3fdcf468060d4bcf1baa90c1 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Tue, 28 Jul 2026 09:31:18 +0000
Subject: [PATCH 2/2] 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
---
 docs/handlers/subscriptions.md     | 25 ++++----
 docs/migration.md                  |  4 +-
 docs/whats-new.md                  |  2 +-
 src/mcp/server/mcpserver/server.py | 12 ++--
 src/mcp/server/subscriptions.py    | 97 ++++++++++++++----------------
 tests/server/test_subscriptions.py | 91 ++++++----------------------
 6 files changed, 85 insertions(+), 146 deletions(-)

diff --git a/docs/handlers/subscriptions.md b/docs/handlers/subscriptions.md
index 917c08129b..7d1ad3c614 100644
--- a/docs/handlers/subscriptions.md
+++ b/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.
 
diff --git a/docs/migration.md b/docs/migration.md
index b1ab4762e0..6cf4bc2008 100644
--- a/docs/migration.md
+++ b/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))
 
diff --git a/docs/whats-new.md b/docs/whats-new.md
index c4dc9c3e40..c9906746d1 100644
--- a/docs/whats-new.md
+++ b/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.
 
diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py
index 8360d60eed..9d081904f8 100644
--- a/src/mcp/server/mcpserver/server.py
+++ b/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
diff --git a/src/mcp/server/subscriptions.py b/src/mcp/server/subscriptions.py
index dc54027706..6a0266fd27 100644
--- a/src/mcp/server/subscriptions.py
+++ b/src/mcp/server/subscriptions.py
@@ -21,10 +21,11 @@
 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.
 
-A `NarrowSubscription` hook, when supplied, decides per request what the
-server actually grants: it can narrow the requested filter (fewer kinds, a
-subset of the URIs) or refuse the subscription outright, and the acknowledged
-filter tells the client what it got.
+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
@@ -61,9 +62,9 @@
 
 __all__ = [
     "SUBSCRIPTION_ID_META_KEY",
+    "AuthorizeSubscription",
     "InMemorySubscriptionBus",
     "ListenHandler",
-    "NarrowSubscription",
     "PromptsListChanged",
     "ResourceUpdated",
     "ResourcesListChanged",
@@ -74,22 +75,21 @@
 
 logger = logging.getLogger(__name__)
 
-NarrowSubscription = Callable[[ServerRequestContext[Any, Any], SubscriptionFilter], Awaitable[SubscriptionFilter]]
-"""Per-request hook deciding what a `subscriptions/listen` request is granted.
+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 the filter to grant - typically the
-request itself, possibly with kinds cleared or `resource_subscriptions` cut
-down to the URIs this caller may watch - or raise `MCPError` to refuse the
-subscription (the client gets an in-band error and no stream). The grant is
-intersected with the request, so the hook can narrow but never widen, and the
-acknowledgment reports the result so the client can see what it got. Any
-other exception refuses the subscription: a raising policy never grants.
-
-The decision 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` is not the
-tool for it: it ends every open stream at once (server shutdown).
+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.
 """
 
 
@@ -162,24 +162,19 @@ def _safe_unsubscribe(unsubscribe: Callable[[], None]) -> None:
         logger.exception("bus unsubscribe raised; continuing stream cleanup")
 
 
-def _honored_subset(requested: SubscriptionFilter, granted: SubscriptionFilter) -> SubscriptionFilter:
+def _honored_subset(requested: SubscriptionFilter) -> SubscriptionFilter:
     """The subset of `requested` the server will deliver, for the ack.
 
-    A kind is honored only when both the request and the grant carry it, and
-    the URI list keeps the requested URIs the grant also names, so a grant can
-    narrow the request but never widen it. With no narrowing policy the grant
-    is the request itself and nothing is dropped. Whether an honored kind ever
-    fires depends on what the server publishes, exactly as a subscription to
-    a nonexistent resource URI is honored and never fires. Non-true flags and
+    Every requested kind is honored - whether an event kind ever fires
+    depends on what the server publishes, exactly as a subscription to a
+    nonexistent resource URI is honored and never fires. Non-true flags and
     an empty URI list are dropped rather than echoed as falsy values.
     """
-    granted_uris = frozenset(granted.resource_subscriptions or ())
-    uris = [uri for uri in requested.resource_subscriptions or () if uri in granted_uris]
     return SubscriptionFilter(
-        tools_list_changed=True if requested.tools_list_changed and granted.tools_list_changed else None,
-        prompts_list_changed=True if requested.prompts_list_changed and granted.prompts_list_changed else None,
-        resources_list_changed=True if requested.resources_list_changed and granted.resources_list_changed else None,
-        resource_subscriptions=uris or None,
+        tools_list_changed=True if requested.tools_list_changed else None,
+        prompts_list_changed=True if requested.prompts_list_changed else None,
+        resources_list_changed=True if requested.resources_list_changed else None,
+        resource_subscriptions=list(requested.resource_subscriptions) if requested.resource_subscriptions else None,
     )
 
 
@@ -196,27 +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.
 
-    Without a `narrow` hook every requested kind and resource URI is
+    Without an `authorize` hook every requested kind and resource URI is
     honored: any caller may watch any URI the server publishes. Pass one to
-    grant per request - see `NarrowSubscription`. `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).
+    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,
         *,
-        narrow: NarrowSubscription | None = None,
+        authorize: AuthorizeSubscription | None = None,
         max_subscriptions: int = 1024,
         max_buffered_events: int = 1024,
     ) -> None:
         self._bus = bus
-        self._narrow = narrow
+        self._authorize = authorize
         self._max_subscriptions = max_subscriptions
         self._max_buffered_events = max_buffered_events
         self._streams: set[anyio.streams.memory.MemoryObjectSendStream[ServerEvent]] = set()
@@ -230,10 +225,10 @@ async def __call__(
         subscription_id = ctx.request_id
         if subscription_id is None:
             raise MCPError(INVALID_REQUEST, "subscriptions/listen requires a request id")
-        granted = await self._grant(ctx, params.notifications)
+        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, granted)
+        honored = _honored_subset(params.notifications)
         honored_uris = frozenset(honored.resource_subscriptions or ())
         meta: dict[str, Any] = {SUBSCRIPTION_ID_META_KEY: subscription_id}
 
@@ -281,21 +276,21 @@ def deliver(event: ServerEvent) -> None:
             recv.close()
         return SubscriptionsListenResult(_meta=meta)
 
-    async def _grant(self, ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
-        """The filter to grant this request: the request itself unless a `narrow` hook decides.
+    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 granted - so a broken policy cannot hand out access.
+        refused, never opened - so a broken policy cannot hand out access.
         """
-        if self._narrow is None:
-            return requested
+        if self._authorize is None:
+            return
         try:
-            return await self._narrow(ctx, requested)
+            await self._authorize(ctx, requested)
         except MCPError:
             raise
         except Exception as exc:  # deny-on-error: a raising policy must never grant
-            logger.exception("subscription narrow hook raised; refusing subscriptions/listen")
+            logger.exception("subscription authorize hook raised; refusing subscriptions/listen")
             raise MCPError(INTERNAL_ERROR, "Subscription authorization failed") from exc
 
     def close(self) -> None:
diff --git a/tests/server/test_subscriptions.py b/tests/server/test_subscriptions.py
index 217b36b93d..2613796884 100644
--- a/tests/server/test_subscriptions.py
+++ b/tests/server/test_subscriptions.py
@@ -284,108 +284,53 @@ async def run() -> None:
 
 
 @pytest.mark.anyio
-async def test_narrow_hook_narrows_the_ack_and_the_delivered_events() -> None:
-    """SDK-defined: a `narrow` hook grants a subset - it can drop a kind and
-    prune the URI list - and the ack and the stream both reflect the grant."""
+async def test_authorize_hook_accepts_and_the_request_is_honored_unchanged() -> None:
+    """SDK-defined: an `authorize` hook that returns accepts the request as
+    asked - it sees the requested filter and the ack echoes it unmodified."""
     bus = InMemorySubscriptionBus()
     seen: list[SubscriptionFilter] = []
 
-    async def narrow(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+    async def allow(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> None:
         seen.append(requested)
-        return SubscriptionFilter(resources_list_changed=True, resource_subscriptions=["r://mine"])
 
-    handler = ListenHandler(bus, narrow=narrow)
+    handler = ListenHandler(bus, authorize=allow)
     session = _RecordingSession()
 
     async with anyio.create_task_group() as tg:
 
         async def run() -> None:
-            await handler(
-                _ctx(session),
-                _params(
-                    tools_list_changed=True,
-                    resources_list_changed=True,
-                    resource_subscriptions=["r://mine", "r://theirs"],
-                ),
-            )
+            await handler(_ctx(session), _params(tools_list_changed=True, resource_subscriptions=["r://a", "r://b"]))
 
         tg.start_soon(run)
         await session.wait_for(1)
-        assert seen == [
-            SubscriptionFilter(
-                tools_list_changed=True, resources_list_changed=True, resource_subscriptions=["r://mine", "r://theirs"]
-            )
-        ]
+        assert seen == [SubscriptionFilter(tools_list_changed=True, resource_subscriptions=["r://a", "r://b"])]
 
         ack, _ = session.sent[0]
         assert isinstance(ack, SubscriptionsAcknowledgedNotification)
-        # tools_list_changed was requested but not granted, and r://theirs was pruned.
         assert ack.params.notifications == SubscriptionFilter(
-            resources_list_changed=True, resource_subscriptions=["r://mine"]
-        )
-
-        await bus.publish(ToolsListChanged())  # requested but not granted
-        await bus.publish(ResourceUpdated(uri="r://theirs"))  # pruned URI
-        await bus.publish(ResourcesListChanged())
-        await bus.publish(ResourceUpdated(uri="r://mine"))
-        await session.wait_for(3)
-        handler.close()
-
-    delivered = [notification for notification, _ in session.sent[1:]]
-    assert isinstance(delivered[0], ResourceListChangedNotification)
-    assert isinstance(delivered[1], ResourceUpdatedNotification)
-    assert delivered[1].params.uri == "r://mine"
-    assert len(delivered) == 2
-
-
-@pytest.mark.anyio
-async def test_narrow_hook_can_only_narrow_never_widen() -> None:
-    """SDK-defined: the grant is intersected with the request, so a hook returning
-    kinds or URIs the client never asked for cannot cause them to be delivered -
-    the spec's never-send-unrequested-types rule holds by construction."""
-    bus = InMemorySubscriptionBus()
-
-    async def widen(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
-        return SubscriptionFilter(
-            tools_list_changed=True, prompts_list_changed=True, resource_subscriptions=["r://a", "r://extra"]
+            tools_list_changed=True, resource_subscriptions=["r://a", "r://b"]
         )
 
-    handler = ListenHandler(bus, narrow=widen)
-    session = _RecordingSession()
-
-    async with anyio.create_task_group() as tg:
-
-        async def run() -> None:
-            await handler(_ctx(session), _params(tools_list_changed=True, resource_subscriptions=["r://a"]))
-
-        tg.start_soon(run)
-        await session.wait_for(1)
-
-        ack, _ = session.sent[0]
-        assert isinstance(ack, SubscriptionsAcknowledgedNotification)
-        assert ack.params.notifications == SubscriptionFilter(tools_list_changed=True, resource_subscriptions=["r://a"])
-
-        await bus.publish(PromptsListChanged())  # granted but never requested
-        await bus.publish(ResourceUpdated(uri="r://extra"))  # granted but never requested
-        await bus.publish(ToolsListChanged())
+        await bus.publish(ResourceUpdated(uri="r://b"))
         await session.wait_for(2)
         handler.close()
 
     delivered = [notification for notification, _ in session.sent[1:]]
-    assert isinstance(delivered[0], ToolListChangedNotification)
+    assert isinstance(delivered[0], ResourceUpdatedNotification)
+    assert delivered[0].params.uri == "r://b"
     assert len(delivered) == 1
 
 
 @pytest.mark.anyio
-async def test_narrow_hook_mcp_error_refuses_the_subscription_pre_ack() -> None:
+async def test_authorize_hook_mcp_error_refuses_the_subscription_pre_ack() -> None:
     """SDK-defined: a hook that raises `MCPError` refuses the whole request -
     the error reaches the client and no ack or stream is opened."""
     bus = _SpyBus()
 
-    async def refuse(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+    async def refuse(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> None:
         raise MCPError(INVALID_REQUEST, "not allowed to watch these resources")
 
-    handler = ListenHandler(bus, narrow=refuse)
+    handler = ListenHandler(bus, authorize=refuse)
     session = _RecordingSession()
 
     with pytest.raises(MCPError) as exc_info:
@@ -396,15 +341,15 @@ async def refuse(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFil
 
 
 @pytest.mark.anyio
-async def test_a_broken_narrow_hook_fails_closed(caplog: pytest.LogCaptureFixture) -> None:
+async def test_a_broken_authorize_hook_fails_closed(caplog: pytest.LogCaptureFixture) -> None:
     """SDK-defined: a hook that raises anything other than `MCPError` is a
     policy error and must never grant - the subscription is refused with an
     internal error rather than falling back to honoring the request."""
 
-    async def broken(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> SubscriptionFilter:
+    async def broken(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> None:
         raise RuntimeError("policy database unavailable")
 
-    handler = ListenHandler(InMemorySubscriptionBus(), narrow=broken)
+    handler = ListenHandler(InMemorySubscriptionBus(), authorize=broken)
     session = _RecordingSession()
 
     with pytest.raises(MCPError) as exc_info:
@@ -412,7 +357,7 @@ async def broken(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFil
     assert exc_info.value.error.code == INTERNAL_ERROR
     assert exc_info.value.error.message == "Subscription authorization failed"
     assert session.sent == []
-    assert "subscription narrow hook raised" in caplog.text
+    assert "subscription authorize hook raised" in caplog.text
 
 
 @pytest.mark.anyio