Підписки

Каталог сервера не є незмінним. Інструменти з'являються під час роботи, а вміст за URI ресурсу змінюється.

Підписки — це спосіб, у який клієнт про це дізнається. Клієнт надсилає один запит subscriptions/listen, і відповідь на цей запит і є потоком: він залишається відкритим і несе сповіщення про зміни, про які попросив клієнт.

Публікація з інструмента

Ваша частина роботи — один рядок: опублікувати зміну.

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") доходить до кожного відкритого потоку, підписаного на цей URI. І більше ні до кого.
• await ctx.notify_tools_changed() доходить до кожного потоку, який попросив про зміни списку інструментів. Клієнт, що його отримав, знову викликає tools/list і тепер бачить sprint_report.
• Споріднені методи — notify_prompts_changed() і notify_resources_changed().
• Немає підписників — немає роботи. Публікація на сервері без слухачів нічого не робить, тож перевіряти, чи хтось слухає, ніколи не потрібно. Ви лише повідомляєте, що змінилося.

MCPServer обслуговує subscriptions/listen за вас. Зобов'язання на рівні протоколу (підтвердження першим кадром, фільтрація для кожного потоку, ідентифікатор підписки в кожному кадрі) — справа 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"}}}

Лише те, про що попросили

Фільтр — це контракт. Потік, що запросив зміни списку інструментів і один URI ресурсу, отримує ці два різновиди й нічого більше. Опублікуйте зміну промпту — і цей потік мовчатиме.

MCPServer зіставляє URI ресурсів як точні рядки, тож потік, у якому вказано board://sprint, нічого не почує про board://sprint/tasks/1. Специфікація дозволяє серверу повідомляти про зміну підресурсу URI, на який оформлено підписку; MCPServer цього ніколи не робить, але клієнти побудовані так, щоб на це очікувати.

Дві речі, якими потік не є:

• Це не журнал для повторного відтворення. Обірваний потік зникає, а події, опубліковані, поки ніхто не був під'єднаний, у чергу не ставляться. Клієнти знову починають слухати й заново отримують дані.
• Це не шлях покоління 2025. Клієнтів, що викликали resources/subscribe, обслуговує ctx.session.send_resource_updated(uri). Методи notify_* доходять лише до потоків subscriptions/listen.

Хто має право спостерігати

За замовчуванням задовольняється кожен запитаний різновид і URI: будь-хто може спостерігати за будь-яким URI, який ви публікуєте. Ваш обробник читання ніхто не викликає, бо ніхто нічого не читає — той, кого ваш обробник files://{name} відхилив би, усе одно може відкрити потік на files://payroll.csv і дізнатися, що файл змінився і коли. Вмісту він ніколи не дізнається й не може з'ясувати, що існує, бо невідомий URI теж задовольняється й просто ніколи не спрацьовує. Витік вузький, але реальний, тож поставте перевірку доступу, перш ніж публікувати URI окремих користувачів із сервера з кількома орендарями.

Ця перевірка — middleware (проміжний шар). Воно бачить запит subscriptions/listen до того, як SDK його підтвердить, і відмовляє, коли хтось просить те, чого не має права читати:

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 — це сирий запит, тому middleware саме перевіряє його в SubscriptionsListenRequestParams і читає фільтр, який запросив клієнт.
• Відмова — це викинутий MCPError перед call_next(ctx): клієнт отримує цю помилку й жодного потоку, а з'єднання живе далі. Тримайте повідомлення однаковим, без згадки URI, щоб відмова ніколи не підтверджувала, які саме URI захищені.
• Одна функція can_access(user, uri) відповідає на обидва питання. Обробник ресурсу викликає її на resources/read; middleware — на subscriptions/listen. Замініть таблицю базою даних або своєю системою RBAC, і обидва шляхи залишаться узгодженими.
• Рішення діє протягом усього життя потоку. Повторної перевірки для кожної події немає, тож якщо доступ може минути посеред потоку (токен, строк дії якого спливає), розірвіть з'єднання цього клієнта, коли це станеться.

