Skip to content

fix(streamable-http): gracefully shutdown while client connected#494

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
jokemanfire merged 3 commits into from
Dec 2, 2025
Merged

fix(streamable-http): gracefully shutdown while client connected #494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
dc2e989
fix(streamable-http): gracefully shutdown while client connected
greenhat616 Oct 21, 2025
370b184
fix: adviced comments
greenhat616 Nov 6, 2025
73c466a
fix: windows test build
greenhat616 Nov 6, 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
19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
keep_alive: Option<Duration>,
ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
use futures::StreamExt;
let stream = SseBody::new(stream.map(|message| {
let data = serde_json::to_string(&message.message).expect("valid message");
let mut sse = Sse::default().data(data);
sse.id = message.event_id;
Result::<Sse, Infallible>::Ok(sse)
}));
let stream = stream
.map(|message| {
let data = serde_json::to_string(&message.message).expect("valid message");
let mut sse = Sse::default().data(data);
sse.id = message.event_id;
Result::<Sse, Infallible>::Ok(sse)
})
.take_until(async move { ct.cancelled().await });
let stream = SseBody::new(stream);

let stream = match keep_alive {
Some(duration) => stream
.with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
.boxed(),
None => stream.boxed(),
};

Response::builder()
.status(http::StatusCode::OK)
.header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down
27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
pub sse_keep_alive: Option<Duration>,
/// If true, the server will create a session for each request and keep it alive.
pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.
/// Cancellation token for the Streamable HTTP server.
///
/// When this token is cancelled, all active sessions are terminated and
/// the server stops accepting new requests.
pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
fn default() -> Self {
Self {
sse_keep_alive: Some(Duration::from_secs(15)),
stateful_mode: true,
cancellation_token: CancellationToken::new(),
}
}
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
.resume(&session_id, last_event_id)
.await
.map_err(internal_error_response("resume session"))?;
Ok(sse_stream_response(stream, self.config.sse_keep_alive))
Ok(sse_stream_response(
stream,
self.config.sse_keep_alive,
self.config.cancellation_token.child_token(),
))
} else {
// create standalone stream
let stream = self
.session_manager
.create_standalone_stream(&session_id)
.await
.map_err(internal_error_response("create standalone stream"))?;
Ok(sse_stream_response(stream, self.config.sse_keep_alive))
Ok(sse_stream_response(
stream,
self.config.sse_keep_alive,
self.config.cancellation_token.child_token(),
))
}
}

Expand DownExpand Up@@ -307,7 +322,11 @@ where
.create_stream(&session_id, message)
.await
.map_err(internal_error_response("get session"))?;
Ok(sse_stream_response(stream, self.config.sse_keep_alive))
Ok(sse_stream_response(
stream,
self.config.sse_keep_alive,
self.config.cancellation_token.child_token(),
))
}
ClientJsonRpcMessage::Notification(_)
| ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
}
}),
self.config.sse_keep_alive,
self.config.cancellation_token.child_token(),
);

