Subscriptions

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.

Publish it from the tool

Your side of it is one line: publish the change.

server.py

from mcp.server.mcpserver import Context, MCPServer

mcp = MCPServer("Sprint Board")

BOARDS = {
    "sprint": {"design": False, "build": False, "ship": False},
    "backlog": {"tidy docs": False},
}


@mcp.resource("board://{name}")
def board(name: str) -> str:
    tasks = BOARDS[name]
    return "\n".join(f"[{'x' if done else ' '}] {task}" for task, done in tasks.items())


@mcp.tool()
async def complete_task(board: str, task: str, ctx: Context) -> str:
    BOARDS[board][task] = True
    await ctx.notify_resource_updated(f"board://{board}")
    return f"{task}: done"


def sprint_report() -> str:
    done = sum(done for tasks in BOARDS.values() for done in tasks.values())
    return f"{done} task(s) done"


@mcp.tool()
async def enable_reports(ctx: Context) -> str:
    mcp.add_tool(sprint_report)
    await ctx.notify_tools_changed()
    return "reporting is live"

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.
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.

Check

On the wire, a stream whose filter named board://sprint looks like this after complete_task runs:

{"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.

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.

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:

It is not a replay log. A dropped stream is gone, and events published while nobody was connected are not queued. Clients re-listen and refetch.
It is not the 2025 path. Clients that called resources/subscribe are served by ctx.session.send_resource_updated(uri). The notify_* methods reach subscriptions/listen streams only.

Warning

Don't publish sensitive per-user URIs through notify_resource_updated on a multi-tenant server. Any client may name any URI in its filter, and MCPServer honors it. The exposure is narrow but real: a subscriber learns that a URI it can guess changed, and when. It never learns content, and it cannot probe what exists, because an unknown URI is honored too and simply never fires. To narrow the filter per client today, serve the method with your own handler on the low-level Server and acknowledge a smaller filter than the client asked for; the acknowledgment is how the client learns what it actually got.

Same handler, every transport

A subscription is a request that stays in flight, so subscriptions/listen is served the same way over stdio (or any duplex stream) as over streamable HTTP: the acknowledgment is the first frame, events follow, and closing the stream sends the empty result. On stdio a client ends a subscription by sending notifications/cancelled for the listen request id, and the server sends nothing further for that id.

That silence is this SDK's reading, and not every SDK reads it the same way: the Go and C# listen handlers return the empty result once the client cancels, so a server built on them may write the listen request's result as a final frame after your cancel. A client written against this SDK never notices, because a result arriving for a request it already ended is a late response and is dropped. If you hand-roll a client, expect that trailing result from those servers and ignore it.

The client end

Here is a client on the other side of that stream, following the board:

client.py

from mcp_types import TextResourceContents

from mcp import Client
from mcp.client.subscriptions import ResourceUpdated, ToolsListChanged

BOARD = "board://sprint"


async def read_board(client: Client, uri: str = BOARD) -> str:
    [contents] = (await client.read_resource(uri)).contents
    assert isinstance(contents, TextResourceContents)
    return contents.text


async def follow_board(client: Client) -> None:
    async with client.listen(tools_list_changed=True, resource_subscriptions=[BOARD]) as sub:
        async for event in sub:
            match event:
                case ResourceUpdated(uri=uri):
                    print(await read_board(client, uri))
                case ToolsListChanged():
                    tools = await client.list_tools()
                    print("tools:", [tool.name for tool in tools.tools])
                case _:
                    pass  # kinds the filter did not ask for never arrive


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        await follow_board(client)

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.

Scaling past one process

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, use the bus the server owns. MCPServer builds one when you pass nothing, and exposes it as mcp.subscriptions:

from mcp.server.subscriptions import ToolsListChanged

mcp = MCPServer("Sprint Board")


async def tools_reloaded() -> None:
    await mcp.subscriptions.publish(ToolsListChanged())  # from a lifespan task, a webhook, anywhere

When you want the streams to end from your side (a clean shutdown, an event source going away), mcp.close_subscriptions() closes every open stream gracefully: each one drains what it had buffered and then receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately, and the connection carries on. Without it, streams end when their client cancels them or disconnects.

The low-level composition

Down on the low-level Server there is no pre-wired anything, and the same parts assemble in three lines:

server.py

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()
listen_handler = ListenHandler(bus)

BOARD = {"design": False, "build": False}

COMPLETE_TASK_SCHEMA: dict[str, Any] = {
    "type": "object",
    "properties": {"task": {"type": "string"}},
    "required": ["task"],
}


async def read_resource(
    ctx: ServerRequestContext[Any], params: types.ReadResourceRequestParams
) -> types.ReadResourceResult:
    board = "\n".join(f"[{'x' if done else ' '}] {task}" for task, done in BOARD.items())
    return types.ReadResourceResult(contents=[types.TextResourceContents(uri=params.uri, text=board)])


async def list_tools(
    ctx: ServerRequestContext[Any], params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
    return types.ListToolsResult(
        tools=[types.Tool(name="complete_task", description="Mark a task done.", input_schema=COMPLETE_TASK_SCHEMA)]
    )


async def call_tool(ctx: ServerRequestContext[Any], params: types.CallToolRequestParams) -> types.CallToolResult:
    args = params.arguments or {}
    BOARD[args["task"]] = True
    await bus.publish(ResourceUpdated(uri="board://sprint"))
    return types.CallToolResult(content=[types.TextContent(type="text", text="done")])


server = Server(
    "sprint-board",
    on_read_resource=read_resource,
    on_list_tools=list_tools,
    on_call_tool=call_tool,
    on_subscriptions_listen=listen_handler,
)

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, 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, and Server.close_subscriptions() is the same verb on the server that registered the handler. Each stream drains its buffered events and then receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. The call returns before those streams finish flushing, so give them a moment before you tear the transport down yourself. Over stdio you rarely need to: when the client's input ends, the driver closes your open streams for you inside a short bounded window, in which each stream's final result is guaranteed to reach the departing peer and any events it still had buffered flush as time allows. Without any close, streams end when the client disconnects.

Recap

A client opts in with one subscriptions/listen request, and the response is the stream. Serving it is built in.
You publish with ctx.notify_*, and the SDK does the stamping, filtering, and lifecycle work.
Events are cues, not payloads. Both ends refetch.
The client end is async with client.listen(...): Subscriptions 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=...).

Running the server that serves all this, behind one replica or twenty, is Deploy & scale.