Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

Merged
alexhancock merged 2 commits into from
Jan 9, 2026
Merged

Implement SEP-1699: Support SSE Polling via Server-Side Disconnect #604

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f2cb7f5
feat: implement SEP-1699 SSE polling via server-side disconnect
DaleSeo Dec 28, 2025
dbd5ae0
test: add tests for priming behavior on stream start and close
DaleSeo Dec 28, 2025
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
8 changes: 7 additions & 1 deletion crates/rmcp/Cargo.toml
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
"dep:http-body-util",
"dep:bytes",
"dep:sse-stream",
"dep:axum",
"tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"
24 changes: 21 additions & 3 deletions crates/rmcp/src/transport/common/server_side_http.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
/// The event ID for this message. When set, clients can use this ID
/// with the `Last-Event-ID` header to resume the stream from this point.
pub event_id: Option<String>,
pub message: Arc<ServerJsonRpcMessage>,
/// The JSON-RPC message content. Set to `None` for priming events.
/// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
pub message: Option<Arc<ServerJsonRpcMessage>>,
/// The retry interval hint for clients. Clients should wait this duration
/// before attempting to reconnect. This maps to the SSE `retry:` field.
pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
use futures::StreamExt;
let stream = stream
.map(|message| {
let data = serde_json::to_string(&message.message).expect("valid message");
let mut sse = Sse::default().data(data);
let mut sse = if let Some(ref msg) = message.message {
let data = serde_json::to_string(msg.as_ref()).expect("valid message");
Sse::default().data(data)
} else {
// Priming event: empty data per SEP-1699 (just "data:\n")
Sse::default().data("")
};

sse.id = message.event_id;

if let Some(retry) = message.retry {
sse.retry = Some(retry.as_millis() as u64);
}

Result::<Sse, Infallible>::Ok(sse)
})
.take_until(async move { ct.cancelled().await });
Expand Down
141 changes: 138 additions & 3 deletions crates/rmcp/src/transport/streamable_http_server/session/local.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
Self::new(tx, None)
}

async fn send(&mut self, message: ServerJsonRpcMessage) {
fn next_event_id(&self) -> EventId {
let index = self.cache.back().map_or(0, |m| {
m.event_id
.as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
.index
+ 1
});
let event_id = EventId {
EventId {
http_request_id: self.http_request_id,
index,
}
}

async fn send(&mut self, message: ServerJsonRpcMessage) {
let event_id = self.next_event_id();
let message = ServerSseMessage {
event_id: Some(event_id.to_string()),
message: Some(Arc::new(message)),
retry: None,
};
self.cache_and_send(message).await;
}

async fn send_priming(&mut self, retry: Duration) {
let event_id = self.next_event_id();
let message = ServerSseMessage {
event_id: Some(event_id.to_string()),
message: Arc::new(message),
message: None,
retry: Some(retry),
};
self.cache_and_send(message).await;
}

async fn cache_and_send(&mut self, message: ServerSseMessage) {
if self.cache.len() >= self.capacity {
self.cache.pop_front();
self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
}
}
}

async fn close_sse_stream(
&mut self,
http_request_id: Option<HttpRequestId>,
retry_interval: Option<Duration>,
) -> Result<(), SessionError> {
match http_request_id {
// Close a request-wise stream
Some(id) => {
let request_wise = self
.tx_router
.get_mut(&id)
.ok_or(SessionError::ChannelClosed(Some(id)))?;

// Send priming event if retry interval is specified
if let Some(interval) = retry_interval {
request_wise.tx.send_priming(interval).await;
}

// Close the stream by dropping the sender
let (tx, _rx) = tokio::sync::mpsc::channel(1);
request_wise.tx.tx = tx;

tracing::debug!(
http_request_id = id,
"closed SSE stream for server-initiated disconnection"
);
Ok(())
}
// Close the standalone (common) stream
None => {
// Send priming event if retry interval is specified
if let Some(interval) = retry_interval {
self.common.send_priming(interval).await;
}

// Close the stream by dropping the sender
let (tx, _rx) = tokio::sync::mpsc::channel(1);
self.common.tx = tx;

tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
Ok(())
}
}
}
}

#[derive(Debug)]
pub enum SessionEvent {
ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
},
Close,
CloseSseStream {
/// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
http_request_id: Option<HttpRequestId>,
/// Optional retry interval. If provided, a priming event is sent before closing.
retry_interval: Option<Duration>,
responder: oneshot::Sender<Result<(), SessionError>>,
},
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
rx.await
.map_err(|_| SessionError::SessionServiceTerminated)?
}

/// Close an SSE stream for a specific request.
///
/// This closes the SSE connection for a POST request stream, but keeps the session
/// and message cache active. Clients can reconnect using the `Last-Event-ID` header
/// via a GET request to resume receiving messages.
///
/// # Arguments
///
/// * `http_request_id` - The HTTP request ID of the stream to close
/// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
pub async fn close_sse_stream(
&self,
http_request_id: HttpRequestId,
retry_interval: Option<Duration>,
) -> Result<(), SessionError> {
let (tx, rx) = tokio::sync::oneshot::channel();
self.event_tx
.send(SessionEvent::CloseSseStream {
http_request_id: Some(http_request_id),
retry_interval,
responder: tx,
})
.await
.map_err(|_| SessionError::SessionServiceTerminated)?;
rx.await
.map_err(|_| SessionError::SessionServiceTerminated)?
}