response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
}
}),
self.config.sse_keep_alive,
self.config.cancellation_token.child_token(),
))
}
ClientJsonRpcMessage::Notification(_notification) => {
Expand Down
4 changes: 3 additions & 1 deletion crates/rmcp/tests/test_with_js.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
.wait()
.await?;

let ct = CancellationToken::new();
let service: StreamableHttpService<Calculator, LocalSessionManager> =
StreamableHttpService::new(
|| Ok(Calculator::new()),
Default::default(),
StreamableHttpServerConfig {
stateful_mode: true,
sse_keep_alive: None,
cancellation_token: ct.child_token(),
},
);
let router = axum::Router::new().nest_service("/mcp", service);
let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
let ct = CancellationToken::new();

let handle = tokio::spawn({
let ct = ct.clone();
async move {
Expand Down
13 changes: 10 additions & 3 deletions examples/servers/src/counter_streamhttp.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
)
.with(tracing_subscriber::fmt::layer())
.init();
let ct = tokio_util::sync::CancellationToken::new();

let service = StreamableHttpService::new(
|| Ok(Counter::new()),
LocalSessionManager::default().into(),
Default::default(),
StreamableHttpServerConfig {
cancellation_token: ct.child_token(),
..Default::default()
},
);

let router = axum::Router::new().nest_service("/mcp", service);
let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
let _ = axum::serve(tcp_listener, router)
.with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
.with_graceful_shutdown(async move {
tokio::signal::ctrl_c().await.unwrap();
ct.cancel();
})
.await;
Ok(())
}
4 changes: 2 additions & 2 deletions examples/transport/src/named-pipe.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
let mut server = ServerOptions::new()
.first_pipe_instance(true)
.create(name)?;
while let Ok(_) = server.connect().await {
while server.connect().await.is_ok() {
let stream = server;
server = ServerOptions::new().create(name)?;
tokio::spawn(async move {
match serve_server(Calculator, stream).await {
match serve_server(Calculator::new(), stream).await {
Ok(server) => {
println!("Server initialized successfully");
if let Err(e) = server.waiting().await {
Expand Down
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
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;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
fix(streamable-http): gracefully shutdown while client connected by greenhat616 · Pull Request #494 · modelcontextprotocol/rust-sdk · GitHub
Skip to content
fix(streamable-http): gracefully shutdown while client connected#494

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

jokemanfire
merged 3 commits into

from

Dec 2, 2025
 Merged


fix(streamable-http): gracefully shutdown while client connected
#494
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
dc2e989

fix(streamable-http): gracefully shutdown while client connected


greenhat616 Oct 21, 2025
370b184

fix: adviced comments


greenhat616 Nov 6, 2025
73c466a

fix: windows test build


greenhat616 Nov 6, 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


19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
 stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
 keep_alive: Option<Duration>,
 ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
 use futures::StreamExt;
 let stream = SseBody::new(stream.map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 }));
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
 let stream = SseBody::new(stream);

 let stream = match keep_alive {
 Some(duration) => stream
 .with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
 .boxed(),
 None => stream.boxed(),
 };

 Response::builder()
 .status(http::StatusCode::OK)
 .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down

27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
 pub sse_keep_alive: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.

 /// Cancellation token for the Streamable HTTP server.
 ///
 /// When this token is cancelled, all active sessions are terminated and
 /// the server stops accepting new requests.
 pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
 }
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 } else {
 // create standalone stream
 let stream = self
 .session_manager
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 }

Expand DownExpand Up@@ -307,7 +322,11 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_)
 | ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );

 response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_notification) => {
Expand Down

4 changes: 3 additions & 1 deletion
crates/rmcp/tests/test_with_js.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 .wait()
 .await?;

 let ct = CancellationToken::new();
 let service: StreamableHttpService<Calculator, LocalSessionManager> =
 StreamableHttpService::new(
 || Ok(Calculator::new()),
 Default::default(),
 StreamableHttpServerConfig {
 stateful_mode: true,
 sse_keep_alive: None,
 cancellation_token: ct.child_token(),
 },
 );
 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
 let ct = CancellationToken::new();

 let handle = tokio::spawn({
 let ct = ct.clone();
 async move {
Expand Down

13 changes: 10 additions & 3 deletions
examples/servers/src/counter_streamhttp.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
 StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
 layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
 )
 .with(tracing_subscriber::fmt::layer())
 .init();
 let ct = tokio_util::sync::CancellationToken::new();

 let service = StreamableHttpService::new(
 || Ok(Counter::new()),
 LocalSessionManager::default().into(),
 Default::default(),
 StreamableHttpServerConfig {
 cancellation_token: ct.child_token(),
 ..Default::default()
 },
 );

 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
 let _ = axum::serve(tcp_listener, router)
 .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
 .with_graceful_shutdown(async move {
 tokio::signal::ctrl_c().await.unwrap();
 ct.cancel();
 })
 .await;
 Ok(())
}

4 changes: 2 additions & 2 deletions
examples/transport/src/named-pipe.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
 let mut server = ServerOptions::new()
 .first_pipe_instance(true)
 .create(name)?;
 while let Ok(_) = server.connect().await {
 while server.connect().await.is_ok() {
 let stream = server;
 server = ServerOptions::new().create(name)?;
 tokio::spawn(async move {
 match serve_server(Calculator, stream).await {
 match serve_server(Calculator::new(), stream).await {
 Ok(server) => {
 println!("Server initialized successfully");
 if let Err(e) = server.waiting().await {
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
fix(streamable-http): gracefully shutdown while client connected by greenhat616 · Pull Request #494 · modelcontextprotocol/rust-sdk · GitHub
Skip to content
fix(streamable-http): gracefully shutdown while client connected#494

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

jokemanfire
merged 3 commits into

from

Dec 2, 2025
 Merged


fix(streamable-http): gracefully shutdown while client connected
#494
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
dc2e989

fix(streamable-http): gracefully shutdown while client connected


greenhat616 Oct 21, 2025
370b184

fix: adviced comments


greenhat616 Nov 6, 2025
73c466a

fix: windows test build


greenhat616 Nov 6, 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


19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
 stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
 keep_alive: Option<Duration>,
 ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
 use futures::StreamExt;
 let stream = SseBody::new(stream.map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 }));
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
 let stream = SseBody::new(stream);

 let stream = match keep_alive {
 Some(duration) => stream
 .with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
 .boxed(),
 None => stream.boxed(),
 };

 Response::builder()
 .status(http::StatusCode::OK)
 .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down

27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
 pub sse_keep_alive: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.

 /// Cancellation token for the Streamable HTTP server.
 ///
 /// When this token is cancelled, all active sessions are terminated and
 /// the server stops accepting new requests.
 pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
 }
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 } else {
 // create standalone stream
 let stream = self
 .session_manager
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 }

Expand DownExpand Up@@ -307,7 +322,11 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_)
 | ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );

 response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_notification) => {
Expand Down

4 changes: 3 additions & 1 deletion
crates/rmcp/tests/test_with_js.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 .wait()
 .await?;

 let ct = CancellationToken::new();
 let service: StreamableHttpService<Calculator, LocalSessionManager> =
 StreamableHttpService::new(
 || Ok(Calculator::new()),
 Default::default(),
 StreamableHttpServerConfig {
 stateful_mode: true,
 sse_keep_alive: None,
 cancellation_token: ct.child_token(),
 },
 );
 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
 let ct = CancellationToken::new();

 let handle = tokio::spawn({
 let ct = ct.clone();
 async move {
Expand Down

13 changes: 10 additions & 3 deletions
examples/servers/src/counter_streamhttp.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
 StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
 layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
 )
 .with(tracing_subscriber::fmt::layer())
 .init();
 let ct = tokio_util::sync::CancellationToken::new();

 let service = StreamableHttpService::new(
 || Ok(Counter::new()),
 LocalSessionManager::default().into(),
 Default::default(),
 StreamableHttpServerConfig {
 cancellation_token: ct.child_token(),
 ..Default::default()
 },
 );

 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
 let _ = axum::serve(tcp_listener, router)
 .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
 .with_graceful_shutdown(async move {
 tokio::signal::ctrl_c().await.unwrap();
 ct.cancel();
 })
 .await;
 Ok(())
}

4 changes: 2 additions & 2 deletions
examples/transport/src/named-pipe.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
 let mut server = ServerOptions::new()
 .first_pipe_instance(true)
 .create(name)?;
 while let Ok(_) = server.connect().await {
 while server.connect().await.is_ok() {
 let stream = server;
 server = ServerOptions::new().create(name)?;
 tokio::spawn(async move {
 match serve_server(Calculator, stream).await {
 match serve_server(Calculator::new(), stream).await {
 Ok(server) => {
 println!("Server initialized successfully");
 if let Err(e) = server.waiting().await {
Expand Down

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

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

jokemanfire
merged 3 commits into

from

Dec 2, 2025
 Merged


fix(streamable-http): gracefully shutdown while client connected
#494
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
dc2e989

fix(streamable-http): gracefully shutdown while client connected


greenhat616 Oct 21, 2025
370b184

fix: adviced comments


greenhat616 Nov 6, 2025
73c466a

fix: windows test build


greenhat616 Nov 6, 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


19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
 stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
 keep_alive: Option<Duration>,
 ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
 use futures::StreamExt;
 let stream = SseBody::new(stream.map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 }));
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
 let stream = SseBody::new(stream);

 let stream = match keep_alive {
 Some(duration) => stream
 .with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
 .boxed(),
 None => stream.boxed(),
 };

 Response::builder()
 .status(http::StatusCode::OK)
 .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down

27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
 pub sse_keep_alive: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.

 /// Cancellation token for the Streamable HTTP server.
 ///
 /// When this token is cancelled, all active sessions are terminated and
 /// the server stops accepting new requests.
 pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
 }
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 } else {
 // create standalone stream
 let stream = self
 .session_manager
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 }

Expand DownExpand Up@@ -307,7 +322,11 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_)
 | ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );

 response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_notification) => {
Expand Down

4 changes: 3 additions & 1 deletion
crates/rmcp/tests/test_with_js.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 .wait()
 .await?;

 let ct = CancellationToken::new();
 let service: StreamableHttpService<Calculator, LocalSessionManager> =
 StreamableHttpService::new(
 || Ok(Calculator::new()),
 Default::default(),
 StreamableHttpServerConfig {
 stateful_mode: true,
 sse_keep_alive: None,
 cancellation_token: ct.child_token(),
 },
 );
 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
 let ct = CancellationToken::new();

 let handle = tokio::spawn({
 let ct = ct.clone();
 async move {
Expand Down

13 changes: 10 additions & 3 deletions
examples/servers/src/counter_streamhttp.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
 StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
 layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
 )
 .with(tracing_subscriber::fmt::layer())
 .init();
 let ct = tokio_util::sync::CancellationToken::new();

 let service = StreamableHttpService::new(
 || Ok(Counter::new()),
 LocalSessionManager::default().into(),
 Default::default(),
 StreamableHttpServerConfig {
 cancellation_token: ct.child_token(),
 ..Default::default()
 },
 );

 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
 let _ = axum::serve(tcp_listener, router)
 .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
 .with_graceful_shutdown(async move {
 tokio::signal::ctrl_c().await.unwrap();
 ct.cancel();
 })
 .await;
 Ok(())
}

4 changes: 2 additions & 2 deletions
examples/transport/src/named-pipe.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
 let mut server = ServerOptions::new()
 .first_pipe_instance(true)
 .create(name)?;
 while let Ok(_) = server.connect().await {
 while server.connect().await.is_ok() {
 let stream = server;
 server = ServerOptions::new().create(name)?;
 tokio::spawn(async move {
 match serve_server(Calculator, stream).await {
 match serve_server(Calculator::new(), stream).await {
 Ok(server) => {
 println!("Server initialized successfully");
 if let Err(e) = server.waiting().await {
Expand Down

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

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

jokemanfire
merged 3 commits into

from

Dec 2, 2025
 Merged


fix(streamable-http): gracefully shutdown while client connected
#494
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
dc2e989

fix(streamable-http): gracefully shutdown while client connected


greenhat616 Oct 21, 2025
370b184

fix: adviced comments


greenhat616 Nov 6, 2025
73c466a

fix: windows test build


greenhat616 Nov 6, 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


19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
 stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
 keep_alive: Option<Duration>,
 ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
 use futures::StreamExt;
 let stream = SseBody::new(stream.map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 }));
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
 let stream = SseBody::new(stream);

 let stream = match keep_alive {
 Some(duration) => stream
 .with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
 .boxed(),
 None => stream.boxed(),
 };

 Response::builder()
 .status(http::StatusCode::OK)
 .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down

27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
 pub sse_keep_alive: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.

 /// Cancellation token for the Streamable HTTP server.
 ///
 /// When this token is cancelled, all active sessions are terminated and
 /// the server stops accepting new requests.
 pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
 }
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 } else {
 // create standalone stream
 let stream = self
 .session_manager
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 }

Expand DownExpand Up@@ -307,7 +322,11 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_)
 | ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );

 response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_notification) => {
Expand Down

4 changes: 3 additions & 1 deletion
crates/rmcp/tests/test_with_js.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 .wait()
 .await?;

 let ct = CancellationToken::new();
 let service: StreamableHttpService<Calculator, LocalSessionManager> =
 StreamableHttpService::new(
 || Ok(Calculator::new()),
 Default::default(),
 StreamableHttpServerConfig {
 stateful_mode: true,
 sse_keep_alive: None,
 cancellation_token: ct.child_token(),
 },
 );
 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
 let ct = CancellationToken::new();

 let handle = tokio::spawn({
 let ct = ct.clone();
 async move {
Expand Down

13 changes: 10 additions & 3 deletions
examples/servers/src/counter_streamhttp.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
 StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
 layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
 )
 .with(tracing_subscriber::fmt::layer())
 .init();
 let ct = tokio_util::sync::CancellationToken::new();

 let service = StreamableHttpService::new(
 || Ok(Counter::new()),
 LocalSessionManager::default().into(),
 Default::default(),
 StreamableHttpServerConfig {
 cancellation_token: ct.child_token(),
 ..Default::default()
 },
 );

 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
 let _ = axum::serve(tcp_listener, router)
 .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
 .with_graceful_shutdown(async move {
 tokio::signal::ctrl_c().await.unwrap();
 ct.cancel();
 })
 .await;
 Ok(())
}

4 changes: 2 additions & 2 deletions
examples/transport/src/named-pipe.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
 let mut server = ServerOptions::new()
 .first_pipe_instance(true)
 .create(name)?;
 while let Ok(_) = server.connect().await {
 while server.connect().await.is_ok() {
 let stream = server;
 server = ServerOptions::new().create(name)?;
 tokio::spawn(async move {
 match serve_server(Calculator, stream).await {
 match serve_server(Calculator::new(), stream).await {
 Ok(server) => {
 println!("Server initialized successfully");
 if let Err(e) = server.waiting().await {
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
fix(streamable-http): gracefully shutdown while client connected by greenhat616 · Pull Request #494 · modelcontextprotocol/rust-sdk · GitHub
Skip to content
fix(streamable-http): gracefully shutdown while client connected#494

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

jokemanfire
merged 3 commits into

from

Dec 2, 2025
 Merged


fix(streamable-http): gracefully shutdown while client connected
#494
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
dc2e989

fix(streamable-http): gracefully shutdown while client connected


greenhat616 Oct 21, 2025
370b184

fix: adviced comments


greenhat616 Nov 6, 2025
73c466a

fix: windows test build


greenhat616 Nov 6, 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


19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
 stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
 keep_alive: Option<Duration>,
 ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
 use futures::StreamExt;
 let stream = SseBody::new(stream.map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 }));
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
 let stream = SseBody::new(stream);

 let stream = match keep_alive {
 Some(duration) => stream
 .with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
 .boxed(),
 None => stream.boxed(),
 };

 Response::builder()
 .status(http::StatusCode::OK)
 .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down

27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
 pub sse_keep_alive: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.

 /// Cancellation token for the Streamable HTTP server.
 ///
 /// When this token is cancelled, all active sessions are terminated and
 /// the server stops accepting new requests.
 pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
 }
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 } else {
 // create standalone stream
 let stream = self
 .session_manager
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 }

Expand DownExpand Up@@ -307,7 +322,11 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_)
 | ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );

 response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_notification) => {
Expand Down

4 changes: 3 additions & 1 deletion
crates/rmcp/tests/test_with_js.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 .wait()
 .await?;

 let ct = CancellationToken::new();
 let service: StreamableHttpService<Calculator, LocalSessionManager> =
 StreamableHttpService::new(
 || Ok(Calculator::new()),
 Default::default(),
 StreamableHttpServerConfig {
 stateful_mode: true,
 sse_keep_alive: None,
 cancellation_token: ct.child_token(),
 },
 );
 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
 let ct = CancellationToken::new();

 let handle = tokio::spawn({
 let ct = ct.clone();
 async move {
Expand Down

13 changes: 10 additions & 3 deletions
examples/servers/src/counter_streamhttp.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
 StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
 layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
 )
 .with(tracing_subscriber::fmt::layer())
 .init();
 let ct = tokio_util::sync::CancellationToken::new();

 let service = StreamableHttpService::new(
 || Ok(Counter::new()),
 LocalSessionManager::default().into(),
 Default::default(),
 StreamableHttpServerConfig {
 cancellation_token: ct.child_token(),
 ..Default::default()
 },
 );

 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
 let _ = axum::serve(tcp_listener, router)
 .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
 .with_graceful_shutdown(async move {
 tokio::signal::ctrl_c().await.unwrap();
 ct.cancel();
 })
 .await;
 Ok(())
}

4 changes: 2 additions & 2 deletions
examples/transport/src/named-pipe.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
 let mut server = ServerOptions::new()
 .first_pipe_instance(true)
 .create(name)?;
 while let Ok(_) = server.connect().await {
 while server.connect().await.is_ok() {
 let stream = server;
 server = ServerOptions::new().create(name)?;
 tokio::spawn(async move {
 match serve_server(Calculator, stream).await {
 match serve_server(Calculator::new(), stream).await {
 Ok(server) => {
 println!("Server initialized successfully");
 if let Err(e) = server.waiting().await {
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
fix(streamable-http): gracefully shutdown while client connected by greenhat616 · Pull Request #494 · modelcontextprotocol/rust-sdk · GitHub
Skip to content
fix(streamable-http): gracefully shutdown while client connected#494

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

jokemanfire
merged 3 commits into

from

Dec 2, 2025
 Merged


fix(streamable-http): gracefully shutdown while client connected
#494
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
dc2e989

fix(streamable-http): gracefully shutdown while client connected


greenhat616 Oct 21, 2025
370b184

fix: adviced comments


greenhat616 Nov 6, 2025
73c466a

fix: windows test build


greenhat616 Nov 6, 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


19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
 stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
 keep_alive: Option<Duration>,
 ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
 use futures::StreamExt;
 let stream = SseBody::new(stream.map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 }));
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
 let stream = SseBody::new(stream);

 let stream = match keep_alive {
 Some(duration) => stream
 .with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
 .boxed(),
 None => stream.boxed(),
 };

 Response::builder()
 .status(http::StatusCode::OK)
 .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down

27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
 pub sse_keep_alive: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.

 /// Cancellation token for the Streamable HTTP server.
 ///
 /// When this token is cancelled, all active sessions are terminated and
 /// the server stops accepting new requests.
 pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
 }
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 } else {
 // create standalone stream
 let stream = self
 .session_manager
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 }

Expand DownExpand Up@@ -307,7 +322,11 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_)
 | ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );

 response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_notification) => {
Expand Down

4 changes: 3 additions & 1 deletion
crates/rmcp/tests/test_with_js.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 .wait()
 .await?;

 let ct = CancellationToken::new();
 let service: StreamableHttpService<Calculator, LocalSessionManager> =
 StreamableHttpService::new(
 || Ok(Calculator::new()),
 Default::default(),
 StreamableHttpServerConfig {
 stateful_mode: true,
 sse_keep_alive: None,
 cancellation_token: ct.child_token(),
 },
 );
 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
 let ct = CancellationToken::new();

 let handle = tokio::spawn({
 let ct = ct.clone();
 async move {
Expand Down

13 changes: 10 additions & 3 deletions
examples/servers/src/counter_streamhttp.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
 StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
 layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
 )
 .with(tracing_subscriber::fmt::layer())
 .init();
 let ct = tokio_util::sync::CancellationToken::new();

 let service = StreamableHttpService::new(
 || Ok(Counter::new()),
 LocalSessionManager::default().into(),
 Default::default(),
 StreamableHttpServerConfig {
 cancellation_token: ct.child_token(),
 ..Default::default()
 },
 );

 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
 let _ = axum::serve(tcp_listener, router)
 .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
 .with_graceful_shutdown(async move {
 tokio::signal::ctrl_c().await.unwrap();
 ct.cancel();
 })
 .await;
 Ok(())
}

4 changes: 2 additions & 2 deletions
examples/transport/src/named-pipe.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
 let mut server = ServerOptions::new()
 .first_pipe_instance(true)
 .create(name)?;
 while let Ok(_) = server.connect().await {
 while server.connect().await.is_ok() {
 let stream = server;
 server = ServerOptions::new().create(name)?;
 tokio::spawn(async move {
 match serve_server(Calculator, stream).await {
 match serve_server(Calculator::new(), stream).await {
 Ok(server) => {
 println!("Server initialized successfully");
 if let Err(e) = server.waiting().await {
Expand Down

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

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

jokemanfire
merged 3 commits into

from

Dec 2, 2025
 Merged


fix(streamable-http): gracefully shutdown while client connected
#494
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
dc2e989

fix(streamable-http): gracefully shutdown while client connected


greenhat616 Oct 21, 2025
370b184

fix: adviced comments


greenhat616 Nov 6, 2025
73c466a

fix: windows test build


greenhat616 Nov 6, 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


19 changes: 13 additions & 6 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@@ -6,6 +6,7 @@ use http::Response;
use http_body::Body;
use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
use sse_stream::{KeepAlive, Sse, SseBody};
use tokio_util::sync::CancellationToken;

use super::http_header::EVENT_STREAM_MIME_TYPE;
use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
Expand DownExpand Up@@ -65,20 +66,26 @@ pub struct ServerSseMessage {
pub(crate) fn sse_stream_response(
 stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
 keep_alive: Option<Duration>,
 ct: CancellationToken,
) -> Response<BoxBody<Bytes, Infallible>> {
 use futures::StreamExt;
 let stream = SseBody::new(stream.map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 }));
 let stream = stream
 .map(|message| {
 let data = serde_json::to_string(&message.message).expect("valid message");
 let mut sse = Sse::default().data(data);
 sse.id = message.event_id;
 Result::<Sse, Infallible>::Ok(sse)
 })
 .take_until(async move { ct.cancelled().await });
 let stream = SseBody::new(stream);

 let stream = match keep_alive {
 Some(duration) => stream
 .with_keep_alive::<TokioTimer>(KeepAlive::new().interval(duration))
 .boxed(),
 None => stream.boxed(),
 };

 Response::builder()
 .status(http::StatusCode::OK)
 .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE)
Expand Down

27 changes: 24 additions & 3 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@@ -6,6 +6,7 @@ use http::{Method, Request, Response, header::ALLOW};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;

use super::session::SessionManager;
use crate::{
Expand DownExpand Up@@ -33,13 +34,19 @@ pub struct StreamableHttpServerConfig {
 pub sse_keep_alive: Option<Duration>,
 /// If true, the server will create a session for each request and keep it alive.
 pub stateful_mode: bool,
Comment thread
greenhat616 marked this conversation as resolved.

 /// Cancellation token for the Streamable HTTP server.
 ///
 /// When this token is cancelled, all active sessions are terminated and
 /// the server stops accepting new requests.
 pub cancellation_token: CancellationToken,
}

impl Default for StreamableHttpServerConfig {
 fn default() -> Self {
 Self {
 sse_keep_alive: Some(Duration::from_secs(15)),
 stateful_mode: true,
 cancellation_token: CancellationToken::new(),
 }
 }
}
Expand DownExpand Up@@ -209,15 +216,23 @@ where
 .resume(&session_id, last_event_id)
 .await
 .map_err(internal_error_response("resume session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 } else {
 // create standalone stream
 let stream = self
 .session_manager
 .create_standalone_stream(&session_id)
 .await
 .map_err(internal_error_response("create standalone stream"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 }

Expand DownExpand Up@@ -307,7 +322,11 @@ where
 .create_stream(&session_id, message)
 .await
 .map_err(internal_error_response("get session"))?;
 Ok(sse_stream_response(stream, self.config.sse_keep_alive))
 Ok(sse_stream_response(
 stream,
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_)
 | ClientJsonRpcMessage::Response(_)
Expand DownExpand Up@@ -380,6 +399,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 );

 response.headers_mut().insert(
Expand DownExpand Up@@ -413,6 +433,7 @@ where
 }
 }),
 self.config.sse_keep_alive,
 self.config.cancellation_token.child_token(),
 ))
 }
 ClientJsonRpcMessage::Notification(_notification) => {
Expand Down

4 changes: 3 additions & 1 deletion
crates/rmcp/tests/test_with_js.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -94,18 +94,20 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 .wait()
 .await?;

 let ct = CancellationToken::new();
 let service: StreamableHttpService<Calculator, LocalSessionManager> =
 StreamableHttpService::new(
 || Ok(Calculator::new()),
 Default::default(),
 StreamableHttpServerConfig {
 stateful_mode: true,
 sse_keep_alive: None,
 cancellation_token: ct.child_token(),
 },
 );
 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(STREAMABLE_HTTP_BIND_ADDRESS).await?;
 let ct = CancellationToken::new();

 let handle = tokio::spawn({
 let ct = ct.clone();
 async move {
Expand Down

13 changes: 10 additions & 3 deletions
examples/servers/src/counter_streamhttp.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
use rmcp::transport::streamable_http_server::{
 StreamableHttpService, session::local::LocalSessionManager,
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tracing_subscriber::{
 layer::SubscriberExt,
Expand All@@ -20,17 +20,24 @@ async fn main() -> anyhow::Result<()> {
 )
 .with(tracing_subscriber::fmt::layer())
 .init();
 let ct = tokio_util::sync::CancellationToken::new();

 let service = StreamableHttpService::new(
 || Ok(Counter::new()),
 LocalSessionManager::default().into(),
 Default::default(),
 StreamableHttpServerConfig {
 cancellation_token: ct.child_token(),
 ..Default::default()
 },
 );

 let router = axum::Router::new().nest_service("/mcp", service);
 let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
 let _ = axum::serve(tcp_listener, router)
 .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
 .with_graceful_shutdown(async move {
 tokio::signal::ctrl_c().await.unwrap();
 ct.cancel();
 })
 .await;
 Ok(())
}

4 changes: 2 additions & 2 deletions
examples/transport/src/named-pipe.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,11 +12,11 @@ async fn main() -> anyhow::Result<()> {
 let mut server = ServerOptions::new()
 .first_pipe_instance(true)
 .create(name)?;
 while let Ok(_) = server.connect().await {
 while server.connect().await.is_ok() {
 let stream = server;
 server = ServerOptions::new().create(name)?;
 tokio::spawn(async move {
 match serve_server(Calculator, stream).await {
 match serve_server(Calculator::new(), stream).await {
 Ok(server) => {
 println!("Server initialized successfully");
 if let Err(e) = server.waiting().await {
Expand Down