Add the subscriptions docs page and example story

- docs/advanced/subscriptions.md: publish-first narrative for MCPServer
  (ctx.notify_*), the filter/no-replay contract, the SubscriptionBus seam
  for multi-replica deployments, and the low-level composition. Snippets
  under docs_src/subscriptions/ with their claims proved by
  tests/docs_src/test_subscriptions.py; cross-linked from the context
  tutorial and the low-level server page.
- examples/stories/subscriptions: promote the deferred stub to a runnable
  story (modern era, both transports, MCPServer and lowlevel variants).
  The client opens a listen stream through the session escape hatch,
  watches one URI and the tool list, observes exact-URI filtering and a
  runtime tool registration announcing itself, and closes the stream by
  cancelling the parked request.

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 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.
+* `on_subscriptions_listen` serves the 2026-07-28 `subscriptions/listen` stream. Pass a `ListenHandler` built over a `SubscriptionBus` and publish events to the bus from your other handlers; see **[Subscriptions](subscriptions.md)** for the full composition.
 * `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/advanced/subscriptions.md

@@ -0,0 +1,80 @@
+# Subscriptions
+
+A server's catalog is not fixed. Tools get registered at runtime, resources change behind their URIs. The client side of that story is a subscription: on the 2026-07-28 protocol, a client that wants to hear about changes sends one `subscriptions/listen` request, and the response to that request *is* the stream — it stays open, carrying exactly the notification kinds the client asked for.
+
+Your side of it is one line: publish the change.
+
+```python title="server.py" hl_lines="16 27"
+--8<-- "docs_src/subscriptions/tutorial001.py"
+```
+
+* `await ctx.notify_resource_updated("note://todo")` delivers `notifications/resources/updated` to every open listen stream that subscribed to that URI. Not to anyone else.
+* `await ctx.notify_tools_changed()` delivers `notifications/tools/list_changed` to every stream that asked for tool-list changes. A client that receives it calls `tools/list` again — and now sees `search`.
+* The siblings are `notify_prompts_changed()` and `notify_resources_changed()`, for the other two list-changed kinds.
+* No subscribers, no work: publishing to an idle server is a no-op. You don't check whether anyone is listening; you state what changed.
+
+The SDK serves `subscriptions/listen` for you — `MCPServer` registers the handler at construction, and the wire obligations (the acknowledgment as the first frame, the per-stream filtering, the subscription id tagged onto every frame) are its job, not yours.
+
+!!! check
+    On the wire, a stream whose filter named `note://todo` looks like this after `edit_note` runs:
+
+    ```json
+    {"method": "notifications/subscriptions/acknowledged",
+     "params": {"notifications": {"resourceSubscriptions": ["note://todo"]}, "_meta": {"io.modelcontextprotocol/subscriptionId": 7}}}
+
+    {"method": "notifications/resources/updated",
+     "params": {"uri": "note://todo", "_meta": {"io.modelcontextprotocol/subscriptionId": 7}}}
+    ```
+
+    The acknowledgment echoes the filter the server agreed to honor, and every frame carries the
+    listen request's JSON-RPC id under `_meta` — that id *is* the subscription id.
+
+## Only what was asked for
+
+The filter is a contract. A stream that requested tool-list changes and one resource URI receives those two kinds and nothing else — publish a prompt change and that stream stays silent. Resource URIs are matched as exact strings: `note://todo` does not cover `note://todo/draft`.
+
+Two more things the stream is *not*:
+
+* **It is not a replay log.** A dropped stream is gone; events published while nobody was connected are not queued. The client's contract is to re-listen and re-fetch what it cares about.
+* **It is not the 2025 path.** Clients on earlier protocol versions that called `resources/subscribe` are served by `ctx.session.send_resource_updated(uri)` — the `notify_*` methods reach `subscriptions/listen` streams only.
+
+## One process is the default. More takes a bus
+
+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 — 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.
+
+```python
+class RedisSubscriptionBus:
+    async def publish(self, event: ServerEvent) -> None:
+        await self.redis.publish("mcp-events", encode(event))  # to every replica
+
+    def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]:
+        ...  # register the local listener; a reader task calls it for arriving events
+```
+
+```python
+mcp = MCPServer("Notebook", subscriptions=RedisSubscriptionBus(...))
+```
+
+The bus carries typed `ServerEvent` values — four small dataclasses — never JSON-RPC. Stamping, filtering, and stream lifecycles stay in the SDK, so a bus implementation cannot break the protocol; it can only move events between processes. The same instance is reachable as `mcp.subscriptions`, which is also how you publish from outside a request: `await mcp.subscriptions.publish(ToolsListChanged())`.
+
+## The low-level composition
+
+Down on the low-level `Server` there is no pre-wired anything — and the same parts assemble in three lines:
+
+```python title="server.py" hl_lines="9 31 39"
+--8<-- "docs_src/subscriptions/tutorial002.py"
+```
+
+* You own the bus, so you publish to it directly: `await bus.publish(ResourceUpdated(uri=...))`. Put it wherever your handlers can reach it — module scope here, the lifespan in a bigger app.
+* `ListenHandler(bus)` is the same handler `MCPServer` registers; `on_subscriptions_listen=` is an ordinary handler slot. Don't want the SDK's semantics? Write your own handler for the slot — the spec obligations come with it.
+* `ListenHandler.close()` gracefully ends every open stream: each one receives the listen request's result as its final frame, the spec's signal that the server ended the subscription deliberately and the client shouldn't re-listen. Without it, streams end when the client disconnects.
+
+## Recap
+
+* A client opts in with one `subscriptions/listen` request; the response is the stream. There is nothing to configure server-side — serving it is built in.
+* You publish: `await ctx.notify_resource_updated(uri)`, `notify_tools_changed()`, `notify_prompts_changed()`, `notify_resources_changed()`. Idle servers make these free.
+* Streams receive only what their filter requested; URIs match exactly; nothing is replayed.
+* Scaling out means implementing `SubscriptionBus` — two methods — over your own pub/sub, and passing it as `MCPServer(subscriptions=...)`.
+* The low-level spelling is the same machinery held in your hands: a bus, `ListenHandler(bus)`, one constructor argument.

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 `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.
+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. The whole story, including scaling out across replicas, is in **[Subscriptions](../advanced/subscriptions.md)**.
 
 !!! check
     Before anyone runs `enable_recommendations`, the tool you are promising does not exist. Call it

docs_src/subscriptions/__init__.py

@@ -0,0 +1,28 @@
+from mcp.server.mcpserver import Context, MCPServer
+
+mcp = MCPServer("Notebook")
+
+NOTES = {"todo": "buy milk", "journal": "day one"}
+
+
+@mcp.resource("note://{name}")
+def note(name: str) -> str:
+    return NOTES[name]
+
+
+@mcp.tool()
+async def edit_note(name: str, text: str, ctx: Context) -> str:
+    NOTES[name] = text
+    await ctx.notify_resource_updated(f"note://{name}")
+    return "saved"
+
+
+def search(query: str) -> list[str]:
+    return [name for name, text in NOTES.items() if query in text]
+
+
+@mcp.tool()
+async def enable_search(ctx: Context) -> str:
+    mcp.add_tool(search)
+    await ctx.notify_tools_changed()
+    return "search is live"

docs_src/subscriptions/tutorial001.py

@@ -0,0 +1,40 @@
+from typing import Any
+
+import mcp_types as types
+
+from mcp.server.context import ServerRequestContext
+from mcp.server.lowlevel import Server
+from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, ResourceUpdated
+
+bus = InMemorySubscriptionBus()
+
+NOTES = {"todo": "buy milk"}
+
+EDIT_NOTE_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "properties": {"name": {"type": "string"}, "text": {"type": "string"}},
+    "required": ["name", "text"],
+}
+
+
+async def list_tools(
+    ctx: ServerRequestContext[Any], params: types.PaginatedRequestParams | None
+) -> types.ListToolsResult:
+    return types.ListToolsResult(
+        tools=[types.Tool(name="edit_note", description="Replace a note's text.", input_schema=EDIT_NOTE_SCHEMA)]
+    )
+
+
+async def call_tool(ctx: ServerRequestContext[Any], params: types.CallToolRequestParams) -> types.CallToolResult:
+    args = params.arguments or {}
+    NOTES[args["name"]] = args["text"]
+    await bus.publish(ResourceUpdated(uri=f"note://{args['name']}"))
+    return types.CallToolResult(content=[types.TextContent(type="text", text="saved")])
+
+
+server = Server(
+    "notebook",
+    on_list_tools=list_tools,
+    on_call_tool=call_tool,
+    on_subscriptions_listen=ListenHandler(bus),
+)

docs_src/subscriptions/tutorial002.py

@@ -68,6 +68,12 @@ lowlevel   = false
 transports = ["in-memory", "http-asgi"]
 era        = "dual-in-body"
 
+[story.subscriptions]
+# subscriptions/listen exists only on the 2026 wire, so there is no legacy leg.
+# The listen request parks for the stream's lifetime; the client ends it by
+# cancelling the awaiting scope (the spec's client-side close).
+era = "modern"
+
 [story.schema_validators]
 
 [story.middleware]
@@ -166,7 +172,6 @@ fixed_port    = 8000                            # issuer/PRM metadata bake in :8
 
 [deferred]
 caching       = "client honouring + per-result override unlanded"
-subscriptions = "#2901 — Client.listen / ServerEventBus"
 tasks         = "SEP-2663 — tasks extension runtime (server-decided augmentation, CreateTaskResult)"
 skills        = "#2896 — SEP-2640"
 events        = "#2901 + #2896"

examples/stories/manifest.toml

@@ -1,27 +1,58 @@
 # subscriptions
 
-The 2026-era `subscriptions/listen` channel: the server publishes change events
-through a `ServerEventBus`, and `Client.listen()` opens an async iterator over
-them. Replaces the handshake-era `resources/subscribe` + standalone-GET
-notification path.
-
-**Status: not yet implemented** ([#2901](https://github.com/modelcontextprotocol/python-sdk/issues/2901)).
-The lowlevel registration surface is in this base —
-[#2967](https://github.com/modelcontextprotocol/python-sdk/pull/2967)
-(`ae13ede`) added the lowlevel `on_subscriptions_listen` handler slot — but
-there is no `Client.listen()` or `ServerEventBus` yet. The runnable story is
-deliberately a follow-up PR to keep this one reviewable.
+Server-originated change notifications on the 2026-07-28 protocol. A client
+opens one `subscriptions/listen` request whose response **is** the stream; the
+server publishes with `ctx.notify_resource_updated(uri)` /
+`ctx.notify_tools_changed()` and the SDK does the wire work (ack-first,
+per-stream filtering, subscription-id tagging). Replaces the handshake-era
+`resources/subscribe` + standalone-GET notification path.
+
+The client edits a note it did not subscribe to (silence), edits the one it
+did (a tagged `notifications/resources/updated`), registers a tool at runtime
+(`notifications/tools/list_changed`, then re-lists and calls it), and finally
+closes the stream by cancelling the parked request.
+
+## Run it
+
+```bash
+# HTTP — the client self-hosts the server on a free port, runs, then tears it
+# down (subscriptions/listen is 2026-era only)
+uv run python -m stories.subscriptions.client --http
+# same, against the lowlevel-API server variant
+uv run python -m stories.subscriptions.client --http --server server_lowlevel
+```
+
+## What to look at
+
+- `client.py` — stream frames arrive as ordinary server notifications via the
+  constructor-only `message_handler=`. There is no client-side listen API yet,
+  so opening the stream drops to the `client.session` escape hatch; the request
+  parks for the stream's lifetime and the client closes the stream by
+  cancelling it. Every frame's `_meta["io.modelcontextprotocol/subscriptionId"]`
+  is the listen request's JSON-RPC id.
+- `server.py` — publishing is one `await ctx.notify_*()` line per change; the
+  filter, the tagging, and the ack ordering are the SDK's job. Publishing with
+  no subscribers is a no-op.
+- `server_lowlevel.py` — the same machinery held by hand: an
+  `InMemorySubscriptionBus`, handlers that `await bus.publish(...)`, and
+  `ListenHandler(bus)` passed as `on_subscriptions_listen=`. A multi-replica
+  deployment swaps the bus for one backed by its own pub/sub
+  (`MCPServer(subscriptions=...)` on the high-level server).
+
+## Caveats
+
+- 2026-era only: on a 2025 connection the method does not exist (clients there
+  use `resources/subscribe` and unsolicited notifications instead), so the
+  story pins the modern era and has no legacy leg.
+- No replay: events published while no stream is open are not queued. The
+  contract after a dropped stream is re-listen and re-fetch.
 
 ## Spec
 
 [Subscriptions — basic utilities](https://modelcontextprotocol.io/specification/draft/basic/utilities/subscriptions)
 
-## Working example elsewhere
-
-The TypeScript SDK ships a runnable `subscriptions` story:
-[typescript-sdk/examples/subscriptions](https://github.com/modelcontextprotocol/typescript-sdk/tree/main/examples/subscriptions).
-
 ## See also
 
-`standalone_get/` (handshake-era server-initiated notifications), `resources/`
-(legacy `subscribe` deliberately omitted).
+`streaming/` (request-scoped notifications), `events/` (the events extension
+on top of this channel, deferred), and `docs/advanced/subscriptions.md` (the
+narrative version).