docs/handlers/subscriptions.md
A server's catalog is not fixed. Tools appear at runtime, and the content behind a resource URI changes.
Subscriptions are how a client hears about it. The client sends one subscriptions/listen request, and the response to that request is the stream: it stays open and carries the change notifications the client asked for.
Your side of it is one line: publish the change.
--8<-- "docs_src/subscriptions/tutorial001.py"
await ctx.notify_resource_updated("board://sprint") reaches every open stream that subscribed to that URI. Nobody else.await ctx.notify_tools_changed() reaches every stream that asked for tool-list changes. A client that receives it calls tools/list again, and now sees sprint_report.notify_prompts_changed() and notify_resources_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.
!!! check
On the wire, a stream whose filter named board://sprint looks like this after complete_task runs:
```json
{"method": "notifications/subscriptions/acknowledged",
"params": {"notifications": {"resourceSubscriptions": ["board://sprint"]}, "_meta": {"io.modelcontextprotocol/subscriptionId": "listen-1"}}}
{"method": "notifications/resources/updated",
"params": {"uri": "board://sprint", "_meta": {"io.modelcontextprotocol/subscriptionId": "listen-1"}}}
```
Note what the update does *not* carry: the board. Every frame carries the listen request's JSON-RPC id under `_meta`, and that id is the subscription id. The client mints it: the Python `Client` uses strings like `"listen-1"`; other clients may use integers.
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.
MCPServer matches resource URIs as exact strings, so a stream that named board://sprint hears nothing about board://sprint/tasks/1. The spec lets a server report a change on a sub-resource of a subscribed URI; MCPServer never does, but clients are built to expect it.
Two things the stream is not:
resources/subscribe are served by ctx.session.send_resource_updated(uri). The notify_* methods reach subscriptions/listen streams only.By default every requested kind and URI is honored: any caller may watch any URI you publish. Nothing consults your read handler, because nobody is reading — a caller your files://{name} handler would turn away can still open a stream on files://payroll.csv and learn that it changed, and when. It never learns content, and it cannot probe what exists, because an unknown URI is honored too and simply never fires. Narrow but real, so gate it before you publish per-user URIs from a multi-tenant server.
The gate is a middleware. It sees the subscriptions/listen request before the SDK acknowledges it and refuses when the caller asks for anything they may not read:
--8<-- "docs_src/subscriptions/tutorial006.py"
ctx.params is the raw request, so the middleware validates it into SubscriptionsListenRequestParams itself and reads the filter the client asked for.MCPError before call_next(ctx): the client gets that error and no stream, and the connection carries on. Keep the message uniform, naming no URI, so a refusal never confirms which URIs are protected.can_access(user, uri) answers both questions. The resource handler asks it on resources/read; the middleware asks it on subscriptions/listen. Swap the table for a database or your RBAC system and both stay in step.The full middleware contract, including what else it wraps and why it is marked provisional, is on Middleware.
Here is a client on the other side of that stream, following the board:
--8<-- "docs_src/subscriptions/tutorial003.py"
Entering client.listen(...) sends the request and waits for your acknowledgment, so the stream is live when the block starts, and each typed event is a cue to refetch, never a payload. That is the whole contract in one screen. Everything else about the client end lives on its own page: watching beside a main flow, stream endings, and re-listening. See Subscriptions under Clients.
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.
from collections.abc import Callable
from redis.asyncio import Redis
from mcp.server.mcpserver import MCPServer
from mcp.server.subscriptions import ServerEvent # SubscriptionBus is a Protocol: no base class
class RedisSubscriptionBus:
def __init__(self, redis: Redis) -> None:
self._redis = redis
self._listeners: dict[object, Callable[[ServerEvent], None]] = {}
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]:
token = object()
self._listeners[token] = listener
def unsubscribe() -> None:
self._listeners.pop(token, None)
return unsubscribe
mcp = MCPServer("Sprint Board", subscriptions=RedisSubscriptionBus(redis))
encode is yours, and so is the reader task on each replica that decodes arriving messages and calls every registered listener. Listeners are synchronous, must not raise, and run on the server's event loop.
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, construct the bus yourself so you hold the reference. MCPServer builds one internally when you pass nothing, and does not expose it.
from mcp.server.subscriptions import InMemorySubscriptionBus, ToolsListChanged
bus = InMemorySubscriptionBus()
mcp = MCPServer("Sprint Board", subscriptions=bus)
async def tools_reloaded() -> None:
await bus.publish(ToolsListChanged()) # from a lifespan task, a webhook, anywhere
Down on the low-level Server there is no pre-wired anything, and the same parts assemble in three lines:
--8<-- "docs_src/subscriptions/tutorial002.py"
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, and on_subscriptions_listen= is an ordinary handler slot. Put your own callable in that slot for different semantics, and the spec obligations move to you: acknowledge first, stamp every frame with the subscription id, deliver nothing outside the filter.ListenHandler.close() ends every open stream gracefully. Each one receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. It returns before those streams finish flushing, so give them a moment before you tear the transport down. Without it, streams end when the client disconnects.subscriptions/listen request, and the response is the stream. Serving it is built in.ctx.notify_*, and the SDK does the stamping, filtering, and lifecycle work.async with client.listen(...): Subscriptions under Clients is that story.Server you assemble the same parts yourself: a bus, ListenHandler(bus), the on_subscriptions_listen slot.SubscriptionBus, two methods, and passing it as MCPServer(subscriptions=...).Running the server that serves all this, behind one replica or twenty, is Deploy & scale.