Повний контракт middleware, зокрема що ще воно обгортає і чому позначене як попереднє, описано на сторінці Middleware.

Клієнтський бік

Ось клієнт на іншому кінці цього потоку, що стежить за дошкою:

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)

Вхід у client.listen(...) надсилає запит і чекає на ваше підтвердження, тож на початку блоку потік уже працює, а кожна типізована подія — це сигнал отримати дані заново, а не корисне навантаження. Оце й увесь контракт на одному екрані. Усе інше про клієнтський бік живе на окремій сторінці: спостереження поруч з основним потоком виконання, завершення потоків і повторне прослуховування. Див. Підписки у розділі Клієнти.

Масштабування за межі одного процесу

Публікації йдуть від обробника до відкритих потоків через SubscriptionBus. Типова шина живе в пам'яті: один процес і всі потоки в ньому. Це правильна відповідь, доки ви не запускаєте репліки за балансувальником навантаження, бо тоді потік клієнта прив'язаний до однієї репліки, а публікація на іншій репліці має до нього дійти.

Цей стик реалізуєте ви: два методи поверх вашого бекенда 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 — ваша функція, як і задача-читач на кожній репліці, що декодує вхідні повідомлення й викликає кожного зареєстрованого слухача. Слухачі синхронні, не повинні викидати винятків і виконуються в циклі подій сервера.

Шина несе типізовані значення ServerEvent — чотири невеликі dataclass-класи, ніколи не JSON-RPC. Проставлення ідентифікаторів, фільтрація та життєві цикли потоків залишаються в SDK, тож реалізація шини не може зламати протокол. Вона може лише переносити події між процесами.

Щоб публікувати поза запитом, створіть шину самі, аби мати на неї посилання. MCPServer будує її всередині, коли ви нічого не передаєте, і не надає до неї доступу.

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

Низькорівнева композиція

На низькорівневому Server нічого заздалегідь не під'єднано, і ті самі частини збираються в три рядки:

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

• Шина належить вам, тож ви публікуєте в неї напряму: await bus.publish(ResourceUpdated(uri=...)). Розмістіть її там, де до неї дістануться обробники: тут — на рівні модуля, у більшому застосунку — у життєвому циклі (lifespan).
• ListenHandler(bus) — той самий обробник, що його реєструє MCPServer, а on_subscriptions_listen= — звичайний слот обробника. Поставте в цей слот власний викликний об'єкт для іншої семантики, і зобов'язання специфікації переходять до вас: спочатку підтвердити, проставити в кожному кадрі ідентифікатор підписки, не доставляти нічого поза фільтром.
• ListenHandler.close() коректно завершує кожен відкритий потік. Кожен отримує результат запиту listen останнім кадром — так специфікація каже, що сервер свідомо завершив підписку. Метод повертається раніше, ніж ці потоки встигнуть усе надіслати, тож дайте їм мить, перш ніж згортати транспорт. Без нього потоки завершуються, коли клієнт від'єднується.

Підсумки

• Клієнт погоджується одним запитом subscriptions/listen, і відповідь — це потік. Його обслуговування вбудоване.
• Ви публікуєте через ctx.notify_*, а SDK проставляє ідентифікатори, фільтрує й керує життєвим циклом.
• Події — це сигнали, а не корисне навантаження. Обидва кінці отримують дані заново.
• Клієнтський бік — це async with client.listen(...): докладніше — на сторінці Підписки у розділі Клієнти.
• На низькорівневому Server ті самі частини ви збираєте самі: шина, ListenHandler(bus), слот on_subscriptions_listen.
• Горизонтальне масштабування — це реалізація SubscriptionBus (два методи) і передавання її як MCPServer(subscriptions=...).

Як запустити сервер, що все це обслуговує, за однією реплікою чи за двадцятьма, — на сторінці Розгортання й масштабування.