/// Close the standalone SSE stream.
///
/// This closes the standalone SSE connection (established via GET request),
/// but keeps the session and message cache active. Clients can reconnect using
/// the `Last-Event-ID` header via a GET request to resume receiving messages.
///
/// # Arguments
///
/// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
pub async fn close_standalone_sse_stream(
&self,
retry_interval: Option<Duration>,
) -> Result<(), SessionError> {
let (tx, rx) = tokio::sync::oneshot::channel();
self.event_tx
.send(SessionEvent::CloseSseStream {
http_request_id: None,
retry_interval,
responder: tx,
})
.await
.map_err(|_| SessionError::SessionServiceTerminated)?;
rx.await
.map_err(|_| SessionError::SessionServiceTerminated)?
}
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
InnerEvent::FromHttpService(SessionEvent::Close) => {
return Err(WorkerQuitReason::TransportClosed);
}
InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
http_request_id,
retry_interval,
responder,
}) => {
let handle_result =
self.close_sse_stream(http_request_id, retry_interval).await;
let _ = responder.send(handle_result);
}
_ => {
// ignore
}
Expand Down
77 changes: 62 additions & 15 deletions crates/rmcp/src/transport/streamable_http_server/tower.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
/// The ping message duration for SSE connections.
pub sse_keep_alive: Option<Duration>,
/// The retry interval for SSE priming events.
pub sse_retry: Option<Duration>,
/// If true, the server will create a session for each request and keep it alive.
/// When enabled, SSE priming events are sent to enable client reconnection.
pub stateful_mode: bool,
/// Cancellation token for the Streamable HTTP server.
///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
fn default() -> Self {
Self {
sse_keep_alive: Some(Duration::from_secs(15)),
sse_retry: Some(Duration::from_secs(3)),
stateful_mode: true,
cancellation_token: CancellationToken::new(),
}
Expand DownExpand Up@@ -216,6 +220,7 @@ where
.resume(&session_id, last_event_id)
.await
.map_err(internal_error_response("resume session"))?;
// Resume doesn't need priming - client already has the event ID
Ok(sse_stream_response(
stream,
self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
.create_standalone_stream(&session_id)
.await
.map_err(internal_error_response("create standalone stream"))?;
// Prepend priming event if sse_retry configured
let stream = if let Some(retry) = self.config.sse_retry {
let priming = ServerSseMessage {
event_id: Some("0".into()),
message: None,
retry: Some(retry),
};
futures::stream::once(async move { priming })
.chain(stream)
.left_stream()
} else {
stream.right_stream()
};
Ok(sse_stream_response(
stream,
self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
.create_stream(&session_id, message)
.await
.map_err(internal_error_response("get session"))?;
// Prepend priming event if sse_retry configured
let stream = if let Some(retry) = self.config.sse_retry {
let priming = ServerSseMessage {
event_id: Some("0".into()),
message: None,
retry: Some(retry),
};
futures::stream::once(async move { priming })
.chain(stream)
.left_stream()
} else {
stream.right_stream()
};
Ok(sse_stream_response(
stream,
self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
.initialize_session(&session_id, message)
.await
.map_err(internal_error_response("create stream"))?;
let stream = futures::stream::once(async move {
ServerSseMessage {
event_id: None,
message: Some(Arc::new(response)),
retry: None,
}
});
// Prepend priming event if sse_retry configured
let stream = if let Some(retry) = self.config.sse_retry {
let priming = ServerSseMessage {
event_id: Some("0".into()),
message: None,
retry: Some(retry),
};
futures::stream::once(async move { priming })
.chain(stream)
.left_stream()
} else {
stream.right_stream()
};
let mut response = sse_stream_response(
futures::stream::once({
async move {
ServerSseMessage {
event_id: None,
message: response.into(),
}
}
}),
stream,
self.config.sse_keep_alive,
self.config.cancellation_token.child_token(),
);
Expand All@@ -424,14 +468,17 @@ where
// on service created
let _ = service.waiting().await;
});
// Stateless mode: no priming (no session to resume)
let stream = ReceiverStream::new(receiver).map(|message| {
tracing::info!(?message);
ServerSseMessage {
event_id: None,
message: Some(Arc::new(message)),
retry: None,
}
});
Ok(sse_stream_response(
ReceiverStream::new(receiver).map(|message| {
tracing::info!(?message);
ServerSseMessage {
event_id: None,
message: message.into(),
}
}),
stream,
self.config.sse_keep_alive,
self.config.cancellation_token.child_token(),
))
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all 
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

 Merged

alexhancock
merged 2 commits into

from

Jan 9, 2026
 Merged


Implement SEP-1699: Support SSE Polling via Server-Side Disconnect
#604
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


2 commits


Select commit
Hold shift + click to select a range
f2cb7f5

feat: implement SEP-1699 SSE polling via server-side disconnect


DaleSeo Dec 28, 2025
dbd5ae0

test: add tests for priming behavior on stream start and close


DaleSeo Dec 28, 2025

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


8 changes: 7 additions & 1 deletion
crates/rmcp/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
 "dep:http-body-util",
 "dep:bytes",
 "dep:sse-stream",
 "dep:axum",
 "tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"

24 changes: 21 additions & 3 deletions
crates/rmcp/src/transport/common/server_side_http.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
 /// The event ID for this message. When set, clients can use this ID
 /// with the `Last-Event-ID` header to resume the stream from this point.
 pub event_id: Option<String>,
 pub message: Arc<ServerJsonRpcMessage>,
 /// The JSON-RPC message content. Set to `None` for priming events.
 /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
 pub message: Option<Arc<ServerJsonRpcMessage>>,
 /// The retry interval hint for clients. Clients should wait this duration
 /// before attempting to reconnect. This maps to the SSE `retry:` field.
 pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
 use futures::StreamExt;
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 let mut sse = if let Some(ref msg) = message.message {
 let data = serde_json::to_string(msg.as_ref()).expect("valid message");
 Sse::default().data(data)
 } else {
 // Priming event: empty data per SEP-1699 (just "data:\n")
 Sse::default().data("")
 };

 sse.id = message.event_id;

 if let Some(retry) = message.retry {
 sse.retry = Some(retry.as_millis() as u64);
 }

 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
Expand Down

141 changes: 138 additions & 3 deletions
crates/rmcp/src/transport/streamable_http_server/session/local.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
 Self::new(tx, None)
 }

async fn send(&mut self, message: ServerJsonRpcMessage) {
 fn next_event_id(&self) -> EventId {
 let index = self.cache.back().map_or(0, |m| {
 m.event_id
 .as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
 .index
 + 1
 });
let event_id = EventId {
 EventId {
 http_request_id: self.http_request_id,
 index,
 }
 }

 async fn send(&mut self, message: ServerJsonRpcMessage) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Some(Arc::new(message)),
 retry: None,
 };
 self.cache_and_send(message).await;
 }

 async fn send_priming(&mut self, retry: Duration) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Arc::new(message),
 message: None,
 retry: Some(retry),
 };
 self.cache_and_send(message).await;
 }

 async fn cache_and_send(&mut self, message: ServerSseMessage) {
 if self.cache.len() >= self.capacity {
 self.cache.pop_front();
 self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
 }
 }
 }

 async fn close_sse_stream(
 &mut self,
 http_request_id: Option<HttpRequestId>,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 match http_request_id {
 // Close a request-wise stream
 Some(id) => {
 let request_wise = self
 .tx_router
 .get_mut(&id)
 .ok_or(SessionError::ChannelClosed(Some(id)))?;

 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 request_wise.tx.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 request_wise.tx.tx = tx;

 tracing::debug!(
 http_request_id = id,
 "closed SSE stream for server-initiated disconnection"
 );
 Ok(())
 }
 // Close the standalone (common) stream
 None => {
 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 self.common.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 self.common.tx = tx;

 tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
 Ok(())
 }
 }
 }
}

#[derive(Debug)]
pub enum SessionEvent {
 ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
 responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
 },
 Close,
 CloseSseStream {
 /// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
 http_request_id: Option<HttpRequestId>,
 /// Optional retry interval. If provided, a priming event is sent before closing.
 retry_interval: Option<Duration>,
 responder: oneshot::Sender<Result<(), SessionError>>,
 },
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close an SSE stream for a specific request.
 ///
 /// This closes the SSE connection for a POST request stream, but keeps the session
 /// and message cache active. Clients can reconnect using the `Last-Event-ID` header
 /// via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `http_request_id` - The HTTP request ID of the stream to close
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_sse_stream(
 &self,
 http_request_id: HttpRequestId,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: Some(http_request_id),
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close the standalone SSE stream.
 ///
 /// This closes the standalone SSE connection (established via GET request),
 /// but keeps the session and message cache active. Clients can reconnect using
 /// the `Last-Event-ID` header via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_standalone_sse_stream(
 &self,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: None,
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
 InnerEvent::FromHttpService(SessionEvent::Close) => {
 return Err(WorkerQuitReason::TransportClosed);
 }
 InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
 http_request_id,
 retry_interval,
 responder,
 }) => {
 let handle_result =
 self.close_sse_stream(http_request_id, retry_interval).await;
 let _ = responder.send(handle_result);
 }
 _ => {
 // ignore
 }
