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

"""

from __future__ import annotations

from collections.abc import Mapping

from dataclasses import dataclass

from typing import Any

from mcp_types import (

NotificationParams,

PromptListChangedNotification,

ResourceListChangedNotification,

ResourceUpdatedNotification,

ResourceUpdatedNotificationParams,

ServerNotification,

SubscriptionFilter,

ToolListChangedNotification,

)

__all__ = [

"LISTEN_STREAM_METHODS",

"SUBSCRIPTION_ID_META_KEY",

"PromptsListChanged",

"ResourceUpdated",

"ResourcesListChanged",

"ServerEvent",

"ToolsListChanged",

"event_from_wire",

"event_matches",

"event_to_notification",

]

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

@dataclass(frozen=True)

class ToolsListChanged:

"""The server's tool list changed."""

@dataclass(frozen=True)

class PromptsListChanged:

"""The server's prompt list changed."""

@dataclass(frozen=True)

class ResourcesListChanged:

"""The server's resource list changed."""

@dataclass(frozen=True)

class ResourceUpdated:

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

uri: str

ServerEvent = ToolsListChanged | PromptsListChanged | ResourcesListChanged | ResourceUpdated

"""An event a server publishes for delivery to listen subscribers."""

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

_LIST_CHANGED_EVENTS: dict[str, ServerEvent] = {

"notifications/tools/list_changed": ToolsListChanged(),

"notifications/prompts/list_changed": PromptsListChanged(),

"notifications/resources/list_changed": ResourcesListChanged(),

}

LISTEN_STREAM_METHODS: frozenset[str] = frozenset({*_LIST_CHANGED_EVENTS, "notifications/resources/updated"})

"""The notification methods that ride `subscriptions/listen` streams at 2026-07-28

(and, at that era, nowhere else): the change-notification vocabulary."""

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

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