feat: integrate list root and client info into hyper runtime

Author: hashemix

crates/rust-mcp-sdk/src/hyper_servers/hyper_runtime.rs

@@ -4,7 +4,8 @@ use crate::{
     mcp_server::HyperServer,
     schema::{
         schema_utils::{NotificationFromServer, RequestFromServer, ResultFromClient},
-        CreateMessageRequestParams, CreateMessageResult, LoggingMessageNotificationParams,
+        CreateMessageRequestParams, CreateMessageResult, InitializeRequestParams,
+        ListRootsRequestParams, ListRootsResult, LoggingMessageNotificationParams,
         PromptListChangedNotificationParams, ResourceListChangedNotificationParams,
         ResourceUpdatedNotificationParams, ToolListChangedNotificationParams,
     },
@@ -99,6 +100,21 @@ impl HyperRuntime {
         runtime.send_notification(notification).await
     }
 
+    /// Request a list of root URIs from the client. Roots allow
+    /// servers to ask for specific directories or files to operate on. A common example
+    /// for roots is providing a set of repositories or directories a server should operate on.
+    /// This request is typically used when the server needs to understand the file system
+    /// structure or access specific locations that the client has permission to read from
+    pub async fn list_roots(
+        &self,
+        session_id: &SessionId,
+        params: Option<ListRootsRequestParams>,
+    ) -> SdkResult<ListRootsResult> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.list_roots(params).await
+    }
+
     pub async fn send_logging_message(
         &self,
         session_id: &SessionId,
@@ -195,4 +211,13 @@ impl HyperRuntime {
         let runtime = runtime.lock().await.to_owned();
         runtime.create_message(params).await
     }
+
+    pub async fn client_info(
+        &self,
+        session_id: &SessionId,
+    ) -> SdkResult<Option<InitializeRequestParams>> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        Ok(runtime.client_info())
+    }
 }

crates/rust-mcp-sdk/src/mcp_runtimes/server_runtime.rs

@@ -87,17 +87,9 @@ impl McpServer for ServerRuntime {
                 .with_message("transport stream does not exists or is closed!".to_string()),
         )?;
 
-        // generate a new request_id for request messages
-        let outgoing_request_id = if message.is_request() {
-            match request_id {
-                Some(_) => Err(RpcError::internal_error().with_message(
-                    "request_id should not have a value when sending a new request".to_string(),
-                )),
-                None => Ok(self.next_request_id(transport).await),
-            }
-        } else {
-            Ok(request_id)
-        }?;
+        let outgoing_request_id = self
+            .request_id_for_message(transport, &message, request_id)
+            .await;
 
         let mcp_message = ServerMessage::from_message(message, outgoing_request_id)?;
         transport
@@ -225,7 +217,18 @@ impl ServerRuntime {
         Ok(())
     }
 
