Wire subscriptions/listen through MCPServer and the transports

MCPServer composes the listen runtime: subscriptions= (default on) and
event_relay= constructor parameters, the hub exposed as mcp.subscriptions,
notify_resource_updated on the server and on Context, and registry
mutations (add/remove tool, add_resource, add_prompt, the new
remove_prompt) publishing change events - only when the registry actually
mutated; duplicate registrations no longer fan out phantom notifications.

Relay pumps are owned by serving scopes, not connection lifespans: the
streamable HTTP app enters hub.run() once per app (lowlevel
streamable_http_app gained a composable lifespan parameter), sse_app and
run_stdio_async enter it once per serving context, and in-process client
connections never touch pumps. Overlapping serving contexts re-enter
without owning, so dual-transport servers compose; previously a second
concurrent SSE connection on a relay-equipped server failed at connect.

publish() is now safe from sync tool bodies (anyio worker threads hop
onto the event loop; nothing-armed publishes are no-ops from any context,
so module-scope registration stays free), declared via a direct sniffio
dependency.

JSON response mode rejects listen only when a handler is registered;
without one the request falls through to the honest METHOD_NOT_FOUND.
Building a JSON-mode app with listen wired still fails at construction,
with remedies ordered for lowlevel users first.

The everything-server fixture gains diagnostic trigger tools that mutate
the live registries and publish a watched-resource update, so the
conformance subscription checks exercise the full cross-request path.
Docs: migration guide entries (including NotificationOptions list-changed
flags being superseded by listen-derived capabilities at 2026-07-28),
low-level server page coverage for remove_request_handler and serving
subscriptions/listen.

Author: maxisbey

docs/advanced/low-level-server.md

@@ -162,6 +162,7 @@ The constructor covers the methods MCP defines. `add_request_handler` covers eve
 * The first argument is the method string. Notifications have a twin, `add_notification_handler`.
 * `params_type` is the model the incoming `params` are validated against **before** your handler runs, so custom methods *do* get the validation tools don't. Subclass `RequestParams` so the `_meta` field parses like every other method's.
 * The handler returns a `BaseModel`, a `dict`, or `None`. The SDK serialises it into the JSON-RPC result.
+* `remove_request_handler(method)` undoes a registration: the method answers `METHOD_NOT_FOUND` again and no capability derives from it anymore. It removes the built-ins too — and does **not** restore any default, so removing `server/discover` leaves the server without one. Removing a method with no handler is a no-op.
 
 One honest caveat: the high-level `Client` only has verbs for the methods MCP defines, so there is no `client.reindex()`. A vendor method is for a peer that already knows it exists: a client you also ship, or another service of yours speaking JSON-RPC.
 
@@ -177,6 +178,30 @@ The handshake belongs to the runner. `server/discover`, `ping`, and every other
 !!! tip
     `Server.middleware`, mentioned in that error, wraps **every** inbound message, including `initialize`. If what you want is to observe or rewrite traffic rather than answer a new method, start at **[Middleware](middleware.md)**.
 
+## Serving `subscriptions/listen`
+
+`MCPServer` serves the 2026-07-28 subscription stream out of the box. Down here it is two pieces from `mcp.server.subscriptions`, composed by hand:
+
+```python
+from mcp.server.subscriptions import SubscriptionHub, attach_listen
+
+hub = SubscriptionHub()
+attach_listen(server, hub)
+```
+
+`attach_listen` registers the SDK-owned `subscriptions/listen` handler — an ordinary request handler, through the same `add_request_handler` you just used. While it is registered, a 2026-07-28 `server/discover` advertises the `listChanged` flags and `resources.subscribe` for the families you serve; `detach_listen(server)` removes it again.
+
+The hub is the publish surface, and nothing publishes automatically down here. You call it when something changes:
+
+```python
+hub.notify_tools_changed()  # also: notify_prompts_changed, notify_resources_changed
+hub.notify_resource_updated("bookshop://inventory")
+```
+
+Both are synchronous and never block: every open stream whose filter matches receives the notification. Enter `async with hub.run():` around serving — a no-op for a plain hub, and the lifetime of the relay pumps when you construct `SubscriptionHub(relay=...)` with an `EventRelay`, the two-method protocol that fans opaque event bytes out across replicas over your broker (Redis pub/sub-shaped). `await hub.aclose()` ends every open stream gracefully: each parked handler flushes its pending events and returns its final result.
+
+One constraint: a listen response is a stream, so `streamable_http_app(json_response=True)` raises `ValueError` while the handler is registered. Detach it — or don't attach — to serve JSON-only.
+
 ## The other handlers
 
 Each of these is one idea you now have the vocabulary for; each has its own chapter.
