@@ -17,7 +17,7 @@ Your side of it is one line: publish the change.
 * The siblings are `notify_prompts_changed()` and `notify_resources_changed()`.
 * No subscribers, no work. Publishing to an idle server is a no-op, so you never check whether anyone is listening. You state what changed.
 
-`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.
+`MCPServer` serves `subscriptions/listen` for you, unless you [turn it off](#turning-it-off). The wire obligations (the acknowledgment as the first frame, per-stream filtering, the subscription id on every frame) are the SDK's job.
 
 !!! check
     On the wire, a stream whose filter named `board://sprint` looks like this after `complete_task` runs:
@@ -74,7 +74,32 @@ Entering `client.listen(...)` sends the request and waits for your acknowledgmen
 
 Publishes travel from your handler to the open streams over a `SubscriptionBus`. The default is in-memory: one process, every stream in it. That is the right answer until you run replicas behind a load balancer, because then a client's stream is pinned to one replica, and a publish on another replica has to reach it.
 
-That seam is yours to implement: two methods over your pub/sub backend.
+With the default bus it can't, because every replica has its own:
+
+```mermaid
+flowchart LR
+    client[Client] --> lb[Load balancer]
+    lb --> stream
+    lb ~~~~ gap
+    lb --> tool
+    subgraph B [Replica B]
+        tool[tools/call] -- publishes --> busB[(bus B)]
+    end
+    gap[(no shared bus)]
+    subgraph A [Replica A]
+        stream[listen stream] -- subscribed --> busA[(bus A)]
+    end
+    style A fill:none
+    style B fill:none
+    style gap fill:none,stroke-dasharray:4 4
+```
+
+Nothing fails: the call succeeds, and the stream stays silent. So behind a load balancer, pick one:
+
+* **You need change notifications.** Give every replica the same bus, below.
+* **You don't.** [Turn them off](#turning-it-off), so no client is promised events it will miss or holds a stream open for them.
+
+The shared bus is yours to implement: two methods over your pub/sub backend.
 
 ```python
 from collections.abc import Callable
@@ -123,6 +148,20 @@ async def tools_reloaded() -> None:
     await bus.publish(ToolsListChanged())  # from a lifespan task, a webhook, anywhere
 ```
 
+## Turning it off
+
+A server whose catalog never changes has nothing to publish. Say so when you build it:
+
+```python title="server.py" hl_lines="3"
+--8<-- "docs_src/subscriptions/tutorial007.py"
+```
+
+* A `2026-07-28` client sees no change notifications advertised, and a `subscriptions/listen` request gets *Method not found* instead of an open stream.
+* `ctx.notify_*` still works and reaches nobody, so your handlers don't change.
+* Clients on earlier protocol versions see no difference.
+
+An open stream is a request that never finishes, so this also matters on a host that bills by request duration.
+
 ## The low-level composition
 
 Down on the low-level `Server` there is no pre-wired anything, and the same parts assemble in three lines:
@@ -143,5 +182,6 @@ Down on the low-level `Server` there is no pre-wired anything, and the same part
 * The client end is `async with client.listen(...)`: **[Subscriptions](../client/subscriptions.md)** under *Clients* is that story.
 * On the low-level `Server` you assemble the same parts yourself: a bus, `ListenHandler(bus)`, the `on_subscriptions_listen` slot.
 * Scaling out means implementing `SubscriptionBus`, two methods, and passing it as `MCPServer(subscriptions=...)`.
+* Nothing to publish, or replicas with no shared bus: `MCPServer(subscriptions=False)` advertises no change notifications and holds no stream.
 
 Running the server that serves all this, behind one replica or twenty, is **[Deploy & scale](../run/deploy.md)**.
