Advertise subscription capabilities era-aware and apply review feedback

The big one: MCPServer served subscriptions/listen but still advertised
listChanged: false / subscribe: false everywhere, so a spec-following
client - which capability-gates its listen filter - would never
subscribe, making the publish surface unreachable. get_capabilities now
takes an optional protocol_version: at 2026-07-28+ the listChanged and
resources.subscribe bits derive from whether subscriptions/listen is
served (that is what those bits mean on a wire whose only delivery
channel is the listen stream); the handshake-era derivation is unchanged
and remains the default. server/discover passes the request's version.
The everything-server gains test_trigger_tool_change /
test_trigger_prompt_change diagnostic tools (backed by a new
MCPServer.remove_prompt mirroring remove_tool), so the conformance
list-changed SHOULD checks now run and pass: server-stateless is 30/30.

Review follow-ups:
- The bus is no longer exposed as a public MCPServer property; Context
  receives it at construction and the notify_* methods publish through
  it directly. Publishing from outside a request means keeping a
  reference to the bus you constructed.
- json_response=True no longer hangs subscriptions/listen: a listen
  response is a notification stream, so it takes the SSE path regardless
  of the JSON-response preference (TypeScript/Go parity).
- ListenHandler gains max_subscriptions (rejected pre-ack past the cap)
  and max_buffered_events (a stream whose client stopped reading is
  ended at the cap; the client re-listens - no replay means the backlog
  was already lossy).
- Honored resource URIs are matched via a frozenset instead of a list
  scan on every publish.
- InMemorySubscriptionBus isolates raising listeners (logged + skipped)
  so one bad listener cannot starve the others or fail the publisher.
- close() docs now say it initiates closure; each stream flushes from
  its own handler task.
- Story and docs corrections: cancelling the parked listen request only
  releases the local task over HTTP today (the stream ends with the
  connection); capability docs/tests updated for the era-aware bits;
  stories index row promoted; small example hardening.

Author: maxisbey

docs/advanced/subscriptions.md

