Skip to content

Commit 5a1ead3

Browse files
committed
feat: add subscription listen streams
1 parent 2e2c791 commit 5a1ead3

28 files changed

Lines changed: 3603 additions & 221 deletions

.github/workflows/conformance.yml

Lines changed: 1 addition & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -81,81 +81,10 @@ jobs:
8181
echo "draft conformance server did not become ready" >&2
8282
exit 1
8383
84-
# Run discovery separately until #985 enables the full draft suite.
85-
- name: Run SEP-2575 discovery contract
86-
run: |
87-
endpoint=http://127.0.0.1:8002/mcp
88-
common_headers=(
89-
-H "Content-Type: application/json"
90-
-H "Accept: application/json, text/event-stream"
91-
-H "Mcp-Method: server/discover"
92-
)
93-
94-
discover_response="$(
95-
curl --fail-with-body --silent --show-error \
96-
"${common_headers[@]}" \
97-
-H "MCP-Protocol-Version: 2026-07-28" \
98-
--data '{
99-
"jsonrpc": "2.0",
100-
"id": "discover",
101-
"method": "server/discover",
102-
"params": {
103-
"_meta": {
104-
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
105-
"io.modelcontextprotocol/clientInfo": {
106-
"name": "conformance-workflow",
107-
"version": "1.0.0"
108-
},
109-
"io.modelcontextprotocol/clientCapabilities": {}
110-
}
111-
}
112-
}' \
113-
"$endpoint"
114-
)"
115-
jq -e '
116-
.result.resultType == "complete" and
117-
(.result.supportedVersions | index("2026-07-28") != null) and
118-
(.result.capabilities | type == "object") and
119-
(.result.serverInfo.name | type == "string") and
120-
.result.ttlMs == 0 and
121-
.result.cacheScope == "private"
122-
' <<<"$discover_response"
123-
124-
status="$(
125-
curl --silent --show-error \
126-
--output /tmp/unsupported-version.json \
127-
--write-out "%{http_code}" \
128-
"${common_headers[@]}" \
129-
-H "MCP-Protocol-Version: 2099-01-01" \
130-
--data '{
131-
"jsonrpc": "2.0",
132-
"id": "unsupported",
133-
"method": "server/discover",
134-
"params": {
135-
"_meta": {
136-
"io.modelcontextprotocol/protocolVersion": "2099-01-01",
137-
"io.modelcontextprotocol/clientInfo": {
138-
"name": "conformance-workflow",
139-
"version": "1.0.0"
140-
},
141-
"io.modelcontextprotocol/clientCapabilities": {}
142-
}
143-
}
144-
}' \
145-
"$endpoint"
146-
)"
147-
test "$status" = "400"
148-
jq -e '
149-
.id == "unsupported" and
150-
.error.code == -32022 and
151-
.error.data.requested == "2099-01-01" and
152-
(.error.data.supported | index("2026-07-28") != null)
153-
' /tmp/unsupported-version.json
154-
155-
# Keep this explicit list until the full draft suite is enabled by #985.
15684
- name: Run supported draft server scenarios
15785
run: |
15886
for scenario in \
87+
server-stateless \
15988
sep-2164-resource-not-found \
16089
caching \
16190
http-header-validation \

README.md

Lines changed: 51 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -882,89 +882,90 @@ context.peer.notify_resource_list_changed().await?;
882882

883883
## Subscriptions
884884

885-
Clients can subscribe to specific resources. When a subscribed resource changes, the server sends a notification and the client can re-read it.
885+
Protocol `2026-07-28` replaces `resources/subscribe`, `resources/unsubscribe`, and
886+
the standalone HTTP GET stream with the transport-neutral, long-lived
887+
`subscriptions/listen` request. Each requested notification category is opt-in.
886888