@@ -168,6 +168,7 @@ Nothing about the fan-out cares which server object a stream is attached to. Two
 * Across real processes, **the SDK ships no bus that can help you.** `SubscriptionBus` is a two-method `Protocol` (`publish` and `subscribe`) that you implement over your own pub/sub backend (Redis, NATS, whatever you already run) and pass as `MCPServer(subscriptions=...)`. **[Subscriptions](../handlers/subscriptions.md#scaling-past-one-process)** has the sketch and the contract.
 * The bus carries four small typed events, never JSON-RPC. Acknowledgment, filtering, and stream lifecycle stay in the SDK, so your bus cannot break the protocol; it can only move events between processes.
 * Streams are **not** resumable and events are **not** replayed. Losing a replica drops its streams; the clients re-listen and re-fetch. There is no event store to share and nothing else to configure. This is the one place where scaling out is genuinely just more of the same.
+* A server that needs no change notifications skips the bus: **[turn them off](../handlers/subscriptions.md#turning-it-off)**.
 
 ## What the SDK does not give you
 
 
@@ -0,0 +1,9 @@
+from mcp.server.mcpserver import MCPServer
+
+mcp = MCPServer("Unit Converter", subscriptions=False)
+
+
+@mcp.tool()
+def km_to_miles(km: float) -> float:
+    """Convert kilometres to miles."""
+    return km * 0.621371
@@ -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
 
@@ -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.
 
 
@@ -1,11 +1,13 @@
 """`docs/{handlers,client}/subscriptions.md`: every claim the two pages make, proved against the real SDK."""
 
 from collections.abc import Awaitable, Callable
-from typing import Any
+from typing import Any, Literal
 
 import anyio
 import mcp_types as types
 import pytest
+from inline_snapshot import snapshot
+from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
 from trio.testing import MockClock
 
 from docs_src.subscriptions import (
@@ -17,14 +19,16 @@
     tutorial004_trio,
     tutorial005,
     tutorial006,
+    tutorial007,
 )
 from mcp import Client
 from mcp.server.auth.middleware.auth_context import auth_context_var
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
 from mcp.server.context import ServerRequestContext
 from mcp.server.lowlevel import Server
-from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, ListenHandler, ToolsListChanged
+from mcp.server.mcpserver import MCPServer
+from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, ListenHandler, ResourceUpdated, ToolsListChanged
 from mcp.shared.exceptions import MCPError
 
 _ReadResource = Callable[
@@ -333,3 +337,62 @@ async def test_the_middleware_refuses_a_listen_the_caller_could_not_read() -> No
                 await client.read_resource("files://payroll.csv")
     finally:
         auth_context_var.reset(reset)
+
+
+def _replica(*, subscriptions: Literal[False] | None = None) -> MCPServer:
+    """One replica of the sprint board: its own server object running tutorial001's publishing tool."""
+    mcp = MCPServer("Sprint Board", subscriptions=subscriptions)
+    mcp.add_tool(tutorial001.complete_task)
+    return mcp
+
+
+async def test_a_publish_on_another_replica_never_reaches_the_stream() -> None:
+    """The load-balancer diagram: with the default bus, a stream on replica A misses a publish on replica B.
+
+    The stream subscribes to both boards, so B's event would be delivered if it crossed. B publishes
+    first, then A publishes a marker: the marker being the first event proves B's never arrived.
+    """
+    async with Client(_replica()) as on_a, Client(_replica()) as on_b:
+        async with on_a.listen(resource_subscriptions=["board://sprint", "board://backlog"]) as sub:
+            routed_to_b = await on_b.call_tool("complete_task", {"board": "sprint", "task": "design"})
+            await on_a.call_tool("complete_task", {"board": "backlog", "task": "tidy docs"})
+            with anyio.fail_after(5):
+                first_event = await anext(sub)
+            assert first_event == ResourceUpdated(uri="board://backlog")
+        assert routed_to_b.is_error is False  # "Nothing fails": the call on B itself succeeded
+
+
+async def test_a_server_with_subscriptions_off_advertises_nothing_and_refuses_listen() -> None:
+    """tutorial007: a 2026-07-28 client sees every change-notification flag false, and listen is method-not-found."""
+    async with Client(tutorial007.mcp) as client:
+        assert client.protocol_version in MODERN_PROTOCOL_VERSIONS
+        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 == types.METHOD_NOT_FOUND
+        # The refusal is in-band: the connection carries on and the tool still answers.
+        result = await client.call_tool("km_to_miles", {"km": 10})
+        assert result.structured_content == {"result": 6.21371}
+
+
+async def test_a_publishing_handler_runs_unchanged_with_subscriptions_off() -> None:
+    """`ctx.notify_*` still works with `subscriptions=False`: tutorial001's publishing tool needs no change."""
+    async with Client(_replica(subscriptions=False)) as client:
+        result = await client.call_tool("complete_task", {"board": "sprint", "task": "design"})
+    assert result.is_error is False
+    assert result.content == [types.TextContent(type="text", text="design: done")]
+
+
+async def test_subscriptions_off_changes_nothing_for_an_earlier_protocol_version() -> None:
+    """A client on the initialize handshake reads the same capabilities whether subscriptions are off or on."""
+    async with Client(_replica(subscriptions=False), mode="legacy") as off, Client(_replica(), mode="legacy") as on:
+        assert off.protocol_version == on.protocol_version
+        assert off.protocol_version in HANDSHAKE_PROTOCOL_VERSIONS
+        assert off.server_capabilities == on.server_capabilities
@@ -16,6 +16,7 @@
     INTERNAL_ERROR,
     INVALID_PARAMS,
     INVALID_REQUEST,
+    METHOD_NOT_FOUND,
     MISSING_REQUIRED_CLIENT_CAPABILITY,
     AudioContent,
     BlobResourceContents,
@@ -3180,6 +3181,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