Expand Down

77 changes: 62 additions & 15 deletions
crates/rmcp/src/transport/streamable_http_server/tower.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
 /// The ping message duration for SSE connections.
 pub sse_keep_alive: Option<Duration>,
 /// The retry interval for SSE priming events.
 pub sse_retry: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 /// When enabled, SSE priming events are sent to enable client reconnection.
 pub stateful_mode: bool,
 /// Cancellation token for the Streamable HTTP server.
 ///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 sse_retry: Some(Duration::from_secs(3)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
Expand DownExpand Up@@ -216,6 +220,7 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 // Resume doesn't need priming - client already has the event ID
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
 .initialize_session(&session_id, message)
 .await
 .map_err(internal_error_response("create stream"))?;
 let stream = futures::stream::once(async move {
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(response)),
 retry: None,
 }
 });
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 let mut response = sse_stream_response(
 futures::stream::once({
 async move {
 ServerSseMessage {
 event_id: None,
 message: response.into(),
 }
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );
Expand All@@ -424,14 +468,17 @@ where
 // on service created
 let _ = service.waiting().await;
 });
 // Stateless mode: no priming (no session to resume)
 let stream = ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(message)),
 retry: None,
 }
 });
 Ok(sse_stream_response(
 ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: message.into(),
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix");
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

 Merged

alexhancock
merged 2 commits into

from

Jan 9, 2026
 Merged


Implement SEP-1699: Support SSE Polling via Server-Side Disconnect
#604
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


2 commits


Select commit
Hold shift + click to select a range
f2cb7f5

feat: implement SEP-1699 SSE polling via server-side disconnect


DaleSeo Dec 28, 2025
dbd5ae0

test: add tests for priming behavior on stream start and close


DaleSeo Dec 28, 2025

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


8 changes: 7 additions & 1 deletion
crates/rmcp/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
 "dep:http-body-util",
 "dep:bytes",
 "dep:sse-stream",
 "dep:axum",
 "tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"

24 changes: 21 additions & 3 deletions
crates/rmcp/src/transport/common/server_side_http.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
 /// The event ID for this message. When set, clients can use this ID
 /// with the `Last-Event-ID` header to resume the stream from this point.
 pub event_id: Option<String>,
 pub message: Arc<ServerJsonRpcMessage>,
 /// The JSON-RPC message content. Set to `None` for priming events.
 /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
 pub message: Option<Arc<ServerJsonRpcMessage>>,
 /// The retry interval hint for clients. Clients should wait this duration
 /// before attempting to reconnect. This maps to the SSE `retry:` field.
 pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
 use futures::StreamExt;
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 let mut sse = if let Some(ref msg) = message.message {
 let data = serde_json::to_string(msg.as_ref()).expect("valid message");
 Sse::default().data(data)
 } else {
 // Priming event: empty data per SEP-1699 (just "data:\n")
 Sse::default().data("")
 };

 sse.id = message.event_id;

 if let Some(retry) = message.retry {
 sse.retry = Some(retry.as_millis() as u64);
 }

 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
Expand Down

141 changes: 138 additions & 3 deletions
crates/rmcp/src/transport/streamable_http_server/session/local.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
 Self::new(tx, None)
 }

async fn send(&mut self, message: ServerJsonRpcMessage) {
 fn next_event_id(&self) -> EventId {
 let index = self.cache.back().map_or(0, |m| {
 m.event_id
 .as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
 .index
 + 1
 });
let event_id = EventId {
 EventId {
 http_request_id: self.http_request_id,
 index,
 }
 }

 async fn send(&mut self, message: ServerJsonRpcMessage) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Some(Arc::new(message)),
 retry: None,
 };
 self.cache_and_send(message).await;
 }

 async fn send_priming(&mut self, retry: Duration) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Arc::new(message),
 message: None,
 retry: Some(retry),
 };
 self.cache_and_send(message).await;
 }

 async fn cache_and_send(&mut self, message: ServerSseMessage) {
 if self.cache.len() >= self.capacity {
 self.cache.pop_front();
 self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
 }
 }
 }

 async fn close_sse_stream(
 &mut self,
 http_request_id: Option<HttpRequestId>,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 match http_request_id {
 // Close a request-wise stream
 Some(id) => {
 let request_wise = self
 .tx_router
 .get_mut(&id)
 .ok_or(SessionError::ChannelClosed(Some(id)))?;

 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 request_wise.tx.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 request_wise.tx.tx = tx;

 tracing::debug!(
 http_request_id = id,
 "closed SSE stream for server-initiated disconnection"
 );
 Ok(())
 }
 // Close the standalone (common) stream
 None => {
 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 self.common.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 self.common.tx = tx;

 tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
 Ok(())
 }
 }
 }
}

#[derive(Debug)]
pub enum SessionEvent {
 ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
 responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
 },
 Close,
 CloseSseStream {
 /// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
 http_request_id: Option<HttpRequestId>,
 /// Optional retry interval. If provided, a priming event is sent before closing.
 retry_interval: Option<Duration>,
 responder: oneshot::Sender<Result<(), SessionError>>,
 },
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close an SSE stream for a specific request.
 ///
 /// This closes the SSE connection for a POST request stream, but keeps the session
 /// and message cache active. Clients can reconnect using the `Last-Event-ID` header
 /// via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `http_request_id` - The HTTP request ID of the stream to close
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_sse_stream(
 &self,
 http_request_id: HttpRequestId,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: Some(http_request_id),
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close the standalone SSE stream.
 ///
 /// This closes the standalone SSE connection (established via GET request),
 /// but keeps the session and message cache active. Clients can reconnect using
 /// the `Last-Event-ID` header via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_standalone_sse_stream(
 &self,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: None,
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
 InnerEvent::FromHttpService(SessionEvent::Close) => {
 return Err(WorkerQuitReason::TransportClosed);
 }
 InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
 http_request_id,
 retry_interval,
 responder,
 }) => {
 let handle_result =
 self.close_sse_stream(http_request_id, retry_interval).await;
 let _ = responder.send(handle_result);
 }
 _ => {
 // ignore
 }
Expand Down

