Abonnements

Le catalogue d’un serveur n’est pas figé. Des outils apparaissent à l’exécution, et le contenu derrière l’URI d’une ressource change.

Les abonnements sont le moyen par lequel un client en est informé. Le client envoie une seule requête subscriptions/listen, et la réponse à cette requête est le flux : elle reste ouverte et transporte les notifications de changement que le client a demandées.

Publier depuis l’outil

Votre part se résume à une ligne : publier le changement.

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") atteint chaque flux ouvert abonné à cet URI. Personne d’autre.
• await ctx.notify_tools_changed() atteint chaque flux qui a demandé les changements de la liste d’outils. Un client qui la reçoit appelle de nouveau tools/list, et voit désormais sprint_report.
• Les méthodes sœurs sont notify_prompts_changed() et notify_resources_changed().
• Pas d’abonnés, pas de travail. Publier sur un serveur inactif est sans effet, vous ne vérifiez donc jamais si quelqu’un écoute. Vous indiquez ce qui a changé.

MCPServer sert subscriptions/listen pour vous. Les obligations sur la liaison (l’accusé de réception comme première trame, le filtrage par flux, l’identifiant d’abonnement sur chaque trame) sont l’affaire du SDK.

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

Seulement ce qui a été demandé

Le filtre est un contrat. Un flux qui a demandé les changements de la liste d’outils et un URI de ressource reçoit ces deux types et rien d’autre. Publiez un changement de prompt et ce flux reste silencieux.

MCPServer compare les URI de ressource comme des chaînes exactes, si bien qu’un flux qui a nommé board://sprint n’entend rien à propos de board://sprint/tasks/1. La spécification permet à un serveur de signaler un changement sur une sous-ressource d’un URI abonné ; MCPServer ne le fait jamais, mais les clients sont conçus pour s’y attendre.

Deux choses que le flux n’est pas :

• Ce n’est pas un journal de relecture. Un flux interrompu est perdu, et les événements publiés pendant que personne n’était connecté ne sont pas mis en file d’attente. Les clients rouvrent l’écoute et récupèrent de nouveau les données.
• Ce n’est pas le chemin 2025. Les clients qui ont appelé resources/subscribe sont servis par ctx.session.send_resource_updated(uri). Les méthodes notify_* n’atteignent que les flux subscriptions/listen.

Décider qui peut observer

Par défaut, chaque type et chaque URI demandés sont honorés : n’importe quel appelant peut observer n’importe quel URI que vous publiez. Rien ne consulte votre gestionnaire (handler) de lecture, car personne ne lit — un appelant que votre gestionnaire files://{name} refuserait peut tout de même ouvrir un flux sur files://payroll.csv et apprendre qu’il a changé, et quand. Il n’apprend jamais le contenu, et il ne peut pas sonder ce qui existe, car un URI inconnu est honoré lui aussi et ne se déclenche tout simplement jamais. La faille est étroite mais réelle : mettez donc un contrôle en place avant de publier des URI propres à chaque utilisateur depuis un serveur multi-locataire.

Le contrôle est un middleware. Il voit la requête subscriptions/listen avant que le SDK n’en accuse réception et refuse lorsque l’appelant demande quoi que ce soit qu’il n’a pas le droit de lire :

from mcp_types import INVALID_REQUEST, SubscriptionsListenRequestParams

from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.mcpserver import MCPServer
from mcp.shared.exceptions import MCPError

# Who may see each file. Replace this table with a database or your RBAC system.
ACCESS = {
    "files://report.pdf": {"alice", "bob"},
    "files://payroll.csv": {"carol"},
}


def can_access(user: str | None, uri: str) -> bool:
    return user is not None and user in ACCESS.get(uri, set())


async def gate_subscriptions(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult:
    if ctx.method == "subscriptions/listen":
        params = SubscriptionsListenRequestParams.model_validate(ctx.params or {}, by_name=False)
        token = get_access_token()
        user = token.subject if token else None
        if not all(can_access(user, uri) for uri in params.notifications.resource_subscriptions or ()):
            raise MCPError(INVALID_REQUEST, "not permitted to watch the requested resources")
    return await call_next(ctx)


mcp = MCPServer("Reports", middleware=[gate_subscriptions])


@mcp.resource("files://{name}")
def file(name: str) -> str:
    uri = f"files://{name}"
    token = get_access_token()
    if not can_access(token.subject if token else None, uri):
        raise MCPError(INVALID_REQUEST, f"Unknown resource: {uri}")
    return f"contents of {name}"

• ctx.params est la requête brute ; le middleware la valide donc lui-même en SubscriptionsListenRequestParams et lit le filtre demandé par le client.
• Le refus est une MCPError levée avant call_next(ctx) : le client reçoit cette erreur et aucun flux, et la connexion continue. Gardez le message uniforme, sans nommer d’URI, afin qu’un refus ne confirme jamais quels URI sont protégés.
• Une seule fonction can_access(user, uri) répond aux deux questions. Le gestionnaire de ressource la pose sur resources/read ; le middleware la pose sur subscriptions/listen. Remplacez la table par une base de données ou votre système RBAC et les deux restent synchronisés.
• La décision vaut pour toute la durée de vie du flux. Il n’y a pas de nouvelle vérification par événement ; si l’accès d’un appelant peut expirer en cours de flux (un jeton qui expire), mettez donc fin à la connexion de cet appelant à ce moment-là.

Le contrat complet du middleware, y compris ce qu’il enveloppe d’autre et pourquoi il est marqué comme provisoire, se trouve sur Middleware.

Côté client

Voici un client de l’autre côté de ce flux, qui suit le tableau :

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

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)

