feat(transport): support streamable http server (#152)

* feat(transport): support streamable http server

* refactor(transport): add common module for shared transport utilities

* perf(tracing): add more log for streamable http session

* fix(test): fix port conflict in test with js

* fix(transport): fix id generating and session management

1. cancel when deleting
2. use `wrapping` and `saturating` correctly

* chore(naming): rename sse server to streamable http server

Author: 4t145

README.md

@@ -199,6 +199,7 @@ See [examples](examples/README.md)
 - `transport-sse-server`: Server SSE transport
 - `transport-child-process`: Client stdio transport
 - `transport-sse`: Client sse transport
+- `transport-streamable-http-server` streamable http server transport
 
 ## Related Resources
 

crates/rmcp/Cargo.toml

@@ -48,10 +48,11 @@ tower-service = { version = "0.3", optional = true }
 # for ws transport
 # tokio-tungstenite ={ version = "0.26", optional = true }
 
-# for sse-server transport
+# for http-server transport
 axum = { version = "0.8", features = [], optional = true }
 rand = { version = "0.9", optional = true }
 tokio-stream = { version = "0.1", optional = true }
+uuid = { version = "1", features = ["v4"], optional = true }
 
 # macro
 rmcp-macros = { version = "0.1", workspace = true, optional = true }
@@ -72,6 +73,16 @@ transport-sse-server = [
     "dep:axum",
     "dep:rand",
     "dep:tokio-stream",
+    "uuid",
+]
+transport-streamable-http-server = [
+    "transport-streamable-http-server-session",
+    "dep:axum",
+    "uuid",
+]
+transport-streamable-http-server-session = [
+    "transport-async-rw",
+    "dep:tokio-stream",
 ]
 # transport-ws = ["transport-io", "dep:tokio-tungstenite"]
 tower = ["dep:tower-service"]
@@ -102,7 +113,7 @@ path = "tests/test_with_python.rs"
 
 [[test]]
 name = "test_with_js"
-required-features = ["server", "client", "transport-sse-server", "transport-child-process"]
+required-features = ["server", "client", "transport-sse-server", "transport-child-process", "transport-streamable-http-server"]
 path = "tests/test_with_js.rs"
 
 [[test]]

crates/rmcp/src/model.rs

@@ -186,7 +186,7 @@ impl<'de> Deserialize<'de> for NumberOrString {
 
 pub type RequestId = NumberOrString;
 
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Hash, Eq)]
 #[serde(transparent)]
 pub struct ProgressToken(pub NumberOrString);
 #[derive(Debug, Clone)]

crates/rmcp/src/service.rs

@@ -591,7 +591,7 @@ where
                 }
             };
 
-            tracing::debug!(?evt, "new event");
+            tracing::trace!(?evt, "new event");
             match evt {
                 // response and error
                 Event::ToSink(m) => {
@@ -657,7 +657,7 @@ where
                 Event::PeerMessage(JsonRpcMessage::Request(JsonRpcRequest {
                     id, request, ..
                 })) => {
-                    tracing::info!(%id, ?request, "received request");
+                    tracing::debug!(%id, ?request, "received request");
                     {
                         let service = shared_service.clone();
                         let sink = sink_proxy_tx.clone();
@@ -675,7 +675,7 @@ where
                             let result = service.handle_request(request, context).await;
                             let response = match result {
                                 Ok(result) => {
-                                    tracing::info!(%id, ?result, "response message");
+                                    tracing::debug!(%id, ?result, "response message");
                                     JsonRpcMessage::response(result, id)
                                 }
                                 Err(error) => {

crates/rmcp/src/transport.rs

@@ -76,6 +76,13 @@ pub use auth::{AuthError, AuthorizationManager, AuthorizationSession, Authorized
 
 // #[cfg(feature = "transport-ws")]
 // pub mod ws;
+#[cfg(feature = "transport-streamable-http-server-session")]
+pub mod streamable_http_server;
+#[cfg(feature = "transport-streamable-http-server")]
+pub use streamable_http_server::axum::StreamableHttpServer;
+
+/// Common use codes
+pub mod common;
 
 pub trait IntoTransport<R, E, A>: Send + 'static
 where

crates/rmcp/src/transport/common.rs

@@ -0,0 +1,5 @@
+#[cfg(any(
+    feature = "transport-streamable-http-server",
+    feature = "transport-sse-server"
+))]
+pub mod axum;

crates/rmcp/src/transport/common/axum.rs

@@ -0,0 +1,9 @@
+use std::{sync::Arc, time::Duration};
+
+pub type SessionId = Arc<str>;
+
+pub fn session_id() -> SessionId {
+    uuid::Uuid::new_v4().to_string().into()
+}
+
+pub const DEFAULT_AUTO_PING_INTERVAL: Duration = Duration::from_secs(15);

crates/rmcp/src/transport/sse_server.rs

@@ -19,14 +19,13 @@ use crate::{
     RoleServer, Service,
     model::ClientJsonRpcMessage,
     service::{RxJsonRpcMessage, TxJsonRpcMessage},
+    transport::common::axum::{DEFAULT_AUTO_PING_INTERVAL, SessionId, session_id},
 };
-type SessionId = Arc<str>;
+
 type TxStore =
     Arc<tokio::sync::RwLock<HashMap<SessionId, tokio::sync::mpsc::Sender<ClientJsonRpcMessage>>>>;
 pub type TransportReceiver = ReceiverStream<RxJsonRpcMessage<RoleServer>>;
 
-const DEFAULT_AUTO_PING_INTERVAL: Duration = Duration::from_secs(15);
-
 #[derive(Clone)]
 struct App {
     txs: TxStore,
@@ -56,11 +55,6 @@ impl App {
     }
 }
 
-fn session_id() -> SessionId {
-    let id = format!("{:016x}", rand::random::<u128>());
-    Arc::from(id)
-}
-
 #[derive(Debug, serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PostEventQuery {

crates/rmcp/src/transport/streamable_http_server.rs

@@ -0,0 +1,3 @@
+#[cfg(feature = "transport-streamable-http-server")]
+pub mod axum;
+pub mod session;