77 changes: 62 additions & 15 deletions
crates/rmcp/src/transport/streamable_http_server/tower.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
 /// The ping message duration for SSE connections.
 pub sse_keep_alive: Option<Duration>,
 /// The retry interval for SSE priming events.
 pub sse_retry: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 /// When enabled, SSE priming events are sent to enable client reconnection.
 pub stateful_mode: bool,
 /// Cancellation token for the Streamable HTTP server.
 ///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 sse_retry: Some(Duration::from_secs(3)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
Expand DownExpand Up@@ -216,6 +220,7 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 // Resume doesn't need priming - client already has the event ID
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
 .initialize_session(&session_id, message)
 .await
 .map_err(internal_error_response("create stream"))?;
 let stream = futures::stream::once(async move {
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(response)),
 retry: None,
 }
 });
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 let mut response = sse_stream_response(
 futures::stream::once({
 async move {
 ServerSseMessage {
 event_id: None,
 message: response.into(),
 }
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );
Expand All@@ -424,14 +468,17 @@ where
 // on service created
 let _ = service.waiting().await;
 });
 // Stateless mode: no priming (no session to resume)
 let stream = ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(message)),
 retry: None,
 }
 });
 Ok(sse_stream_response(
 ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: message.into(),
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms");
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

 Merged

alexhancock
merged 2 commits into

from

Jan 9, 2026
 Merged


Implement SEP-1699: Support SSE Polling via Server-Side Disconnect
#604
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


2 commits


Select commit
Hold shift + click to select a range
f2cb7f5

feat: implement SEP-1699 SSE polling via server-side disconnect


DaleSeo Dec 28, 2025
dbd5ae0

test: add tests for priming behavior on stream start and close


DaleSeo Dec 28, 2025

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


8 changes: 7 additions & 1 deletion
crates/rmcp/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
 "dep:http-body-util",
 "dep:bytes",
 "dep:sse-stream",
 "dep:axum",
 "tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"

24 changes: 21 additions & 3 deletions
crates/rmcp/src/transport/common/server_side_http.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
 /// The event ID for this message. When set, clients can use this ID
 /// with the `Last-Event-ID` header to resume the stream from this point.
 pub event_id: Option<String>,
 pub message: Arc<ServerJsonRpcMessage>,
 /// The JSON-RPC message content. Set to `None` for priming events.
 /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
 pub message: Option<Arc<ServerJsonRpcMessage>>,
 /// The retry interval hint for clients. Clients should wait this duration
 /// before attempting to reconnect. This maps to the SSE `retry:` field.
 pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
 use futures::StreamExt;
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 let mut sse = if let Some(ref msg) = message.message {
 let data = serde_json::to_string(msg.as_ref()).expect("valid message");
 Sse::default().data(data)
 } else {
 // Priming event: empty data per SEP-1699 (just "data:\n")
 Sse::default().data("")
 };

 sse.id = message.event_id;

 if let Some(retry) = message.retry {
 sse.retry = Some(retry.as_millis() as u64);
 }

 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
Expand Down

141 changes: 138 additions & 3 deletions
crates/rmcp/src/transport/streamable_http_server/session/local.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
 Self::new(tx, None)
 }

async fn send(&mut self, message: ServerJsonRpcMessage) {
 fn next_event_id(&self) -> EventId {
 let index = self.cache.back().map_or(0, |m| {
 m.event_id
 .as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
 .index
 + 1
 });
let event_id = EventId {
 EventId {
 http_request_id: self.http_request_id,
 index,
 }
 }

 async fn send(&mut self, message: ServerJsonRpcMessage) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Some(Arc::new(message)),
 retry: None,
 };
 self.cache_and_send(message).await;
 }

 async fn send_priming(&mut self, retry: Duration) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Arc::new(message),
 message: None,
 retry: Some(retry),
 };
 self.cache_and_send(message).await;
 }

 async fn cache_and_send(&mut self, message: ServerSseMessage) {
 if self.cache.len() >= self.capacity {
 self.cache.pop_front();
 self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
 }
 }
 }

 async fn close_sse_stream(
 &mut self,
 http_request_id: Option<HttpRequestId>,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 match http_request_id {
 // Close a request-wise stream
 Some(id) => {
 let request_wise = self
 .tx_router
 .get_mut(&id)
 .ok_or(SessionError::ChannelClosed(Some(id)))?;

 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 request_wise.tx.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 request_wise.tx.tx = tx;

 tracing::debug!(
 http_request_id = id,
 "closed SSE stream for server-initiated disconnection"
 );
 Ok(())
 }
 // Close the standalone (common) stream
 None => {
 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 self.common.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 self.common.tx = tx;

 tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
 Ok(())
 }
 }
 }
}

#[derive(Debug)]
pub enum SessionEvent {
 ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
 responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
 },
 Close,
 CloseSseStream {
 /// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
 http_request_id: Option<HttpRequestId>,
 /// Optional retry interval. If provided, a priming event is sent before closing.
 retry_interval: Option<Duration>,
 responder: oneshot::Sender<Result<(), SessionError>>,
 },
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close an SSE stream for a specific request.
 ///
 /// This closes the SSE connection for a POST request stream, but keeps the session
 /// and message cache active. Clients can reconnect using the `Last-Event-ID` header
 /// via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `http_request_id` - The HTTP request ID of the stream to close
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_sse_stream(
 &self,
 http_request_id: HttpRequestId,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: Some(http_request_id),
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close the standalone SSE stream.
 ///
 /// This closes the standalone SSE connection (established via GET request),
 /// but keeps the session and message cache active. Clients can reconnect using
 /// the `Last-Event-ID` header via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_standalone_sse_stream(
 &self,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: None,
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
 InnerEvent::FromHttpService(SessionEvent::Close) => {
 return Err(WorkerQuitReason::TransportClosed);
 }
 InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
 http_request_id,
 retry_interval,
 responder,
 }) => {
 let handle_result =
 self.close_sse_stream(http_request_id, retry_interval).await;
 let _ = responder.send(handle_result);
 }
 _ => {
 // ignore
 }
Expand Down

77 changes: 62 additions & 15 deletions
crates/rmcp/src/transport/streamable_http_server/tower.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
 /// The ping message duration for SSE connections.
 pub sse_keep_alive: Option<Duration>,
 /// The retry interval for SSE priming events.
 pub sse_retry: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 /// When enabled, SSE priming events are sent to enable client reconnection.
 pub stateful_mode: bool,
 /// Cancellation token for the Streamable HTTP server.
 ///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 sse_retry: Some(Duration::from_secs(3)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