@@ -192,7 +217,7 @@ Each of these is one idea you now have the vocabulary for; each has its own chap
 * An exception in a handler is a `-32603` protocol error. A tool error the model can read is a `CallToolResult` with `is_error=True` that **you** return.
 * `_meta` on the result is addressed to the client application, not the model.
 * `Server[T]` is generic in what its lifespan yields; `ctx.lifespan_context` is a typed `T`.
-* `add_request_handler(method, params_type, handler)` serves any method. `initialize` is reserved.
+* `add_request_handler(method, params_type, handler)` serves any method; `remove_request_handler(method)` stops serving one (no default comes back). `initialize` is reserved.
 * The capabilities a `Server` advertises are derived from which handlers you registered.
 
 `Client(server)` treated both servers identically because they *are* the same protocol, which is the whole point. The next layer down isn't a class at all: it's **[Middleware](middleware.md)**.

docs/client/index.md

@@ -145,7 +145,7 @@ The resource verbs come in pairs: two ways to list, one way to read.
 
 `read_resource` returns `contents`, a list of `TextResourceContents` or `BlobResourceContents`. Same idea as tool content: narrow with `isinstance`, then read `.text` (or `.blob`).
 
-A client can also **subscribe** to a resource and be told when it changes: `subscribe_resource(uri)` and `unsubscribe_resource(uri)`, same shape as everything else here. `MCPServer` doesn't implement that half. It says so up front (`server_capabilities.resources.subscribe` is `False`) and answers the request with an `MCPError`: `-32601`, *Method not found*. A server that does support subscriptions is built on the low-level `Server` (**[The low-level Server](../advanced/low-level-server.md)**).
+A client can also **subscribe** to a resource and be told when it changes. The legacy per-resource pair — `subscribe_resource(uri)` and `unsubscribe_resource(uri)` — is not served by `MCPServer`: the request is answered with an `MCPError`: `-32601`, *Method not found*. At protocol version 2026-07-28 resource subscriptions are delivered over the `subscriptions/listen` stream instead, which `MCPServer` serves out of the box (that is what `server_capabilities.resources.subscribe` advertises on a 2026 connection); a server that serves the legacy pair is built on the low-level `Server` (**[The low-level Server](../advanced/low-level-server.md)**).
 
 ## Prompts
 

docs/migration.md

@@ -431,6 +431,17 @@ For protocol 2026-07-28 over Streamable HTTP, a tool's input-schema property may
 
 On protocol 2026-07-28, servers attach caching hints (`ttlMs`, `cacheScope`) to the cacheable results, and `Client` now honors them: `list_tools`, `list_prompts`, `list_resources`, `list_resource_templates`, and `read_resource` may serve a cached response instead of making a round trip, for as long as the server's `ttlMs` says the result is fresh. With the default configuration, servers that send no hints, including every pre-2026 server, see identical call-for-call behavior, because hint-less results are not cached (a `CacheConfig.default_ttl_ms` above zero caches them too). Pass `Client(..., cache=False)` to disable the cache and restore v1 behavior exactly; per-call control (`cache_mode`) and configuration (`CacheConfig`) are described in [Caching hints](advanced/caching.md).
 
