subscriptions

Typed event vocabulary for subscriptions/listen (2026-07-28, SEP-2575), shared by server and client.

Every event is a level trigger ("this changed, refetch if you care"), so both sides bound buffers by dedupe.

SUBSCRIPTION_ID_META_KEY module-attribute

SUBSCRIPTION_ID_META_KEY = (
    "io.modelcontextprotocol/subscriptionId"
)

The _meta key on every listen-stream frame; the value is the subscriptions/listen request's JSON-RPC id.

ToolsListChanged dataclass

The server's tool list changed.

Source code in src/mcp/shared/subscriptions.py

@dataclass(frozen=True)
class ToolsListChanged:
    """The server's tool list changed."""

PromptsListChanged dataclass

The server's prompt list changed.

Source code in src/mcp/shared/subscriptions.py

@dataclass(frozen=True)
class PromptsListChanged:
    """The server's prompt list changed."""

ResourcesListChanged dataclass

The server's resource list changed.

Source code in src/mcp/shared/subscriptions.py

@dataclass(frozen=True)
class ResourcesListChanged:
    """The server's resource list changed."""

ResourceUpdated dataclass

The resource at uri changed and may need to be read again.

Source code in src/mcp/shared/subscriptions.py

@dataclass(frozen=True)
class ResourceUpdated:
    """The resource at `uri` changed and may need to be read again."""

    uri: str

ServerEvent module-attribute

ServerEvent = (
    ToolsListChanged
    | PromptsListChanged
    | ResourcesListChanged
    | ResourceUpdated
)

An event a server publishes for delivery to listen subscribers.

event_to_notification

event_to_notification(
    event: ServerEvent, meta: dict[str, Any]
) -> ServerNotification

Build the stamped wire notification for event (the server's direction).

Source code in src/mcp/shared/subscriptions.py

def event_to_notification(event: ServerEvent, meta: dict[str, Any]) -> ServerNotification:
    """Build the stamped wire notification for `event` (the server's direction)."""
    if isinstance(event, ToolsListChanged):
        return ToolListChangedNotification(params=NotificationParams(_meta=meta))
    if isinstance(event, PromptsListChanged):
        return PromptListChangedNotification(params=NotificationParams(_meta=meta))
    if isinstance(event, ResourcesListChanged):
        return ResourceListChangedNotification(params=NotificationParams(_meta=meta))
    return ResourceUpdatedNotification(params=ResourceUpdatedNotificationParams(uri=event.uri, _meta=meta))

event_from_wire

event_from_wire(
    method: str, params: Mapping[str, Any] | None
) -> ServerEvent | None

The event a raw listen-stream frame announces, or None if it carries none.

Takes the raw wire dict: the client demultiplexes before the typed notification parse.

Source code in src/mcp/shared/subscriptions.py

def event_from_wire(method: str, params: Mapping[str, Any] | None) -> ServerEvent | None:
    """The event a raw listen-stream frame announces, or None if it carries none.

    Takes the raw wire dict: the client demultiplexes before the typed notification parse."""
    if (event := _LIST_CHANGED_EVENTS.get(method)) is not None:
        return event
    if method == "notifications/resources/updated":
        uri = (params or {}).get("uri")
        if isinstance(uri, str):
            return ResourceUpdated(uri=uri)
    return None

event_matches

event_matches(
    honored: SubscriptionFilter,
    uris: frozenset[str],
    event: ServerEvent,
) -> bool

Whether event is within the stream's honored filter (uris: the honored resource subscriptions as a set).

The admission predicate both sides share: server delivery and client intake honor only what was acknowledged.

Source code in src/mcp/shared/subscriptions.py

def event_matches(honored: SubscriptionFilter, uris: frozenset[str], event: ServerEvent) -> bool:
    """Whether `event` is within the stream's honored filter (`uris`: the honored resource subscriptions as a set).

    The admission predicate both sides share: server delivery and client intake honor only what was acknowledged."""
    if isinstance(event, ToolsListChanged):
        return honored.tools_list_changed is True
    if isinstance(event, PromptsListChanged):
        return honored.prompts_list_changed is True
    if isinstance(event, ResourcesListChanged):
        return honored.resources_list_changed is True
    return event.uri in uris