Expand DownExpand Up@@ -216,6 +220,7 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 // Resume doesn't need priming - client already has the event ID
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
 .initialize_session(&session_id, message)
 .await
 .map_err(internal_error_response("create stream"))?;
 let stream = futures::stream::once(async move {
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(response)),
 retry: None,
 }
 });
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 let mut response = sse_stream_response(
 futures::stream::once({
 async move {
 ServerSseMessage {
 event_id: None,
 message: response.into(),
 }
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );
Expand All@@ -424,14 +468,17 @@ where
 // on service created
 let _ = service.waiting().await;
 });
 // Stateless mode: no priming (no session to resume)
 let stream = ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(message)),
 retry: None,
 }
 });
 Ok(sse_stream_response(
 ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: message.into(),
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links");
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

 Merged

alexhancock
merged 2 commits into

from

Jan 9, 2026
 Merged


Implement SEP-1699: Support SSE Polling via Server-Side Disconnect
#604
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


2 commits


Select commit
Hold shift + click to select a range
f2cb7f5

feat: implement SEP-1699 SSE polling via server-side disconnect


DaleSeo Dec 28, 2025
dbd5ae0

test: add tests for priming behavior on stream start and close


DaleSeo Dec 28, 2025

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


8 changes: 7 additions & 1 deletion
crates/rmcp/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
 "dep:http-body-util",
 "dep:bytes",
 "dep:sse-stream",
 "dep:axum",
 "tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"

24 changes: 21 additions & 3 deletions
crates/rmcp/src/transport/common/server_side_http.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
 /// The event ID for this message. When set, clients can use this ID
 /// with the `Last-Event-ID` header to resume the stream from this point.
 pub event_id: Option<String>,
 pub message: Arc<ServerJsonRpcMessage>,
 /// The JSON-RPC message content. Set to `None` for priming events.
 /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
 pub message: Option<Arc<ServerJsonRpcMessage>>,
 /// The retry interval hint for clients. Clients should wait this duration
 /// before attempting to reconnect. This maps to the SSE `retry:` field.
 pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
 use futures::StreamExt;
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 let mut sse = if let Some(ref msg) = message.message {
 let data = serde_json::to_string(msg.as_ref()).expect("valid message");
 Sse::default().data(data)
 } else {
 // Priming event: empty data per SEP-1699 (just "data:\n")
 Sse::default().data("")
 };

 sse.id = message.event_id;

 if let Some(retry) = message.retry {
 sse.retry = Some(retry.as_millis() as u64);
 }

 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
Expand Down

141 changes: 138 additions & 3 deletions
crates/rmcp/src/transport/streamable_http_server/session/local.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
 Self::new(tx, None)
 }

async fn send(&mut self, message: ServerJsonRpcMessage) {
 fn next_event_id(&self) -> EventId {
 let index = self.cache.back().map_or(0, |m| {
 m.event_id
 .as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
 .index
 + 1
 });
let event_id = EventId {
 EventId {
 http_request_id: self.http_request_id,
 index,
 }
 }

 async fn send(&mut self, message: ServerJsonRpcMessage) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Some(Arc::new(message)),
 retry: None,
 };
 self.cache_and_send(message).await;
 }

 async fn send_priming(&mut self, retry: Duration) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Arc::new(message),
 message: None,
 retry: Some(retry),
 };
 self.cache_and_send(message).await;
 }

 async fn cache_and_send(&mut self, message: ServerSseMessage) {
 if self.cache.len() >= self.capacity {
 self.cache.pop_front();
 self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
 }
 }
 }

 async fn close_sse_stream(
 &mut self,
 http_request_id: Option<HttpRequestId>,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 match http_request_id {
 // Close a request-wise stream
 Some(id) => {
 let request_wise = self
 .tx_router
 .get_mut(&id)
 .ok_or(SessionError::ChannelClosed(Some(id)))?;

 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 request_wise.tx.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 request_wise.tx.tx = tx;

 tracing::debug!(
 http_request_id = id,
 "closed SSE stream for server-initiated disconnection"
 );
 Ok(())
 }
 // Close the standalone (common) stream
 None => {
 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 self.common.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 self.common.tx = tx;

 tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
 Ok(())
 }
 }
 }
}

#[derive(Debug)]
pub enum SessionEvent {
 ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
 responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
 },
 Close,
 CloseSseStream {
 /// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
 http_request_id: Option<HttpRequestId>,
 /// Optional retry interval. If provided, a priming event is sent before closing.
 retry_interval: Option<Duration>,
 responder: oneshot::Sender<Result<(), SessionError>>,
 },
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close an SSE stream for a specific request.
 ///
 /// This closes the SSE connection for a POST request stream, but keeps the session
 /// and message cache active. Clients can reconnect using the `Last-Event-ID` header
 /// via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `http_request_id` - The HTTP request ID of the stream to close
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_sse_stream(
 &self,
 http_request_id: HttpRequestId,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: Some(http_request_id),
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close the standalone SSE stream.
 ///
 /// This closes the standalone SSE connection (established via GET request),
 /// but keeps the session and message cache active. Clients can reconnect using
 /// the `Last-Event-ID` header via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_standalone_sse_stream(
 &self,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: None,
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
 InnerEvent::FromHttpService(SessionEvent::Close) => {
 return Err(WorkerQuitReason::TransportClosed);
 }
 InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
 http_request_id,
 retry_interval,
 responder,
 }) => {
 let handle_result =
 self.close_sse_stream(http_request_id, retry_interval).await;
 let _ = responder.send(handle_result);
 }
 _ => {
 // ignore
 }
Expand Down

77 changes: 62 additions & 15 deletions
crates/rmcp/src/transport/streamable_http_server/tower.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
 /// The ping message duration for SSE connections.
 pub sse_keep_alive: Option<Duration>,
 /// The retry interval for SSE priming events.
 pub sse_retry: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 /// When enabled, SSE priming events are sent to enable client reconnection.
 pub stateful_mode: bool,
 /// Cancellation token for the Streamable HTTP server.
 ///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 sse_retry: Some(Duration::from_secs(3)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