@@ -57,7 +57,7 @@ class RedisSubscriptionBus:
 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 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. To publish from outside a request, keep a reference to the bus you constructed and `await bus.publish(ToolsListChanged())` — the server holds the same instance.
 
 ## The low-level composition
 

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 be told when a resource changes. On 2025-era connections that is `subscribe_resource(uri)` / `unsubscribe_resource(uri)` - a method pair `MCPServer` doesn't implement, so on the 2026-07-28 wire (where those verbs no longer exist) the request answers `-32601`, *Method not found*. The 2026 replacement is a `subscriptions/listen` stream, which `MCPServer` *does* serve - `server_capabilities.resources.subscribe` is `True` there, and the server side of the story is **[Subscriptions](../advanced/subscriptions.md)**.
 
 ## Prompts
 

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:

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 (
@@ -603,6 +603,34 @@ async def test_reconnection(ctx: Context) -> str:
     return "Reconnection test completed"
 
 
+def _dynamic_tool() -> str:
+    """A tool registered and removed by test_trigger_tool_change."""
+    return "dynamic"
+
+
+def _dynamic_prompt() -> str:
+    """A prompt registered and removed by test_trigger_prompt_change."""
+    return "dynamic"
+
+
+@mcp.tool()
+async def test_trigger_tool_change(ctx: Context) -> str:
+    """Mutates the tool list and announces it to subscriptions/listen streams (SEP-2575)"""
+    mcp.add_tool(_dynamic_tool, name="test_dynamic_tool")
+    mcp.remove_tool("test_dynamic_tool")
+    await ctx.notify_tools_changed()
+    return "tool list changed"
+
+
+@mcp.tool()
+async def test_trigger_prompt_change(ctx: Context) -> str:
+    """Mutates the prompt list and announces it to subscriptions/listen streams (SEP-2575)"""
+    mcp.add_prompt(Prompt.from_function(_dynamic_prompt, name="test_dynamic_prompt", description="dynamic"))
+    mcp.remove_prompt("test_dynamic_prompt")
+    await ctx.notify_prompts_changed()
+    return "prompt list changed"
+
+
 # Resources
 @mcp.resource("test://static-text")
 def static_text_resource() -> str:

examples/stories/README.md

@@ -148,14 +148,14 @@ opens with a banner saying what replaces it.
 | [`starlette_mount`](starlette_mount/) | mounting `streamable_http_app()` under a Starlette/FastAPI sub-path | current |
 | [`sse_polling`](sse_polling/) | SEP-1699 `closeSSE()` + `Last-Event-ID` resume via `EventStore` | legacy |
 | [`standalone_get`](standalone_get/) | server-initiated `list_changed` over the sessionful GET stream | legacy |
+| [`subscriptions`](subscriptions/) | `subscriptions/listen` streams: `ctx.notify_*`, `SubscriptionBus`, `ListenHandler` | current |
 | [`reconnect`](reconnect/) | explicit `discover()`, persist `DiscoverResult`, zero-RTT reconnect | current |
 | [`bearer_auth`](bearer_auth/) | `TokenVerifier` + `AuthSettings` bearer gate, PRM metadata, `get_access_token()` | current |
 | [`oauth`](oauth/) | full `authorization_code` grant against an in-process AS | current |
 | [`oauth_client_credentials`](oauth_client_credentials/) | `client_credentials` grant; minimal in-process token endpoint | current |
 | [`identity_assertion`](identity_assertion/) | SEP-990 enterprise IdP flow: present an ID-JAG under the `jwt-bearer` grant | current |
 | **— deferred (README only) —** | | |
 | [`caching`](caching/) | `CacheableResult` ttl/scope hints; client honouring | not yet implemented |
-| [`subscriptions`](subscriptions/) | `subscriptions/listen`, `ServerEventBus`, `Client.listen()` | not yet implemented — [#2901](https://github.com/modelcontextprotocol/python-sdk/issues/2901) |
 | [`tasks`](tasks/) | `io.modelcontextprotocol/tasks` extension | not yet implemented |
 | [`apps`](apps/) | MCP Apps: `ui://` resource + `_meta.ui` | not yet implemented — [#2896](https://github.com/modelcontextprotocol/python-sdk/issues/2896) |
 | [`skills`](skills/) | SEP-2640 skills extension | not yet implemented — [#2896](https://github.com/modelcontextprotocol/python-sdk/issues/2896) |

examples/stories/subscriptions/README.md

@@ -10,7 +10,8 @@ per-stream filtering, subscription-id tagging). Replaces the handshake-era
 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.
+stops listening - cancelling the parked request releases the local task, and
+closing the connection ends the stream server-side.
 
 ## Run it
 
@@ -27,9 +28,10 @@ uv run python -m stories.subscriptions.client --http --server server_lowlevel
 - `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.
+  parks for the stream's lifetime. Cancelling it releases the local task; over
+  HTTP the server-side stream ends when the connection closes. 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.

examples/stories/subscriptions/client.py

@@ -38,9 +38,11 @@ async def wait_for(count: int) -> None:
 
         async with anyio.create_task_group() as tg:
             # There is no client-side listen API yet, so the story drops to the
-            # `client.session` escape hatch: the request parks for the stream's
-            # lifetime, so it runs as a task and the client closes the stream by
-            # cancelling it (the spec's client-side close).
+            # `client.session` escape hatch. The request parks for the stream's
+            # lifetime, so it runs as a task; cancelling it releases the local
+            # awaiting scope. In-memory that also ends the server's stream; over
+            # HTTP today nothing aborts the POST, so the server-side stream ends
+            # when the connection closes (the `Client` exit right below).
             async def listen() -> None:
                 request = types.SubscriptionsListenRequest(
                     params=types.SubscriptionsListenRequestParams(
@@ -69,16 +71,17 @@ async def listen() -> None:
             updated = received[1]
             assert isinstance(updated, types.ResourceUpdatedNotification), updated
             assert updated.params.uri == "note://todo"
-            assert updated.params.meta == ack.params.meta
+            assert updated.params.meta is not None
+            assert updated.params.meta[SUBSCRIPTION_ID] == ack.params.meta[SUBSCRIPTION_ID]
             assert len(received) == 2, "the journal edit must not have been delivered"
 
             # ── a runtime tool registration announces itself ──
             await client.call_tool("enable_search", {})
             await wait_for(3)
             assert isinstance(received[2], types.ToolListChangedNotification), received[2]
 
-            # The client is done listening: closing the stream is cancelling the
-            # parked request's scope.
+            # The client is done listening: cancel the parked request and let
+            # the connection teardown below end the stream server-side.
             tg.cancel_scope.cancel()
 
         # list_changed told us to re-fetch - the new tool is callable, and the

examples/stories/subscriptions/server.py

@@ -22,11 +22,16 @@ async def edit_note(name: str, text: str, ctx: Context) -> str:
     def search(query: str) -> list[str]:
         return [name for name, text in notes.items() if query in text]
 
+    enabled = False
+
     @mcp.tool()
     async def enable_search(ctx: Context) -> str:
         """Register the `search` tool at runtime and tell subscribers the list changed."""
-        mcp.add_tool(search)
-        await ctx.notify_tools_changed()
+        nonlocal enabled
+        if not enabled:
+            enabled = True
+            mcp.add_tool(search)
+            await ctx.notify_tools_changed()
         return "search is live"
 
     return mcp

examples/stories/subscriptions/server_lowlevel.py

@@ -56,7 +56,7 @@ async def call_tool(ctx: ServerRequestContext[Any], params: types.CallToolReques
             search_enabled = True
             await bus.publish(ToolsListChanged())
             return types.CallToolResult(content=[types.TextContent(text="search is live")])
-        assert params.name == "search"
+        assert params.name == "search" and search_enabled
         matches = [name for name, text in notes.items() if args["query"] in text]
         return types.CallToolResult(content=[types.TextContent(text=", ".join(matches))])
 

src/mcp/server/_streamable_http_modern.py

@@ -272,7 +272,10 @@ async def handle_modern_request(
         progress_token=progress_token_from_params(req.params),
     )
 
-    if json_response:
+    if json_response and req.method != "subscriptions/listen":
+        # A listen response IS a notification stream, so it always takes the
+        # SSE path below regardless of the JSON-response preference (the
+        # TypeScript and Go SDKs route it the same way).
         msg = await _to_jsonrpc_response(
             req.id, serve_one(app, dctx, req.method, req.params, connection=connection, lifespan_state=lifespan_state)
         )