-    pub(crate) async fn next_request_id(
+    /// Determines the request ID for an outgoing MCP message.
+    ///
+    /// For requests, generates a new ID using the internal counter. For responses or errors,
+    /// uses the provided `request_id`. Notifications receive no ID.
+    ///
+    /// # Arguments
+    /// * `message` - The MCP message to evaluate.
+    /// * `request_id` - An optional existing request ID (required for responses/errors).
+    ///
+    /// # Returns
+    /// An `Option<RequestId>`: `Some` for requests or responses/errors, `None` for notifications.
+    pub(crate) async fn request_id_for_message(
         &self,
         transport: &Arc<
             dyn TransportDispatcher<
@@ -236,12 +239,16 @@ impl ServerRuntime {
                 ServerMessage,
             >,
         >,
+        message: &MessageFromServer,
+        request_id: Option<RequestId>,
     ) -> Option<RequestId> {
         let message_sender = transport.message_sender();
         let guard = message_sender.read().await;
-        guard
-            .as_ref()
-            .map(|dispatcher| dispatcher.next_request_id())
+        if let Some(dispatcher) = guard.as_ref() {
+            dispatcher.request_id_for_message(message, request_id)
+        } else {
+            None
+        }
     }
 
     pub(crate) async fn handle_message(

crates/rust-mcp-sdk/tests/test_streamable_http.rs

@@ -3,13 +3,14 @@ use std::{collections::HashMap, error::Error, sync::Arc, time::Duration, vec};
 use hyper::StatusCode;
 use rust_mcp_schema::{
     schema_utils::{
-        ClientJsonrpcRequest, ClientMessage, ClientMessages, FromMessage, NotificationFromServer,
-        ResultFromServer, RpcMessage, SdkError, SdkErrorCodes, ServerJsonrpcNotification,
-        ServerJsonrpcResponse, ServerMessages,
+        ClientJsonrpcRequest, ClientJsonrpcResponse, ClientMessage, ClientMessages, FromMessage,
+        NotificationFromServer, RequestFromServer, ResultFromServer, RpcMessage, SdkError,
+        SdkErrorCodes, ServerJsonrpcNotification, ServerJsonrpcRequest, ServerJsonrpcResponse,
+        ServerMessages,
     },
-    CallToolRequest, CallToolRequestParams, ListToolsRequest, LoggingLevel,
-    LoggingMessageNotificationParams, RequestId, RootsListChangedNotification, ServerNotification,
-    ServerResult,
+    CallToolRequest, CallToolRequestParams, ListPromptsRequestParams, ListRootsRequestParams,
+    ListRootsResult, ListToolsRequest, LoggingLevel, LoggingMessageNotificationParams, RequestId,
+    RootsListChangedNotification, ServerNotification, ServerRequest, ServerResult,
 };
 use rust_mcp_sdk::mcp_server::HyperServerOptions;
 use serde_json::{json, Map, Value};
@@ -364,6 +365,80 @@ async fn should_establish_standalone_stream_and_receive_server_messages() {
     server.hyper_runtime.await_server().await.unwrap()
 }
 
+// should establish standalone SSE stream and receive server-initiated messages
+#[tokio::test]
+async fn should_establish_standalone_stream_and_receive_server_requests() {
+    let (server, session_id) = initialize_server(None).await.unwrap();
+    let response = get_standalone_stream(&server.streamable_url, &session_id).await;
+
+    assert_eq!(response.status(), StatusCode::OK);
+
+    assert_eq!(
+        response
+            .headers()
+            .get("mcp-session-id")
+            .unwrap()
+            .to_str()
+            .unwrap(),
+        session_id
+    );
+
+    assert_eq!(
+        response
+            .headers()
+            .get("content-type")
+            .unwrap()
+            .to_str()
+            .unwrap(),
+        "text/event-stream"
+    );
+
+    let hyper_server = Arc::new(server.hyper_runtime);
+    let hyper_server_clone = hyper_server.clone();
+    let session_id_clone = session_id.to_string();
+
+    tokio::spawn(async move {
+        // Send a server-initiated notification that should appear on SSE stream with a valid request_id
+        hyper_server_clone
+            .list_roots(&session_id_clone, None)
+            .await
+            .unwrap();
+    });
+
+    tokio::time::sleep(Duration::from_millis(2250)).await;
+
+    let json_rpc_message: ClientJsonrpcResponse = ClientJsonrpcResponse::new(
+        RequestId::Integer(0),
+        ListRootsResult {
+            meta: None,
+            roots: vec![],
+        }
+        .into(),
+    );
+
+    send_post_request(
+        &server.streamable_url,
+        &serde_json::to_string(&json_rpc_message).unwrap(),
+        Some(&session_id),
+        None,
+    )
+    .await
+    .expect("Request failed");
+
+    let event = read_sse_event(response).await.unwrap();
+
+    let message: ServerJsonrpcRequest = serde_json::from_str(&event).unwrap();
+
+    println!(">>> message {:?} ", message);
+
+    let RequestFromServer::ServerRequest(ServerRequest::ListRootsRequest(_)) = message.request
+    else {
+        panic!("invalid message received!");
+    };
+
+    hyper_server.graceful_shutdown(ONE_MILLISECOND);
+}
+
 // should not close GET SSE stream after sending multiple server notifications
 #[tokio::test]
 async fn should_not_close_get_sse_stream() {