Expand DownExpand Up@@ -216,6 +220,7 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 // Resume doesn't need priming - client already has the event ID
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
 .initialize_session(&session_id, message)
 .await
 .map_err(internal_error_response("create stream"))?;
 let stream = futures::stream::once(async move {
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(response)),
 retry: None,
 }
 });
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 let mut response = sse_stream_response(
 futures::stream::once({
 async move {
 ServerSseMessage {
 event_id: None,
 message: response.into(),
 }
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );
Expand All@@ -424,14 +468,17 @@ where
 // on service created
 let _ = service.waiting().await;
 });
 // Stateless mode: no priming (no session to resume)
 let stream = ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(message)),
 retry: None,
 }
 });
 Ok(sse_stream_response(
 ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: message.into(),
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default");
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

 Merged

alexhancock
merged 2 commits into

from

Jan 9, 2026
 Merged


Implement SEP-1699: Support SSE Polling via Server-Side Disconnect
#604
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


2 commits


Select commit
Hold shift + click to select a range
f2cb7f5

feat: implement SEP-1699 SSE polling via server-side disconnect


DaleSeo Dec 28, 2025
dbd5ae0

test: add tests for priming behavior on stream start and close


DaleSeo Dec 28, 2025

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


8 changes: 7 additions & 1 deletion
crates/rmcp/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
 "dep:http-body-util",
 "dep:bytes",
 "dep:sse-stream",
 "dep:axum",
 "tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"

24 changes: 21 additions & 3 deletions
crates/rmcp/src/transport/common/server_side_http.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
 /// The event ID for this message. When set, clients can use this ID
 /// with the `Last-Event-ID` header to resume the stream from this point.
 pub event_id: Option<String>,
 pub message: Arc<ServerJsonRpcMessage>,
 /// The JSON-RPC message content. Set to `None` for priming events.
 /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
 pub message: Option<Arc<ServerJsonRpcMessage>>,
 /// The retry interval hint for clients. Clients should wait this duration
 /// before attempting to reconnect. This maps to the SSE `retry:` field.
 pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
 use futures::StreamExt;
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 let mut sse = if let Some(ref msg) = message.message {
 let data = serde_json::to_string(msg.as_ref()).expect("valid message");
 Sse::default().data(data)
 } else {
 // Priming event: empty data per SEP-1699 (just "data:\n")
 Sse::default().data("")
 };

 sse.id = message.event_id;

 if let Some(retry) = message.retry {
 sse.retry = Some(retry.as_millis() as u64);
 }

 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
Expand Down

141 changes: 138 additions & 3 deletions
crates/rmcp/src/transport/streamable_http_server/session/local.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
 Self::new(tx, None)
 }

async fn send(&mut self, message: ServerJsonRpcMessage) {
 fn next_event_id(&self) -> EventId {
 let index = self.cache.back().map_or(0, |m| {
 m.event_id
 .as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
 .index
 + 1
 });
let event_id = EventId {
 EventId {
 http_request_id: self.http_request_id,
 index,
 }
 }

 async fn send(&mut self, message: ServerJsonRpcMessage) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Some(Arc::new(message)),
 retry: None,
 };
 self.cache_and_send(message).await;
 }

 async fn send_priming(&mut self, retry: Duration) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Arc::new(message),
 message: None,
 retry: Some(retry),
 };
 self.cache_and_send(message).await;
 }

 async fn cache_and_send(&mut self, message: ServerSseMessage) {
 if self.cache.len() >= self.capacity {
 self.cache.pop_front();
 self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
 }
 }
 }

 async fn close_sse_stream(
 &mut self,
 http_request_id: Option<HttpRequestId>,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 match http_request_id {
 // Close a request-wise stream
 Some(id) => {
 let request_wise = self
 .tx_router
 .get_mut(&id)
 .ok_or(SessionError::ChannelClosed(Some(id)))?;

 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 request_wise.tx.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 request_wise.tx.tx = tx;

 tracing::debug!(
 http_request_id = id,
 "closed SSE stream for server-initiated disconnection"
 );
 Ok(())
 }
 // Close the standalone (common) stream
 None => {
 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 self.common.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 self.common.tx = tx;

 tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
 Ok(())
 }
 }
 }
}

#[derive(Debug)]
pub enum SessionEvent {
 ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
 responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
 },
 Close,
 CloseSseStream {
 /// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
 http_request_id: Option<HttpRequestId>,
 /// Optional retry interval. If provided, a priming event is sent before closing.
 retry_interval: Option<Duration>,
 responder: oneshot::Sender<Result<(), SessionError>>,
 },
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close an SSE stream for a specific request.
 ///
 /// This closes the SSE connection for a POST request stream, but keeps the session
 /// and message cache active. Clients can reconnect using the `Last-Event-ID` header
 /// via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `http_request_id` - The HTTP request ID of the stream to close
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_sse_stream(
 &self,
 http_request_id: HttpRequestId,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: Some(http_request_id),
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close the standalone SSE stream.
 ///
 /// This closes the standalone SSE connection (established via GET request),
 /// but keeps the session and message cache active. Clients can reconnect using
 /// the `Last-Event-ID` header via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_standalone_sse_stream(
 &self,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: None,
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
 InnerEvent::FromHttpService(SessionEvent::Close) => {
 return Err(WorkerQuitReason::TransportClosed);
 }
 InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
 http_request_id,
 retry_interval,
 responder,
 }) => {
 let handle_result =
 self.close_sse_stream(http_request_id, retry_interval).await;
 let _ = responder.send(handle_result);
 }
 _ => {
 // ignore
 }
Expand Down

77 changes: 62 additions & 15 deletions
crates/rmcp/src/transport/streamable_http_server/tower.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
 /// The ping message duration for SSE connections.
 pub sse_keep_alive: Option<Duration>,
 /// The retry interval for SSE priming events.
 pub sse_retry: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 /// When enabled, SSE priming events are sent to enable client reconnection.
 pub stateful_mode: bool,
 /// Cancellation token for the Streamable HTTP server.
 ///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 sse_retry: Some(Duration::from_secs(3)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
Expand DownExpand Up@@ -216,6 +220,7 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 // Resume doesn't need priming - client already has the event ID
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
 .initialize_session(&session_id, message)
 .await
 .map_err(internal_error_response("create stream"))?;
 let stream = futures::stream::once(async move {
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(response)),
 retry: None,
 }
 });
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 let mut response = sse_stream_response(
 futures::stream::once({
 async move {
 ServerSseMessage {
 event_id: None,
 message: response.into(),
 }
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );
Expand All@@ -424,14 +468,17 @@ where
 // on service created
 let _ = service.waiting().await;
 });
 // Stateless mode: no priming (no session to resume)
 let stream = ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(message)),
 retry: None,
 }
 });
 Ok(sse_stream_response(
 ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: message.into(),
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers");
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

 Merged

alexhancock
merged 2 commits into

from

Jan 9, 2026
 Merged


Implement SEP-1699: Support SSE Polling via Server-Side Disconnect
#604
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


2 commits


Select commit
Hold shift + click to select a range
f2cb7f5

feat: implement SEP-1699 SSE polling via server-side disconnect


DaleSeo Dec 28, 2025
dbd5ae0

test: add tests for priming behavior on stream start and close


DaleSeo Dec 28, 2025

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


8 changes: 7 additions & 1 deletion
crates/rmcp/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
 "dep:http-body-util",
 "dep:bytes",
 "dep:sse-stream",
 "dep:axum",
 "tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"

24 changes: 21 additions & 3 deletions
crates/rmcp/src/transport/common/server_side_http.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
 /// The event ID for this message. When set, clients can use this ID
 /// with the `Last-Event-ID` header to resume the stream from this point.
 pub event_id: Option<String>,
 pub message: Arc<ServerJsonRpcMessage>,
 /// The JSON-RPC message content. Set to `None` for priming events.
 /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
 pub message: Option<Arc<ServerJsonRpcMessage>>,
 /// The retry interval hint for clients. Clients should wait this duration
 /// before attempting to reconnect. This maps to the SSE `retry:` field.
 pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
 use futures::StreamExt;
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 let mut sse = if let Some(ref msg) = message.message {
 let data = serde_json::to_string(msg.as_ref()).expect("valid message");
 Sse::default().data(data)
 } else {
 // Priming event: empty data per SEP-1699 (just "data:\n")
 Sse::default().data("")
 };

 sse.id = message.event_id;

 if let Some(retry) = message.retry {
 sse.retry = Some(retry.as_millis() as u64);
 }

 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
Expand Down

141 changes: 138 additions & 3 deletions
crates/rmcp/src/transport/streamable_http_server/session/local.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
 Self::new(tx, None)
 }

async fn send(&mut self, message: ServerJsonRpcMessage) {
 fn next_event_id(&self) -> EventId {
 let index = self.cache.back().map_or(0, |m| {
 m.event_id
 .as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
 .index
 + 1
 });
let event_id = EventId {
 EventId {
 http_request_id: self.http_request_id,
 index,
 }
 }

 async fn send(&mut self, message: ServerJsonRpcMessage) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Some(Arc::new(message)),
 retry: None,
 };
 self.cache_and_send(message).await;
 }

 async fn send_priming(&mut self, retry: Duration) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Arc::new(message),
 message: None,
 retry: Some(retry),
 };
 self.cache_and_send(message).await;
 }

 async fn cache_and_send(&mut self, message: ServerSseMessage) {
 if self.cache.len() >= self.capacity {
 self.cache.pop_front();
 self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
 }
 }
 }

 async fn close_sse_stream(
 &mut self,
 http_request_id: Option<HttpRequestId>,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 match http_request_id {
 // Close a request-wise stream
 Some(id) => {
 let request_wise = self
 .tx_router
 .get_mut(&id)
 .ok_or(SessionError::ChannelClosed(Some(id)))?;

 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 request_wise.tx.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 request_wise.tx.tx = tx;

 tracing::debug!(
 http_request_id = id,
 "closed SSE stream for server-initiated disconnection"
 );
 Ok(())
 }
 // Close the standalone (common) stream
 None => {
 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 self.common.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 self.common.tx = tx;

 tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
 Ok(())
 }
 }
 }
}

