diff --git a/docs/handlers/subscriptions.md b/docs/handlers/subscriptions.md index 99b1bc98a..7d1ad3c61 100644 --- a/docs/handlers/subscriptions.md +++ b/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 diff --git a/docs/migration.md b/docs/migration.md index 7b5950e7c..6cf4bc200 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`, `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. diff --git a/docs/whats-new.md b/docs/whats-new.md index 9826fa590..c9906746d 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), 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/connection.py b/src/mcp/server/connection.py index ca05928df..254fcbd24 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 33a0b08c8..9d081904f 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 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 diff --git a/src/mcp/server/subscriptions.py b/src/mcp/server/subscriptions.py index f5d3b5b0a..6a0266fd2 100644 --- a/src/mcp/server/subscriptions.py +++ b/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. diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index eb212dafd..f22979f46 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 579e3522f..261379688 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,83 @@ async def run() -> None: assert isinstance(session.sent[1][0], ToolListChangedNotification) +@pytest.mark.anyio +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 allow(ctx: ServerRequestContext[Any, Any], requested: SubscriptionFilter) -> None: + seen.append(requested) + + 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, resource_subscriptions=["r://a", "r://b"])) + + tg.start_soon(run) + await session.wait_for(1) + assert seen == [SubscriptionFilter(tools_list_changed=True, resource_subscriptions=["r://a", "r://b"])] + + ack, _ = session.sent[0] + assert isinstance(ack, SubscriptionsAcknowledgedNotification) + assert ack.params.notifications == SubscriptionFilter( + tools_list_changed=True, resource_subscriptions=["r://a", "r://b"] + ) + + 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], ResourceUpdatedNotification) + assert delivered[0].params.uri == "r://b" + assert len(delivered) == 1 + + +@pytest.mark.anyio +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) -> None: + raise MCPError(INVALID_REQUEST, "not allowed to watch these resources") + + handler = ListenHandler(bus, authorize=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_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) -> None: + raise RuntimeError("policy database unavailable") + + handler = ListenHandler(InMemorySubscriptionBus(), authorize=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 authorize 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."""