Entrer dans client.listen(...) envoie la requête et attend votre accusé de réception : le flux est donc actif quand le bloc commence, et chaque événement typé est un signal pour récupérer de nouveau les données, jamais une charge utile. C’est tout le contrat, en un seul écran. Tout le reste concernant le côté client a sa propre page : observer à côté d’un traitement principal, fins de flux et réouverture de l’écoute. Voir Abonnements sous Clients.

Passer à l’échelle au-delà d’un seul processus

Les publications voyagent de votre gestionnaire vers les flux ouverts via un SubscriptionBus. Le bus par défaut est en mémoire : un processus, et tous les flux qu’il contient. C’est la bonne réponse jusqu’au jour où vous exécutez des réplicas derrière un répartiteur de charge, car le flux d’un client est alors épinglé à un réplica, et une publication sur un autre réplica doit l’atteindre.

Cette jointure est à vous d’implémenter : deux méthodes au-dessus de votre backend pub/sub.

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 est à vous, tout comme la tâche de lecture sur chaque réplica qui décode les messages entrants et appelle chaque écouteur enregistré. Les écouteurs sont synchrones, ne doivent pas lever d’exception et s’exécutent sur la boucle d’événements du serveur.

Le bus transporte des valeurs ServerEvent typées, quatre petites dataclasses, jamais du JSON-RPC. Le marquage, le filtrage et le cycle de vie des flux restent dans le SDK, si bien qu’une implémentation de bus ne peut pas casser le protocole. Elle ne peut que déplacer des événements entre processus.

Pour publier en dehors d’une requête, construisez le bus vous-même afin d’en garder la référence. MCPServer en construit un en interne lorsque vous ne passez rien, et ne l’expose pas.

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

La composition bas niveau

Sur le Server bas niveau, rien n’est précâblé, et les mêmes pièces s’assemblent en trois lignes :

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,
)

• Le bus vous appartient, vous y publiez donc directement : await bus.publish(ResourceUpdated(uri=...)). Placez-le là où vos gestionnaires peuvent l’atteindre : la portée du module ici, le cycle de vie (lifespan) dans une application plus grande.
• ListenHandler(bus) est le même gestionnaire que celui qu’enregistre MCPServer, et on_subscriptions_listen= est un emplacement de gestionnaire ordinaire. Mettez votre propre callable dans cet emplacement pour une sémantique différente, et les obligations de la spécification vous reviennent : accuser réception d’abord, marquer chaque trame avec l’identifiant d’abonnement, ne rien livrer en dehors du filtre.
• ListenHandler.close() termine proprement chaque flux ouvert. Chacun reçoit le résultat de la requête listen comme dernière trame, ce qui est la manière pour la spécification de dire que le serveur a mis fin à l’abonnement délibérément. Elle rend la main avant que ces flux n’aient fini de se vider : laissez-leur donc un instant avant de démonter le transport. Sans cela, les flux se terminent lorsque le client se déconnecte.

Récapitulatif

• Un client s’inscrit avec une seule requête subscriptions/listen, et la réponse est le flux. La prise en charge est intégrée.
• Vous publiez avec ctx.notify_*, et le SDK se charge du marquage, du filtrage et du cycle de vie.
• Les événements sont des signaux, pas des charges utiles. Les deux extrémités récupèrent de nouveau les données.
• Le côté client, c’est async with client.listen(...) : tous les détails sont dans Abonnements sous Clients.
• Sur le Server bas niveau, vous assemblez vous-même les mêmes pièces : un bus, ListenHandler(bus), l’emplacement on_subscriptions_listen.
• Passer à l’échelle horizontalement signifie implémenter SubscriptionBus, deux méthodes, et le passer via MCPServer(subscriptions=...).

Exécuter le serveur qui sert tout cela, derrière un réplica ou vingt, c’est Déployer et passer à l’échelle.