#[derive(Debug)]
pub enum SessionEvent {
 ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
 responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
 },
 Close,
 CloseSseStream {
 /// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
 http_request_id: Option<HttpRequestId>,
 /// Optional retry interval. If provided, a priming event is sent before closing.
 retry_interval: Option<Duration>,
 responder: oneshot::Sender<Result<(), SessionError>>,
 },
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close an SSE stream for a specific request.
 ///
 /// This closes the SSE connection for a POST request stream, but keeps the session
 /// and message cache active. Clients can reconnect using the `Last-Event-ID` header
 /// via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `http_request_id` - The HTTP request ID of the stream to close
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_sse_stream(
 &self,
 http_request_id: HttpRequestId,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: Some(http_request_id),
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close the standalone SSE stream.
 ///
 /// This closes the standalone SSE connection (established via GET request),
 /// but keeps the session and message cache active. Clients can reconnect using
 /// the `Last-Event-ID` header via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_standalone_sse_stream(
 &self,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: None,
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
 InnerEvent::FromHttpService(SessionEvent::Close) => {
 return Err(WorkerQuitReason::TransportClosed);
 }
 InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
 http_request_id,
 retry_interval,
 responder,
 }) => {
 let handle_result =
 self.close_sse_stream(http_request_id, retry_interval).await;
 let _ = responder.send(handle_result);
 }
 _ => {
 // ignore
 }
Expand Down

77 changes: 62 additions & 15 deletions
crates/rmcp/src/transport/streamable_http_server/tower.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
 /// The ping message duration for SSE connections.
 pub sse_keep_alive: Option<Duration>,
 /// The retry interval for SSE priming events.
 pub sse_retry: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 /// When enabled, SSE priming events are sent to enable client reconnection.
 pub stateful_mode: bool,
 /// Cancellation token for the Streamable HTTP server.
 ///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 sse_retry: Some(Duration::from_secs(3)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
