Skip to content

feat: add subscription listen streams (SEP-2575) #1000

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
DaleSeo wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat: add subscription listen streams (SEP-2575) #1000

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 1 addition & 72 deletions .github/workflows/conformance.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,81 +81,10 @@ jobs:
echo "draft conformance server did not become ready" >&2
exit 1

# Run discovery separately until #985 enables the full draft suite.
- name: Run SEP-2575 discovery contract
run: |
endpoint=http://127.0.0.1:8002/mcp
common_headers=(
-H "Content-Type: application/json"
-H "Accept: application/json, text/event-stream"
-H "Mcp-Method: server/discover"
)

discover_response="$(
curl --fail-with-body --silent --show-error \
"${common_headers[@]}" \
-H "MCP-Protocol-Version: 2026-07-28" \
--data '{
"jsonrpc": "2.0",
"id": "discover",
"method": "server/discover",
"params": {
"_meta": {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientInfo": {
"name": "conformance-workflow",
"version": "1.0.0"
},
"io.modelcontextprotocol/clientCapabilities": {}
}
}
}' \
"$endpoint"
)"
jq -e '
.result.resultType == "complete" and
(.result.supportedVersions | index("2026-07-28") != null) and
(.result.capabilities | type == "object") and
(.result.serverInfo.name | type == "string") and
.result.ttlMs == 0 and
.result.cacheScope == "private"
' <<<"$discover_response"

status="$(
curl --silent --show-error \
--output /tmp/unsupported-version.json \
--write-out "%{http_code}" \
"${common_headers[@]}" \
-H "MCP-Protocol-Version: 2099-01-01" \
--data '{
"jsonrpc": "2.0",
"id": "unsupported",
"method": "server/discover",
"params": {
"_meta": {
"io.modelcontextprotocol/protocolVersion": "2099-01-01",
"io.modelcontextprotocol/clientInfo": {
"name": "conformance-workflow",
"version": "1.0.0"
},
"io.modelcontextprotocol/clientCapabilities": {}
}
}
}' \
"$endpoint"
)"
test "$status" = "400"
jq -e '
.id == "unsupported" and
.error.code == -32022 and
.error.data.requested == "2099-01-01" and
(.error.data.supported | index("2026-07-28") != null)
' /tmp/unsupported-version.json

# Keep this explicit list until the full draft suite is enabled by #985.
- name: Run supported draft server scenarios
run: |
for scenario in \
server-stateless \
sep-2164-resource-not-found \
caching \
http-header-validation \
Expand Down
101 changes: 51 additions & 50 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -882,89 +882,90 @@ context.peer.notify_resource_list_changed().await?;

## Subscriptions

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

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

### Server-side

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

```rust
use rmcp::{ErrorData as McpError, ServerHandler, model::*, service::RequestContext, RoleServer};
use std::sync::Arc;
use tokio::sync::Mutex;
use std::collections::HashSet;

#[derive(Clone)]
struct MyServer {
subscriptions: Arc<Mutex<HashSet<String>>>,
}
use rmcp::{
ErrorData, ServerHandler,
model::*,
service::SubscriptionContext,
};

impl ServerHandler for MyServer {
fn get_info(&self) -> ServerInfo {
ServerInfo::new(
ServerCapabilities::builder()
.enable_resources()
.enable_resources_subscribe()
.enable_tools()
.enable_tool_list_changed()
.build(),
)
}

async fn subscribe(
fn accepted_subscription_filter(
&self,
request: SubscribeRequestParams,
_context: RequestContext<RoleServer>,
) -> Result<(), McpError> {
self.subscriptions.lock().await.insert(request.uri);
Ok(())
requested: &SubscriptionFilter,
) -> Option<SubscriptionFilter> {
Some(requested.clone())
}

async fn unsubscribe(
&self,
request: UnsubscribeRequestParams,
_context: RequestContext<RoleServer>,
) -> Result<(), McpError> {
self.subscriptions.lock().await.remove(&request.uri);
async fn listen(&self, context: SubscriptionContext) -> Result<(), ErrorData> {
if context.accepted().tools_list_changed == Some(true) {
context.sink().notify_tool_list_changed().await
.map_err(|error| ErrorData::internal_error(error.to_string(), None))?;
}
context.cancelled().await;
Ok(())
}
}
```

When a subscribed resource changes, notify the client:

```rust
// Check if the resource has subscribers, then notify
context.peer.notify_resource_updated(
ResourceUpdatedNotificationParam::new("file:///config.json"),
).await?;
```
The SDK intersects the handler's filter with the requested categories and the
capabilities advertised by `get_info()`. It sends the acknowledgment before
`listen`, tags every sink notification with the listen request ID, and rejects
categories or resource URIs outside the accepted filter.

### Client-side

```rust
use rmcp::model::*;

// Subscribe to updates for a resource
client.subscribe(SubscribeRequestParams::new("file:///config.json")).await?;
let mut subscription = client.listen(
SubscriptionFilter::builder()
.tools_list_changed()
.resource_subscription("file:///config.json")
.build(),
).await?;

println!("accepted: {:?}", subscription.acknowledged());
while let Some(notification) = subscription.next().await? {
println!("notification: {notification:?}");
}

// Unsubscribe when no longer needed
client.unsubscribe(UnsubscribeRequestParams::new("file:///config.json")).await?;
subscription.cancel().await?;
```

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

```rust
impl ClientHandler for MyClient {
async fn on_resource_updated(
&self,
params: ResourceUpdatedNotificationParam,
_context: NotificationContext<RoleClient>,
) {
// Re-read the resource at params.uri
}
}
```
For older negotiated protocol versions, the deprecated `subscribe()` and
`unsubscribe()` APIs retain their legacy wire behavior. Modern Streamable HTTP
uses the listen POST response stream directly and does not use sessions, GET,
DELETE, or `Last-Event-ID`. After an abrupt transport close, call `listen`
again; subscription state is not resumed across HTTP or stdio reconnects.

See the
[modern subscription server](examples/servers/src/subscriptions_streamhttp.rs)
and [client](examples/clients/src/subscriptions_streamhttp.rs) examples.

---

Expand Down
Loading