Make the subscription bus async and fix review findings

Review follow-ups to the subscriptions/listen runtime:

- Rename EventBus to SubscriptionBus: it carries exactly the four
  subscription event kinds, and the generic name collided conceptually
  with the unrelated EventStore in the same package.
- Make SubscriptionBus.publish async. Backend implementations (Redis,
  NATS) do network I/O on publish; a sync protocol would force them to
  block the loop or spawn their own tasks. This also makes the Context
  notify_* methods genuinely async, so they cannot be called from sync
  handlers (which run on worker threads where waking the listen streams
  is unsafe) - the illegal context is now unrepresentable instead of
  guarded.
- Subscribe each listen stream to the bus before sending the ack: an
  event published while the ack write was suspended used to be lost
  after the client had been told the subscription was live. The ack is
  still the first frame - the handler task alone writes the stream and
  only drains the buffer after the ack send returns.
- Register bus listeners under per-subscription tokens: equal callables
  (e.g. the same bound method subscribed twice) used to collapse in the
  set, so one unsubscribe silently detached both registrations.
- Drop the stdio claim from ListenHandler's docstring (no 2026-era
  stdio serving exists yet) and document that resource-updated URIs are
  matched as exact strings and do not reach legacy resources/subscribe
  subscribers.

Author: maxisbey

docs/advanced/low-level-server.md

