Let MCPServer opt out of serving subscriptions/listen

`MCPServer` always registered a `subscriptions/listen` handler, so on the
2026-07-28 wire every server advertised `listChanged` and
`resources.subscribe` as true and held a listen stream open for any client
that asked, even when it never publishes a change. There was no supported
way to turn that off.

The existing `subscriptions` keyword now also accepts `False`. The server
then registers no listen handler: `server/discover` reports the
change-notification flags as false and a `subscriptions/listen` request is
answered with method-not-found. `ctx.notify_*` stays callable and reaches
nobody. `None` and a bus keep their meaning, so the default is unchanged.

Fixes #3357

Author: maxisbey

docs/handlers/subscriptions.md

@@ -19,6 +19,8 @@ Your side of it is one line: publish the change.
 
 `MCPServer` serves `subscriptions/listen` for you. The wire obligations (the acknowledgment as the first frame, per-stream filtering, the subscription id on every frame) are the SDK's job.
 
+A server that never publishes can opt out with `MCPServer("Static", subscriptions=False)`: it advertises no change notifications, and a `subscriptions/listen` request gets a *Method not found* error instead of an open stream.
+
 !!! check
     On the wire, a stream whose filter named `board://sprint` looks like this after `complete_task` runs:
 

src/mcp/server/mcpserver/server.py

@@ -181,7 +181,7 @@ def __init__(
         resource_security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
         request_state_security: RequestStateSecurity | None = None,
         cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
-        subscriptions: SubscriptionBus | None = None,
+        subscriptions: SubscriptionBus | Literal[False] | None = None,
         middleware: Sequence[ServerMiddleware[Any]] | None = None,
     ):
         self._resource_security = resource_security
@@ -204,8 +204,11 @@ 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.
-        self._subscriptions: SubscriptionBus = subscriptions if subscriptions is not None else InMemorySubscriptionBus()
+        # backend to fan events out across replicas. `False` leaves `subscriptions/listen`
+        # unserved; the bus stays so `ctx.notify_*` keeps working, with nobody listening.
+        self._subscriptions: SubscriptionBus = (
+            InMemorySubscriptionBus() if subscriptions is None or subscriptions is False else subscriptions
+        )
         self._lowlevel_server = Server(
             name=name or "mcp-server",
             title=title,
@@ -222,7 +225,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=None if subscriptions is False else ListenHandler(self._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

src/mcp/server/subscriptions.py

@@ -10,8 +10,8 @@
   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=`.
+  `MCPServer` registers one unless constructed with `subscriptions=False`;
+  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.
 

tests/server/mcpserver/test_server.py

@@ -13,6 +13,7 @@
     INTERNAL_ERROR,
     INVALID_PARAMS,
     INVALID_REQUEST,
+    METHOD_NOT_FOUND,
     MISSING_REQUIRED_CLIENT_CAPABILITY,
     AudioContent,
     BlobResourceContents,
@@ -3109,6 +3110,61 @@ async def ask(ctx: Context) -> str:
     assert seen == [ToolsListChanged(), ResourcesListChanged(), PromptsListChanged()]
 
 
+async def test_server_advertises_change_notifications_on_the_modern_wire_by_default() -> None:
+    """SDK-defined: a default `MCPServer` serves `subscriptions/listen`, so `server/discover`
+    reports every change-notification flag true."""
+    mcp = MCPServer("board")
+
+    with anyio.fail_after(5):
+        async with Client(mcp) as client:
+            assert client.server_capabilities.model_dump(by_alias=True, exclude_none=True) == snapshot(
+                {
+                    "prompts": {"listChanged": True},
+                    "resources": {"subscribe": True, "listChanged": True},
+                    "tools": {"listChanged": True},
+                }
+            )
+
+
+async def test_subscriptions_false_neither_advertises_nor_serves_listen() -> None:
+    """SDK-defined: `MCPServer(subscriptions=False)` registers no `subscriptions/listen` handler,
+    so `server/discover` reports every change-notification flag false and a listen request is
+    refused with method-not-found instead of opening a stream."""
+    mcp = MCPServer("static", subscriptions=False)
+
+    with anyio.fail_after(5):
+        async with Client(mcp) as client:
+            assert client.server_capabilities.model_dump(by_alias=True, exclude_none=True) == snapshot(
+                {
+                    "prompts": {"listChanged": False},
+                    "resources": {"subscribe": False, "listChanged": False},
+                    "tools": {"listChanged": False},
+                }
+            )
+            # Entering is where the request is sent; `__aenter__` directly avoids an unreachable with-body.
+            with pytest.raises(MCPError) as exc_info:
+                await client.listen(tools_list_changed=True).__aenter__()
+            assert exc_info.value.error.code == METHOD_NOT_FOUND
+
+
+async def test_notify_still_succeeds_when_subscriptions_are_off() -> None:
+    """SDK-defined: with `subscriptions=False` a handler that publishes a change completes as
+    usual; the event has no stream to reach."""
+    mcp = MCPServer("static", subscriptions=False)
+
+    @mcp.tool()
+    async def touch(ctx: Context) -> str:
+        await ctx.notify_tools_changed()
+        return "ok"
+
+    with anyio.fail_after(5):
+        async with Client(mcp) as client:
+            result = await client.call_tool("touch")
+
+    assert result.is_error is False
+    assert result.content == [TextContent(type="text", text="ok")]
+
+
 def test_context_mcp_server_outside_request_raises() -> None:
     with pytest.raises(ValueError, match="outside of a request"):
         _ = Context().mcp_server