"""Server-side `subscriptions/listen` support (2026-07-28, SEP-2575). On the 2026-07-28 wire there is no standing GET stream: a client opts in to server events by sending a `subscriptions/listen` request whose response IS the stream. This module provides the two pieces a server needs: - `SubscriptionBus`: the pluggable fan-out seam. The bus carries typed `ServerEvent` values, not wire notifications - the listen handler owns subscription-id stamping and per-stream filtering, so a custom bus (e.g. backed by Redis pub/sub for multi-replica deployments) never sees JSON-RPC. The in-process default is `InMemorySubscriptionBus`. - `ListenHandler`: the request handler that serves `subscriptions/listen`. `MCPServer` registers one automatically; lowlevel `Server` users pass an instance as `on_subscriptions_listen=`. The event vocabulary lives in `mcp.shared.subscriptions`, shared with the client driver, and is re-exported here. Per the spec, the handler acknowledges first (the ack is the first frame on the stream), tags every frame with the listen request's JSON-RPC id under `_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. """ from __future__ import annotations import logging from collections.abc import Callable from typing import Any, Protocol import anyio import anyio.lowlevel import anyio.streams.memory from mcp_types import ( INTERNAL_ERROR, INVALID_REQUEST, SubscriptionFilter, SubscriptionsAcknowledgedNotification, SubscriptionsAcknowledgedNotificationParams, SubscriptionsListenRequestParams, SubscriptionsListenResult, ) from mcp.server.context import ServerRequestContext from mcp.shared.exceptions import MCPError from mcp.shared.subscriptions import ( SUBSCRIPTION_ID_META_KEY, PromptsListChanged, ResourcesListChanged, ResourceUpdated, ServerEvent, ToolsListChanged, event_matches, event_to_notification, ) __all__ = [ "SUBSCRIPTION_ID_META_KEY", "InMemorySubscriptionBus", "ListenHandler", "PromptsListChanged", "ResourceUpdated", "ResourcesListChanged", "ServerEvent", "SubscriptionBus", "ToolsListChanged", ] logger = logging.getLogger(__name__) class SubscriptionBus(Protocol): """Fan-out seam between event publishers and open listen streams. Implement this over an external pub/sub backend (Redis, NATS, ...) to fan events out across replicas: `publish` forwards the event to the backend, and each replica's bus invokes its local listeners for events arriving from the backend. The same instance can be shared across servers. `publish` is async so backend implementations can do network I/O. `subscribe` is synchronous local registration. Listeners are synchronous, must not raise, and are invoked on the server's event loop. """ async def publish(self, event: ServerEvent) -> None: """Deliver `event` to every subscribed listener.""" ... def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]: """Register `listener` and return an idempotent unsubscribe callable.""" ... class InMemorySubscriptionBus: """In-process `SubscriptionBus`: synchronous fan-out to listeners in subscription order.""" def __init__(self) -> None: # Keyed by a per-subscription token so the same callable can be # registered more than once (bound methods compare equal). self._listeners: dict[object, Callable[[ServerEvent], None]] = {} async def publish(self, event: ServerEvent) -> None: """Deliver `event` to every subscribed listener. A raising listener is logged and skipped: one bad listener must not starve the others or fail the publishing handler. Ends with a checkpoint so a burst of publishes from one task lets listen streams drain between events instead of overflowing their buffers unread. """ for listener in list(self._listeners.values()): try: listener(event) except Exception: # fan-out boundary: isolate listeners from each other logger.exception("subscription listener raised; continuing") await anyio.lowlevel.checkpoint() def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]: """Register `listener` and return an idempotent unsubscribe callable.""" token = object() self._listeners[token] = listener def unsubscribe() -> None: self._listeners.pop(token, None) return unsubscribe def _safe_unsubscribe(unsubscribe: Callable[[], None]) -> None: """Run a bus's unsubscribe callable, isolating the stream from it raising. The callable comes from a custom `SubscriptionBus`; a raising one is logged and skipped so it cannot stop the stream's own cleanup from releasing its subscription slot. """ try: unsubscribe() except Exception: # fan-out boundary: a raising bus must not skip stream cleanup logger.exception("bus unsubscribe raised; continuing stream cleanup") def _honored_subset(requested: 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 an empty URI list are dropped rather than echoed as falsy values. """ 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, ) class ListenHandler: """Serves `subscriptions/listen`: one call is one subscription stream. Register on a lowlevel `Server` via `on_subscriptions_listen=` (or `add_request_handler`); `MCPServer` does so automatically. Each call acknowledges the honored filter first, then forwards matching bus events onto the request's response stream until the client disconnects (which cancels the handler; the stream just ends, per the spec's abrupt-close contract) or `close` ends all streams gracefully. 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). """ def __init__(self, bus: SubscriptionBus, *, max_subscriptions: int = 1024, max_buffered_events: int = 1024) -> None: self._bus = bus self._max_subscriptions = max_subscriptions self._max_buffered_events = max_buffered_events self._streams: set[anyio.streams.memory.MemoryObjectSendStream[ServerEvent]] = set() async def __call__( self, ctx: ServerRequestContext[Any, Any], params: SubscriptionsListenRequestParams, ) -> SubscriptionsListenResult: """Serve one listen stream.""" subscription_id = ctx.request_id if subscription_id is None: raise MCPError(INVALID_REQUEST, "subscriptions/listen requires a request id") if len(self._streams) >= self._max_subscriptions: raise MCPError(INTERNAL_ERROR, "Subscription limit reached") honored = _honored_subset(params.notifications) honored_uris = frozenset(honored.resource_subscriptions or ()) meta: dict[str, Any] = {SUBSCRIPTION_ID_META_KEY: subscription_id} # Buffered so publishers don't block on a slow consumer (the transport # write happens in this handler task, not the publisher's). A stream # whose backlog hits the cap is ended - see the class docstring. send, recv = anyio.create_memory_object_stream[ServerEvent](self._max_buffered_events) def deliver(event: ServerEvent) -> None: if event_matches(honored, honored_uris, event): try: send.send_nowait(event) except anyio.ClosedResourceError: # `close` closed this stream; the loop below is unwinding. pass except anyio.WouldBlock: logger.warning("listen stream %r backlog full; ending the stream", subscription_id) # Release the subscription slot now: the handler's own # cleanup can be wedged in a transport write that closing # this buffer cannot wake (a client that stopped reading). self._streams.discard(send) send.close() # Subscribe before sending the ack so an event published while the # ack write is suspended is buffered rather than lost. The ack is # still the first frame: this task alone writes the stream, and it # only starts draining the buffer after the ack send returns. unsubscribe = self._bus.subscribe(deliver) self._streams.add(send) try: await ctx.session.send_notification( SubscriptionsAcknowledgedNotification( params=SubscriptionsAcknowledgedNotificationParams(notifications=honored, _meta=meta) ), related_request_id=subscription_id, ) async for event in recv: await ctx.session.send_notification( event_to_notification(event, meta), related_request_id=subscription_id ) finally: _safe_unsubscribe(unsubscribe) self._streams.discard(send) send.close() recv.close() return SubscriptionsListenResult(_meta=meta) def close(self) -> None: """Initiate graceful closure of every open listen stream. Each stream then drains its buffered events and sends its `SubscriptionsListenResult` (stamped with the subscription id) as the final frame from its own handler task - the spec's graceful closure flow, telling clients the stream ended deliberately rather than dropping. This method only initiates that; it does not wait for the streams to finish flushing. """ for stream in list(self._streams): stream.close()