protocol_droid/application/mcp/
subscription.rs

1//! `subscriptions/listen`, independent of the transport that carries it.
2//!
3//! SEP-2575 (`2026-07-28`) replaced the HTTP `GET` stream and
4//! `resources/subscribe` with one request: the client names the notifications
5//! it wants, and the server acknowledges the subset it will honour and then
6//! tags every notification with `_meta["io.modelcontextprotocol/subscriptionId"]`,
7//! whose value is the JSON-RPC id of the `subscriptions/listen` request. On
8//! Streamable HTTP the notifications ride that request's response stream; on
9//! stdio they share the one channel and the id is how a client tells them
10//! apart. The decisions — what to honour, how to tag, what to acknowledge —
11//! are the same on both, so they live here once.
12//!
13//! - "The server MUST NOT send notification types the client has not
14//!   explicitly requested."
15//! - The acknowledgement lists "the subset of requested notification types the
16//!   server agreed to honor", omitting any it does not support.
17//! - It "MUST be the first message the server sends carrying the
18//!   subscription's ID".
19
20use std::sync::Arc;
21
22use serde_json::{Map, Value};
23
24use crate::application::server::context::PeerSession;
25use crate::application::server::{ListKind, ListSubscription, Server};
26use crate::application::session::DynFrameSink;
27use crate::domain::Feature;
28
29/// The `_meta` key every message on a subscription carries.
30pub const SUBSCRIPTION_ID_KEY: &str = "io.modelcontextprotocol/subscriptionId";
31
32/// The filter members, with the feature each needs and the kind it maps to.
33const FILTER: [(&str, Feature, ListKind); 3] = [
34    (
35        "toolsListChanged",
36        Feature::ToolsListChanged,
37        ListKind::Tools,
38    ),
39    (
40        "promptsListChanged",
41        Feature::PromptsListChanged,
42        ListKind::Prompts,
43    ),
44    (
45        "resourcesListChanged",
46        Feature::ResourcesListChanged,
47        ListKind::Resources,
48    ),
49];
50
51/// What this server agreed to send on one subscription.
52#[derive(Debug, Clone, Default)]
53pub struct Honoured {
54    /// The `SubscriptionFilter` the acknowledgement reports.
55    pub filter: Map<String, Value>,
56    /// The list notifications to subscribe to.
57    pub kinds: Vec<ListKind>,
58    /// The resource URIs to watch for `notifications/resources/updated`.
59    pub resources: Vec<String>,
60}
61
62/// Decides which of `requested` this server will honour for `peer`.
63///
64/// A member is honoured when the client asked for it and the request's own
65/// profile allows it. A resource URI is honoured only for a resource this
66/// server lists, because a watch on a URI it does not have is a wait for an
67/// update that cannot come.
68#[must_use]
69pub fn honour(requested: &Map<String, Value>, peer: &PeerSession, server: &Server) -> Honoured {
70    let mut honoured = Honoured::default();
71    for (member, feature, kind) in FILTER {
72        if requested.get(member).and_then(Value::as_bool) == Some(true) && peer.allows(feature) {
73            honoured.filter.insert(member.to_owned(), Value::Bool(true));
74            honoured.kinds.push(kind);
75        }
76    }
77    if peer.allows(Feature::ResourcesSubscribe) {
78        honoured.resources = requested
79            .get("resourceSubscriptions")
80            .and_then(Value::as_array)
81            .into_iter()
82            .flatten()
83            .filter_map(Value::as_str)
84            .filter(|uri| server.reader(uri).is_some())
85            .map(ToOwned::to_owned)
86            .collect();
87    }
88    if !honoured.resources.is_empty() {
89        honoured.filter.insert(
90            "resourceSubscriptions".to_owned(),
91            Value::Array(
92                honoured
93                    .resources
94                    .iter()
95                    .cloned()
96                    .map(Value::String)
97                    .collect(),
98            ),
99        );
100    }
101    honoured
102}
103
104/// The `_meta` object tagging a subscription opened by request `id`.
105#[must_use]
106pub fn tag(id: &Value) -> Value {
107    serde_json::json!({ SUBSCRIPTION_ID_KEY: id })
108}
109
110/// The `notifications/subscriptions/acknowledged` frame.
111#[must_use]
112pub fn acknowledgement(honoured: &Honoured, meta: &Value) -> Value {
113    serde_json::json!({
114        "jsonrpc": "2.0",
115        "method": "notifications/subscriptions/acknowledged",
116        "params": {
117            "notifications": honoured.filter,
118            "_meta": meta,
119        },
120    })
121}
122
123/// Subscribes `sink` to everything `honoured` names, tagged with `meta`.
124///
125/// Call it only after the acknowledgement has been written, so no change can
126/// reach the client ahead of it. The returned guard ends both the list and the
127/// resource subscriptions when it is dropped.
128#[must_use]
129pub fn start(
130    server: &Server,
131    sink: &Arc<dyn DynFrameSink>,
132    peer: &PeerSession,
133    honoured: &Honoured,
134    meta: &Value,
135) -> ListSubscription {
136    let handle = server.handle();
137    let subscription = handle.subscribe(sink, peer.profile(), &honoured.kinds, Some(meta));
138    for uri in &honoured.resources {
139        handle.watch_resource(sink, uri, Some(meta));
140    }
141    subscription
142}
protocol_droid/application/mcp/subscription.rs

protocol_droid/application/mcp/
subscription.rs