+### `MCPServer` serves `subscriptions/listen`; JSON response mode requires `subscriptions=False` ([SEP-2575](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2575))
+
+On protocol 2026-07-28, `MCPServer` serves the `subscriptions/listen` stream out of the box: registry mutations (`add_tool`, `remove_tool`, `add_resource`, `add_prompt`, and the decorators) automatically publish the matching list-changed notification to open streams, and `mcp.notify_resource_updated(uri)` / `ctx.notify_resource_updated(uri)` publish resource updates. The 2026 `server/discover` capabilities reflect this: the `listChanged` flags and `resources.subscribe` now derive from whether the listen runtime is wired, so a default `MCPServer` advertises all four (the 2025-era `initialize` capabilities are unchanged). Consequences to migrate around:
+
+- A listen response is a stream, so `streamable_http_app(json_response=True)` and `run(transport="streamable-http", json_response=True)` now raise `ValueError` for a server with the listen runtime wired. Construct with `MCPServer(..., subscriptions=False)` (or serve SSE) — the server then advertises no subscription-delivered capabilities and answers `subscriptions/listen` with `-32601`.
+- Low-level `Server` users opt in explicitly: `attach_listen(server, SubscriptionHub())` from `mcp.server.subscriptions`. Multi-replica deployments pass an `EventRelay` (`MCPServer(event_relay=...)`) to fan events out across processes; without one, delivery is per-replica.
+- The lowlevel `NotificationOptions` knobs are ignored on 2026-07-28 connections: the `notification_options` passed to `Server.create_initialization_options()` / `get_capabilities()` only affect the legacy (≤2025-11-25) capability derivation. At 2026-07-28 the `listChanged` flags and `resources.subscribe` derive solely from whether a `subscriptions/listen` handler is registered.
+- With an `EventRelay` configured, the relay pumps run once per served app (the streamable-HTTP app, `sse_app()`, or a stdio run), for exactly as long as it serves. An in-process `Client(server)` is not a serving context and does not start them. Events published before serving starts are delivered to local streams only — never queued for the relay — so restarts produce no cross-replica replay.
+
+The legacy `resources/subscribe` / `resources/unsubscribe` pair is removed from the 2026-07-28 wire; serving it for ≤2025-11-25 clients still requires registering handlers by hand — see [Registering lowlevel handlers from `MCPServer`](#registering-lowlevel-handlers-from-mcpserver).
+
 ### Server extensions API ([SEP-2133](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2133))
 
 `MCPServer` now accepts opt-in extensions that bundle MCP behaviour behind a
@@ -786,6 +797,8 @@ mcp._lowlevel_server.add_request_handler("resources/subscribe", SubscribeRequest
 
 `_lowlevel_server` is private and may change. A public way to register these handlers on `MCPServer` is planned; until then, use this workaround or use the lowlevel `Server` directly.
 
+These handlers serve ≤2025-11-25 clients only: `logging/setLevel` and the subscribe pair are removed from the 2026-07-28 wire (`METHOD_NOT_FOUND` regardless of handlers), and resource updates are delivered over the `subscriptions/listen` stream instead — see [`MCPServer` serves `subscriptions/listen`](#mcpserver-serves-subscriptionslisten-json-response-mode-requires-subscriptionsfalse-sep-2575).
+
 ### `MCPServer`'s `Context` logging: `message` renamed to `data`, `extra` removed
 
 On the high-level `Context` object (`mcp.server.mcpserver.Context`), `log()`, `.debug()`, `.info()`, `.warning()`, and `.error()` now take `data: Any` instead of `message: str`, matching the MCP spec's `LoggingMessageNotificationParams.data` field which allows any JSON-serializable value. The `extra` parameter has been removed — pass structured data directly as `data`.

docs/tutorial/first-steps.md

@@ -97,7 +97,7 @@ asyncio.run(main())
 ```
 
 ```text
-{'prompts': {'list_changed': False}, 'resources': {'subscribe': False, 'list_changed': False}, 'tools': {'list_changed': False}}
+{'prompts': {'list_changed': True}, 'resources': {'subscribe': True, 'list_changed': True}, 'tools': {'list_changed': True}}
 ```
 
 That dictionary is the server's half of the handshake:
@@ -108,7 +108,7 @@ That dictionary is the server's half of the handshake:
 | `resources` | `resources/list`, `resources/templates/list`, `resources/read` |
 | `prompts`   | `prompts/list`, `prompts/get`                               |
 
-`MCPServer` serves all three primitives, so all three are always declared.
+`MCPServer` serves all three primitives, so all three are always declared. The `list_changed` and `subscribe` flags are `True` because `MCPServer` also serves the `subscriptions/listen` stream out of the box: a client can subscribe to list-change and resource-update notifications for each family.
 
 Notice what isn't there. `completions` (argument autocomplete for resource templates and prompts) needs a handler you write, this server doesn't have one, so the capability is absent and a well-behaved client won't ask. That's the rule for everything optional: register the thing and the capability appears; **[Completions](completions.md)** proves it.
 

examples/servers/everything-server/mcp_everything_server/server.py

@@ -16,7 +16,7 @@
 import click
 from mcp.server import ServerRequestContext
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.server.mcpserver.prompts.base import UserMessage
+from mcp.server.mcpserver.prompts.base import Prompt, UserMessage
 from mcp.server.streamable_http import EventCallback, EventMessage, EventStore
 from mcp.shared.exceptions import MCPError
 from mcp_types import (
@@ -96,6 +96,7 @@ async def replay_events_after(self, last_event_id: EventId, send_callback: Event
 # Server state
 resource_subscriptions: set[str] = set()
 watched_resource_content = "Watched resource content"
+watched_resource_revision = 0
 
 # Create event store for SSE resumability (SEP-1699)
 event_store = InMemoryEventStore()
@@ -603,6 +604,45 @@ async def test_reconnection(ctx: Context) -> str:
     return "Reconnection test completed"
 
 
+# Subscriptions/listen diagnostic hooks (SEP-2575). The conformance suite mutates
+# server lists exclusively through these tools; registry mutations auto-publish
+# list-changed notifications to matching open listen streams.
+def _dummy_tool() -> str:
+    """A dummy tool added and removed by test_trigger_tool_change."""
+    return "dummy"
+
+
+@mcp.tool()
+def test_trigger_tool_change() -> str:
+    """Tests subscriptions/listen tools/list_changed delivery by mutating the tool list (SEP-2575)"""
+    mcp.add_tool(_dummy_tool, name="test_dummy_tool")
+    mcp.remove_tool("test_dummy_tool")
+    return "Tool list changed"
+
+
+def _dummy_prompt() -> list[UserMessage]:
+    """A dummy prompt added and removed by test_trigger_prompt_change."""
+    return [UserMessage(role="user", content=TextContent(type="text", text="dummy"))]
+
+
+@mcp.tool()
+def test_trigger_prompt_change() -> str:
+    """Tests subscriptions/listen prompts/list_changed delivery by mutating the prompt list (SEP-2575)"""
+    mcp.add_prompt(Prompt.from_function(_dummy_prompt, name="test_dummy_prompt"))
+    mcp.remove_prompt("test_dummy_prompt")
+    return "Prompt list changed"
+
+
+@mcp.tool()
+def test_trigger_resource_update() -> str:
+    """Tests subscriptions/listen resources/updated delivery by updating test://watched-resource (SEP-2575)"""
+    global watched_resource_content, watched_resource_revision
+    watched_resource_revision += 1
+    watched_resource_content = f"Watched resource content (revision {watched_resource_revision})"
+    mcp.notify_resource_updated("test://watched-resource")
+    return "Watched resource updated"
+
+
 # Resources
 @mcp.resource("test://static-text")
 def static_text_resource() -> str:

examples/snippets/servers/mcpserver_quickstart.py

@@ -39,4 +39,4 @@ def greet_user(name: str, style: str = "friendly") -> str:
 
 # Run with streamable HTTP transport
 if __name__ == "__main__":
-    mcp.run(transport="streamable-http", json_response=True)
+    mcp.run(transport="streamable-http")

examples/snippets/servers/oauth_server.py

@@ -19,6 +19,9 @@ async def verify_token(self, token: str) -> AccessToken | None:
 # Create MCPServer instance as a Resource Server
 mcp = MCPServer(
     "Weather Service",
+    # JSON response mode (below) cannot carry the subscriptions/listen stream,
+    # so construct without the listen runtime
+    subscriptions=False,
     # Token verifier for authentication
     token_verifier=SimpleTokenVerifier(),
     # Auth settings for RFC 9728 Protected Resource Metadata

examples/snippets/servers/streamable_config.py

@@ -4,7 +4,9 @@
 
 from mcp.server.mcpserver import MCPServer
 
-mcp = MCPServer("StatelessServer")
+# JSON response mode (below) cannot carry the subscriptions/listen stream,
+# so construct without the listen runtime.
+mcp = MCPServer("StatelessServer", subscriptions=False)
 
 
 # Add a simple tool to demonstrate the server

examples/snippets/servers/streamable_http_basic_mounting.py

@@ -11,8 +11,9 @@
 
 from mcp.server.mcpserver import MCPServer
 
-# Create MCP server
-mcp = MCPServer("My App")
+# Create MCP server. JSON response mode (below) cannot carry the
+# subscriptions/listen stream, so construct without the listen runtime.
+mcp = MCPServer("My App", subscriptions=False)
 
 
 @mcp.tool()

examples/snippets/servers/streamable_http_host_mounting.py

@@ -11,8 +11,9 @@
 
 from mcp.server.mcpserver import MCPServer
 
-# Create MCP server
-mcp = MCPServer("MCP Host App")
+# Create MCP server. JSON response mode (below) cannot carry the
+# subscriptions/listen stream, so construct without the listen runtime.
+mcp = MCPServer("MCP Host App", subscriptions=False)
 
 
 @mcp.tool()