887-
**MCP Spec:** [Resources - Subscriptions](https://modelcontextprotocol.io/specification/2025-11-25/server/resources#subscriptions)
889+
**MCP Spec:** [Subscriptions](https://modelcontextprotocol.io/specification/draft/basic/patterns/subscriptions)
888890

889891
### Server-side
890892

891-
Enable subscriptions in the resources capability and implement the `subscribe()` / `unsubscribe()` handlers:
893+
Declare the notification capabilities you serve, return the accepted subset,
894+
and use the filter-enforcing subscription sink:
892895

893896
```rust
894-
use rmcp::{ErrorData as McpError, ServerHandler, model::*, service::RequestContext, RoleServer};
895-
use std::sync::Arc;
896-
use tokio::sync::Mutex;
897-
use std::collections::HashSet;
898-
899-
#[derive(Clone)]
900-
struct MyServer {
901-
subscriptions: Arc<Mutex<HashSet<String>>>,
902-
}
897+
use rmcp::{
898+
ErrorData, ServerHandler,
899+
model::*,
900+
service::SubscriptionContext,
901+
};
903902

904903
impl ServerHandler for MyServer {
905904
fn get_info(&self) -> ServerInfo {
906905
ServerInfo::new(
907906
ServerCapabilities::builder()
908-
.enable_resources()
909-
.enable_resources_subscribe()
907+
.enable_tools()
908+
.enable_tool_list_changed()
910909
.build(),
911910
)
912911
}
913912

914-
async fn subscribe(
913+
fn accepted_subscription_filter(
915914
&self,
916-
request: SubscribeRequestParams,
917-
_context: RequestContext<RoleServer>,
918-
) -> Result<(), McpError> {
919-
self.subscriptions.lock().await.insert(request.uri);
920-
Ok(())
915+
requested: &SubscriptionFilter,
916+
) -> Option<SubscriptionFilter> {
917+
Some(requested.clone())
921918
}
922919

923-
async fn unsubscribe(
924-
&self,
925-
request: UnsubscribeRequestParams,
926-
_context: RequestContext<RoleServer>,
927-
) -> Result<(), McpError> {
928-
self.subscriptions.lock().await.remove(&request.uri);
920+
async fn listen(&self, context: SubscriptionContext) -> Result<(), ErrorData> {
921+
if context.accepted().tools_list_changed == Some(true) {
922+
context.sink().notify_tool_list_changed().await
923+
.map_err(|error| ErrorData::internal_error(error.to_string(), None))?;
924+
}
925+
context.cancelled().await;
929926
Ok(())
930927
}
931928
}
932929
```
933930

934-
When a subscribed resource changes, notify the client:
935-
936-
```rust
937-
// Check if the resource has subscribers, then notify
938-
context.peer.notify_resource_updated(
939-
ResourceUpdatedNotificationParam::new("file:///config.json"),
940-
).await?;
941-
```
931+
The SDK intersects the handler's filter with the requested categories and the
932+
capabilities advertised by `get_info()`. It sends the acknowledgment before
933+
`listen`, tags every sink notification with the listen request ID, and rejects
934+
categories or resource URIs outside the accepted filter.
942935

943936
### Client-side
944937

945938
```rust
946939
use rmcp::model::*;
947940

948-
// Subscribe to updates for a resource
949-
client.subscribe(SubscribeRequestParams::new("file:///config.json")).await?;
941+
let mut subscription = client.listen(
942+
SubscriptionFilter::builder()
943+
.tools_list_changed()
944+
.resource_subscription("file:///config.json")
945+
.build(),
946+
).await?;
947+
948+
println!("accepted: {:?}", subscription.acknowledged());
949+
while let Some(notification) = subscription.next().await? {
950+
println!("notification: {notification:?}");
951+
}
950952

951-
// Unsubscribe when no longer needed
952-
client.unsubscribe(UnsubscribeRequestParams::new("file:///config.json")).await?;
953+
subscription.cancel().await?;
953954
```
954955

955-
Handle update notifications in `ClientHandler`:
956+
`listen()` buffers up to 64 notifications per subscription. Use
957+
`listen_with_capacity()` to choose a different non-zero capacity; if a consumer
958+
falls behind, `Subscription::end()` reports `SubscriptionEnd::Lagged`.
956959

957-
```rust
958-
impl ClientHandler for MyClient {
959-
async fn on_resource_updated(
960-
&self,
961-
params: ResourceUpdatedNotificationParam,
962-
_context: NotificationContext<RoleClient>,
963-
) {
964-
// Re-read the resource at params.uri
965-
}
966-
}
967-
```
960+
For older negotiated protocol versions, the deprecated `subscribe()` and
961+
`unsubscribe()` APIs retain their legacy wire behavior. Modern Streamable HTTP
962+
uses the listen POST response stream directly and does not use sessions, GET,
963+
DELETE, or `Last-Event-ID`. After an abrupt transport close, call `listen`
964+
again; subscription state is not resumed across HTTP or stdio reconnects.
965+
966+
See the
967+
[modern subscription server](examples/servers/src/subscriptions_streamhttp.rs)
968+
and [client](examples/clients/src/subscriptions_streamhttp.rs) examples.
968969

969970
---
970971

0 commit comments

Comments
 (0)