Expand DownExpand Up@@ -216,6 +220,7 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 // Resume doesn't need priming - client already has the event ID
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
 .initialize_session(&session_id, message)
 .await
 .map_err(internal_error_response("create stream"))?;
 let stream = futures::stream::once(async move {
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(response)),
 retry: None,
 }
 });
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 let mut response = sse_stream_response(
 futures::stream::once({
 async move {
 ServerSseMessage {
 event_id: None,
 message: response.into(),
 }
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );
Expand All@@ -424,14 +468,17 @@ where
 // on service created
 let _ = service.waiting().await;
 });
 // Stateless mode: no priming (no session to resume)
 let stream = ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(message)),
 retry: None,
 }
 });
 Ok(sse_stream_response(
 ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: message.into(),
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode");
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
Implement SEP-1699: Support SSE Polling via Server-Side Disconnect#604

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

 Merged

alexhancock
merged 2 commits into

from

Jan 9, 2026
 Merged


Implement SEP-1699: Support SSE Polling via Server-Side Disconnect
#604
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


2 commits


Select commit
Hold shift + click to select a range
f2cb7f5

feat: implement SEP-1699 SSE polling via server-side disconnect


DaleSeo Dec 28, 2025
dbd5ae0

test: add tests for priming behavior on stream start and close


DaleSeo Dec 28, 2025

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


8 changes: 7 additions & 1 deletion
crates/rmcp/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -97,6 +97,7 @@ server-side-http = [
 "dep:http-body-util",
 "dep:bytes",
 "dep:sse-stream",
 "dep:axum",
 "tower",
]

Expand DownExpand Up@@ -201,4 +202,9 @@ path = "tests/test_elicitation.rs"
[[test]]
name = "test_task"
required-features = ["server", "client", "macros"]
path = "tests/test_task.rs"
path = "tests/test_task.rs"

[[test]]
name = "test_streamable_http_priming"
required-features = ["server", "client", "transport-streamable-http-server", "reqwest"]
path = "tests/test_streamable_http_priming.rs"

24 changes: 21 additions & 3 deletions
crates/rmcp/src/transport/common/server_side_http.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -59,8 +59,15 @@ impl sse_stream::Timer for TokioTimer {

#[derive(Debug, Clone)]
pub struct ServerSseMessage {
 /// The event ID for this message. When set, clients can use this ID
 /// with the `Last-Event-ID` header to resume the stream from this point.
 pub event_id: Option<String>,
 pub message: Arc<ServerJsonRpcMessage>,
 /// The JSON-RPC message content. Set to `None` for priming events.
 /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699)
 pub message: Option<Arc<ServerJsonRpcMessage>>,
 /// The retry interval hint for clients. Clients should wait this duration
 /// before attempting to reconnect. This maps to the SSE `retry:` field.
 pub retry: Option<Duration>,
}

pub(crate) fn sse_stream_response(
Expand All@@ -71,9 +78,20 @@ pub(crate) fn sse_stream_response(
 use futures::StreamExt;
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 let mut sse = if let Some(ref msg) = message.message {
 let data = serde_json::to_string(msg.as_ref()).expect("valid message");
 Sse::default().data(data)
 } else {
 // Priming event: empty data per SEP-1699 (just "data:\n")
 Sse::default().data("")
 };

 sse.id = message.event_id;

 if let Some(retry) = message.retry {
 sse.retry = Some(retry.as_millis() as u64);
 }

 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
Expand Down

141 changes: 138 additions & 3 deletions
crates/rmcp/src/transport/streamable_http_server/session/local.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ impl CachedTx {
 Self::new(tx, None)
 }

async fn send(&mut self, message: ServerJsonRpcMessage) {
 fn next_event_id(&self) -> EventId {
 let index = self.cache.back().map_or(0, |m| {
 m.event_id
 .as_deref()
Expand All@@ -211,14 +211,33 @@ impl CachedTx {
 .index
 + 1
 });
let event_id = EventId {
 EventId {
 http_request_id: self.http_request_id,
 index,
 }
 }

 async fn send(&mut self, message: ServerJsonRpcMessage) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Some(Arc::new(message)),
 retry: None,
 };
 self.cache_and_send(message).await;
 }

 async fn send_priming(&mut self, retry: Duration) {
 let event_id = self.next_event_id();
 let message = ServerSseMessage {
 event_id: Some(event_id.to_string()),
 message: Arc::new(message),
 message: None,
 retry: Some(retry),
 };
 self.cache_and_send(message).await;
 }

 async fn cache_and_send(&mut self, message: ServerSseMessage) {
 if self.cache.len() >= self.capacity {
 self.cache.pop_front();
 self.cache.push_back(message.clone());
Expand DownExpand Up@@ -525,7 +544,53 @@ impl LocalSessionWorker {
 }
 }
 }

 async fn close_sse_stream(
 &mut self,
 http_request_id: Option<HttpRequestId>,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 match http_request_id {
 // Close a request-wise stream
 Some(id) => {
 let request_wise = self
 .tx_router
 .get_mut(&id)
 .ok_or(SessionError::ChannelClosed(Some(id)))?;

 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 request_wise.tx.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 request_wise.tx.tx = tx;

 tracing::debug!(
 http_request_id = id,
 "closed SSE stream for server-initiated disconnection"
 );
 Ok(())
 }
 // Close the standalone (common) stream
 None => {
 // Send priming event if retry interval is specified
 if let Some(interval) = retry_interval {
 self.common.send_priming(interval).await;
 }

 // Close the stream by dropping the sender
 let (tx, _rx) = tokio::sync::mpsc::channel(1);
 self.common.tx = tx;

 tracing::debug!("closed standalone SSE stream for server-initiated disconnection");
 Ok(())
 }
 }
 }
}

#[derive(Debug)]
pub enum SessionEvent {
 ClientMessage {
Expand All@@ -548,6 +613,13 @@ pub enum SessionEvent {
 responder: oneshot::Sender<Result<ServerJsonRpcMessage, SessionError>>,
 },
 Close,
 CloseSseStream {
 /// The HTTP request ID to close. If `None`, closes the standalone (common) stream.
 http_request_id: Option<HttpRequestId>,
 /// Optional retry interval. If provided, a priming event is sent before closing.
 retry_interval: Option<Duration>,
 responder: oneshot::Sender<Result<(), SessionError>>,
 },
}

#[derive(Debug, Clone)]
Expand DownExpand Up@@ -683,6 +755,60 @@ impl LocalSessionHandle {
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close an SSE stream for a specific request.
 ///
 /// This closes the SSE connection for a POST request stream, but keeps the session
 /// and message cache active. Clients can reconnect using the `Last-Event-ID` header
 /// via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `http_request_id` - The HTTP request ID of the stream to close
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_sse_stream(
 &self,
 http_request_id: HttpRequestId,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: Some(http_request_id),
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }

 /// Close the standalone SSE stream.
 ///
 /// This closes the standalone SSE connection (established via GET request),
 /// but keeps the session and message cache active. Clients can reconnect using
 /// the `Last-Event-ID` header via a GET request to resume receiving messages.
 ///
 /// # Arguments
 ///
 /// * `retry_interval` - Optional retry interval. If provided, a priming event is sent
 pub async fn close_standalone_sse_stream(
 &self,
 retry_interval: Option<Duration>,
 ) -> Result<(), SessionError> {
 let (tx, rx) = tokio::sync::oneshot::channel();
 self.event_tx
 .send(SessionEvent::CloseSseStream {
 http_request_id: None,
 retry_interval,
 responder: tx,
 })
 .await
 .map_err(|_| SessionError::SessionServiceTerminated)?;
 rx.await
 .map_err(|_| SessionError::SessionServiceTerminated)?
 }
}

pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
Expand DownExpand Up@@ -848,6 +974,15 @@ impl Worker for LocalSessionWorker {
 InnerEvent::FromHttpService(SessionEvent::Close) => {
 return Err(WorkerQuitReason::TransportClosed);
 }
 InnerEvent::FromHttpService(SessionEvent::CloseSseStream {
 http_request_id,
 retry_interval,
 responder,
 }) => {
 let handle_result =
 self.close_sse_stream(http_request_id, retry_interval).await;
 let _ = responder.send(handle_result);
 }
 _ => {
 // ignore
 }
Expand Down

77 changes: 62 additions & 15 deletions
crates/rmcp/src/transport/streamable_http_server/tower.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,10 @@ use crate::{
pub struct StreamableHttpServerConfig {
 /// The ping message duration for SSE connections.
 pub sse_keep_alive: Option<Duration>,
 /// The retry interval for SSE priming events.
 pub sse_retry: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 /// When enabled, SSE priming events are sent to enable client reconnection.
 pub stateful_mode: bool,
 /// Cancellation token for the Streamable HTTP server.
 ///
Expand All@@ -45,6 +48,7 @@ impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 sse_retry: Some(Duration::from_secs(3)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
Expand DownExpand Up@@ -216,6 +220,7 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 // Resume doesn't need priming - client already has the event ID
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand All@@ -228,6 +233,19 @@ where
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -322,6 +340,19 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
Expand DownExpand Up@@ -389,15 +420,28 @@ where
 .initialize_session(&session_id, message)
 .await
 .map_err(internal_error_response("create stream"))?;
 let stream = futures::stream::once(async move {
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(response)),
 retry: None,
 }
 });
 // Prepend priming event if sse_retry configured
 let stream = if let Some(retry) = self.config.sse_retry {
 let priming = ServerSseMessage {
 event_id: Some("0".into()),
 message: None,
 retry: Some(retry),
 };
 futures::stream::once(async move { priming })
 .chain(stream)
 .left_stream()
 } else {
 stream.right_stream()
 };
 let mut response = sse_stream_response(
 futures::stream::once({
 async move {
 ServerSseMessage {
 event_id: None,
 message: response.into(),
 }
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );
Expand All@@ -424,14 +468,17 @@ where
 // on service created
 let _ = service.waiting().await;
 });
 // Stateless mode: no priming (no session to resume)
 let stream = ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: Some(Arc::new(message)),
 retry: None,
 }
 });
 Ok(sse_stream_response(
 ReceiverStream::new(receiver).map(|message| {
 tracing::info!(?message);
 ServerSseMessage {
 event_id: None,
 message: message.into(),
 }
 }),
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
Expand Down
Loading