@@ -183,7 +183,7 @@ Each of these is one idea you now have the vocabulary for; each has its own chap
 
 * `on_call_tool`, `on_get_prompt`, and `on_read_resource` may return an `InputRequiredResult` instead of their normal result to pause the call and ask the client for input; see **[Multi-round-trip requests](multi-round-trip.md)**. True to this tier, nothing is installed for you: where `MCPServer` seals `requestState` by default, here the `request_state` you set crosses the wire exactly as written until you opt in with `server.middleware.append(RequestStateBoundary(RequestStateSecurity(keys=[...]), default_audience=server.name))`: one line (both names import from `mcp.server.request_state`) for the identical sealing and verification `MCPServer` performs (**[Protecting `requestState`](multi-round-trip.md#protecting-requeststate)**).
 * `on_list_resources`, `on_read_resource`, `on_list_prompts`, `on_get_prompt`, `on_completion` are the same `(ctx, params) -> result` shape for the other primitives.
-* `on_subscriptions_listen` serves the 2026-07-28 `subscriptions/listen` stream. Pass an `mcp.server.subscriptions.ListenHandler` built over an `EventBus` (the in-memory default, or your own — e.g. Redis-backed), keep the bus where your other handlers can reach it (the lifespan is a natural home), and publish `ServerEvent`s to it. The handler owns the wire semantics: ack-first, per-stream filtering, and subscription-id tagging.
+* `on_subscriptions_listen` serves the 2026-07-28 `subscriptions/listen` stream. Pass an `mcp.server.subscriptions.ListenHandler` built over a `SubscriptionBus` (the in-memory default, or your own — e.g. Redis-backed), keep the bus where your other handlers can reach it (the lifespan is a natural home), and publish `ServerEvent`s to it. The handler owns the wire semantics: ack-first, per-stream filtering, and subscription-id tagging.
 * `server.streamable_http_app()` returns the same Starlette app `MCPServer`'s does; deploy it the way **[Running your server](../run/index.md)** deploys any other ASGI app. There is no `server.run(transport=...)` down here: `server.run(read_stream, write_stream, server.create_initialization_options())` drives one connection over a pair of streams, and that one line is the whole story.
 
 ## Recap

docs/tutorial/context.md

@@ -104,7 +104,7 @@ What a server offers is not fixed at import time. Register a tool at runtime, th
 
 The siblings are `send_resource_list_changed()`, `send_prompt_list_changed()`, and `send_resource_updated(uri)` for a change to one specific resource.
 
-On a 2026-07-28 connection, clients receive change notifications only on a `subscriptions/listen` stream they opened. The `Context` publish methods — `ctx.notify_tools_changed()`, `ctx.notify_prompts_changed()`, `ctx.notify_resources_changed()`, and `ctx.notify_resource_updated(uri)` — deliver to every subscribed stream at once, and are synchronous (no `await`). Behind a load balancer, pass your own `EventBus` implementation as `MCPServer(subscriptions=...)` to fan events out across replicas; the in-process default covers a single server.
+On a 2026-07-28 connection, clients receive change notifications only on a `subscriptions/listen` stream they opened — the `send_*` methods above do not reach those streams. The `Context` publish methods — `await ctx.notify_tools_changed()`, `await ctx.notify_prompts_changed()`, `await ctx.notify_resources_changed()`, and `await ctx.notify_resource_updated(uri)` — deliver to every subscribed stream at once. Behind a load balancer, pass your own `SubscriptionBus` implementation as `MCPServer(subscriptions=...)` to fan events out across replicas; the in-process default covers a single server.
 
 !!! check
     Before anyone runs `enable_recommendations`, the tool you are promising does not exist. Call it

src/mcp/server/mcpserver/context.py

@@ -16,7 +16,12 @@
     elicit_with_validation,
 )
 from mcp.server.lowlevel.helper_types import ReadResourceContents
-from mcp.server.subscriptions import PromptsListChanged, ResourcesListChanged, ResourceUpdated, ToolsListChanged
+from mcp.server.subscriptions import (
+    PromptsListChanged,
+    ResourcesListChanged,
+    ResourceUpdated,
+    ToolsListChanged,
+)
 from mcp.shared.exceptions import MCPDeprecationWarning
 
 if TYPE_CHECKING:
@@ -110,21 +115,27 @@ async def report_progress(self, progress: float, total: float | None = None, mes
         """
         await self.request_context.session.report_progress(progress, total, message)
 
-    def notify_tools_changed(self) -> None:
+    async def notify_tools_changed(self) -> None:
         """Publish a tools list-changed event to `subscriptions/listen` subscribers."""
-        self.mcp_server.subscriptions.publish(ToolsListChanged())
+        await self.mcp_server.subscriptions.publish(ToolsListChanged())
 
-    def notify_prompts_changed(self) -> None:
+    async def notify_prompts_changed(self) -> None:
         """Publish a prompts list-changed event to `subscriptions/listen` subscribers."""
-        self.mcp_server.subscriptions.publish(PromptsListChanged())
+        await self.mcp_server.subscriptions.publish(PromptsListChanged())
 
-    def notify_resources_changed(self) -> None:
+    async def notify_resources_changed(self) -> None:
         """Publish a resources list-changed event to `subscriptions/listen` subscribers."""
-        self.mcp_server.subscriptions.publish(ResourcesListChanged())
+        await self.mcp_server.subscriptions.publish(ResourcesListChanged())
+
+    async def notify_resource_updated(self, uri: str | AnyUrl) -> None:
+        """Publish a resource-updated event for `uri` to `subscriptions/listen` subscribers.
 
-    def notify_resource_updated(self, uri: str | AnyUrl) -> None:
-        """Publish a resource-updated event for `uri` to `subscriptions/listen` subscribers."""
-        self.mcp_server.subscriptions.publish(ResourceUpdated(uri=str(uri)))
+        The URI is matched as an exact string against each stream's filter.
+        Reaches `subscriptions/listen` streams only; clients on earlier
+        protocol versions that used `resources/subscribe` are notified via
+        `ctx.session.send_resource_updated(uri)` instead.
+        """
+        await self.mcp_server.subscriptions.publish(ResourceUpdated(uri=str(uri)))
 
     async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]:
         """Read a resource by URI.

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 StreamableHTTPSessionManager
-from mcp.server.subscriptions import EventBus, InMemoryEventBus, ListenHandler
+from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
 from mcp.shared.uri_template import UriTemplate
@@ -183,7 +183,7 @@ def __init__(
         resource_security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
         request_state_security: RequestStateSecurity | None = None,
         cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
-        subscriptions: EventBus | None = None,
+        subscriptions: SubscriptionBus | None = None,
     ):
         self._resource_security = resource_security
         self.settings = Settings(
@@ -204,9 +204,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 `EventBus` implementation over an external pub/sub
+        # in-process; pass an `SubscriptionBus` implementation over an external pub/sub
         # backend to fan events out across replicas.
-        self._subscriptions: EventBus = subscriptions if subscriptions is not None else InMemoryEventBus()
+        self._subscriptions: SubscriptionBus = subscriptions if subscriptions is not None else InMemorySubscriptionBus()
         self._lowlevel_server = Server(
             name=name or "mcp-server",
             title=title,
@@ -293,7 +293,7 @@ def version(self) -> str | None:
         return self._lowlevel_server.version
 
     @property
-    def subscriptions(self) -> EventBus:
+    def subscriptions(self) -> SubscriptionBus:
         """The `subscriptions/listen` event bus.
 
         Publish a `ServerEvent` here (or via the `Context.notify_*` methods)

src/mcp/server/subscriptions.py

@@ -4,11 +4,11 @@
 server events by sending a `subscriptions/listen` request whose response IS
 the stream. This module provides the two pieces a server needs:
 
-- `EventBus`: the pluggable fan-out seam. The bus carries typed `ServerEvent`
+- `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 `InMemoryEventBus`.
+  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=`.
@@ -81,19 +81,20 @@ class ResourceUpdated:
 """An event a server publishes for delivery to listen subscribers."""
 
 
-class EventBus(Protocol):
+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.
 
-    Both methods are synchronous and must be called from the server's event
-    loop thread. Listeners must not raise.
+    `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.
     """
 
-    def publish(self, event: ServerEvent) -> None:
+    async def publish(self, event: ServerEvent) -> None:
         """Deliver `event` to every subscribed listener."""
         ...
 
@@ -102,23 +103,26 @@ def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], Non
         ...
 
 
-class InMemoryEventBus:
-    """In-process `EventBus`: synchronous fan-out to a set of listeners."""
+class InMemorySubscriptionBus:
+    """In-process `SubscriptionBus`: synchronous fan-out to listeners in subscription order."""
 
     def __init__(self) -> None:
-        self._listeners: set[Callable[[ServerEvent], None]] = set()
+        # 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]] = {}
 
-    def publish(self, event: ServerEvent) -> None:
+    async def publish(self, event: ServerEvent) -> None:
         """Deliver `event` to every subscribed listener."""
-        for listener in list(self._listeners):
+        for listener in list(self._listeners.values()):
             listener(event)
 
     def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]:
         """Register `listener` and return an idempotent unsubscribe callable."""
-        self._listeners.add(listener)
+        token = object()
+        self._listeners[token] = listener
 
         def unsubscribe() -> None:
-            self._listeners.discard(listener)
+            self._listeners.pop(token, None)
 
         return unsubscribe
 
@@ -172,10 +176,10 @@ class ListenHandler:
     contract) or `close` ends all streams gracefully.
 
     Requires a transport that can stream a request's response (streamable
-    HTTP's SSE mode, stdio).
+    HTTP's SSE mode).
     """
 
-    def __init__(self, bus: EventBus) -> None:
+    def __init__(self, bus: SubscriptionBus) -> None:
         self._bus = bus
         self._streams: set[anyio.streams.memory.MemoryObjectSendStream[ServerEvent]] = set()
 
@@ -191,14 +195,6 @@ async def __call__(
         honored = _honored_subset(params.notifications)
         meta: dict[str, Any] = {SUBSCRIPTION_ID_META_KEY: subscription_id}
 
-        # Ack first, subscribe second: no event can precede the ack frame.
-        await ctx.session.send_notification(
-            SubscriptionsAcknowledgedNotification(
-                params=SubscriptionsAcknowledgedNotificationParams(notifications=honored, _meta=meta)
-            ),
-            related_request_id=subscription_id,
-        )
-
         # Unbounded buffer so publishers never block on a slow consumer (the
         # transport write happens in this handler task, not the publisher's).
         send, recv = anyio.create_memory_object_stream[ServerEvent](math.inf)
@@ -208,12 +204,22 @@ def deliver(event: ServerEvent) -> None:
                 try:
                     send.send_nowait(event)
                 except anyio.ClosedResourceError:
-                    # `aclose` closed this stream; the loop below is unwinding.
+                    # `close` closed this stream; the loop below is unwinding.
                     pass
 
+        # 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

tests/server/mcpserver/test_server.py

@@ -52,7 +52,7 @@
 from mcp.server.mcpserver.resources import FileResource, FunctionResource
 from mcp.server.mcpserver.utilities.types import Audio, Image
 from mcp.server.subscriptions import (
-    InMemoryEventBus,
+    InMemorySubscriptionBus,
     PromptsListChanged,
     ResourcesListChanged,
     ResourceUpdated,
@@ -2259,8 +2259,8 @@ async def probe(ctx: Context) -> str:
 
 
 def test_subscriptions_bus_defaults_to_in_memory_and_accepts_custom() -> None:
-    assert isinstance(MCPServer().subscriptions, InMemoryEventBus)
-    bus = InMemoryEventBus()
+    assert isinstance(MCPServer().subscriptions, InMemorySubscriptionBus)
+    bus = InMemorySubscriptionBus()
     assert MCPServer(subscriptions=bus).subscriptions is bus
 
 
@@ -2270,11 +2270,11 @@ async def test_context_notify_methods_publish_to_the_subscriptions_bus() -> None
     mcp.subscriptions.subscribe(seen.append)
 
     @mcp.tool()
-    def touch(ctx: Context) -> str:
-        ctx.notify_tools_changed()
-        ctx.notify_prompts_changed()
-        ctx.notify_resources_changed()
-        ctx.notify_resource_updated("r://x")
+    async def touch(ctx: Context) -> str:
+        await ctx.notify_tools_changed()
+        await ctx.notify_prompts_changed()
+        await ctx.notify_resources_changed()
+        await ctx.notify_resource_updated("r://x")
         return "ok"
 
     with anyio.fail_after(5):