upgrade rmcp crate (#7518)

Author: centdix

backend/Cargo.lock

@@ -40,7 +40,7 @@ mcp = ["dep:rmcp"]
 python = []
 
 [dependencies]
-rmcp = { version = "0.8.1", features=["transport-streamable-http-server", "transport-streamable-http-server-session", "transport-worker"], optional = true }
+rmcp = { version = "0.12.0", features=["transport-streamable-http-server", "transport-streamable-http-server-session", "transport-worker"], optional = true }
 windmill-queue.workspace = true
 windmill-common = { workspace = true, default-features = false }
 windmill-audit.workspace = true

backend/windmill-api/Cargo.toml

@@ -19,10 +19,8 @@ use crate::oauth2_oss::SlackVerifier;
 use crate::smtp_server_oss::SmtpServer;
 
 #[cfg(feature = "mcp")]
-use crate::mcp::{extract_and_store_workspace_id, setup_mcp_server, shutdown_mcp_server};
+use crate::mcp::{extract_and_store_workspace_id, setup_mcp_server};
 use crate::triggers::start_all_listeners;
-#[cfg(feature = "mcp")]
-use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
 use tower_http::catch_panic::CatchPanicLayer;
 
 use crate::tracing_init::MyOnFailure;
@@ -400,14 +398,17 @@ pub async fn run_server(
 
     // Setup MCP server
     #[allow(unused_variables)]
-    let (mcp_router, mcp_session_manager) = {
+    let (mcp_router, mcp_cancellation_token) = {
         #[cfg(feature = "mcp")]
         if server_mode || mcp_mode {
-            let (mcp_router, mcp_session_manager) = setup_mcp_server().await?;
+            let (mcp_router, mcp_cancellation_token) = setup_mcp_server().await?;
             let mcp_middleware = axum::middleware::from_fn(extract_and_store_workspace_id);
-            (mcp_router.layer(mcp_middleware), Some(mcp_session_manager))
+            (
+                mcp_router.layer(mcp_middleware),
+                Some(mcp_cancellation_token),
+            )
         } else {
-            (Router::new(), Option::<Arc<LocalSessionManager>>::None)
+            (Router::new(), None)
         }
 
         #[cfg(not(feature = "mcp"))]
@@ -757,8 +758,8 @@ pub async fn run_server(
         tracing::info!("Graceful shutdown of server");
 
         #[cfg(feature = "mcp")]
-        if let Some(mcp_session_manager) = mcp_session_manager {
-            shutdown_mcp_server(mcp_session_manager).await;
+        if let Some(mcp_cancellation_token) = mcp_cancellation_token {
+            mcp_cancellation_token.cancel();
             tracing::info!("MCP server shutdown");
         }
     });

backend/windmill-api/src/lib.rs

@@ -8,4 +8,4 @@ pub mod tools;
 pub mod utils;
 
 // Re-export main components
-pub use server::{extract_and_store_workspace_id, setup_mcp_server, shutdown_mcp_server, list_tools_service};
+pub use server::{extract_and_store_workspace_id, list_tools_service, setup_mcp_server};

backend/windmill-api/src/mcp/mod.rs

@@ -18,6 +18,7 @@ use rmcp::{
 };
 use serde_json::Value;
 use tokio::try_join;
+use tokio_util::sync::CancellationToken;
 use windmill_common::db::UserDB;
 use windmill_common::worker::to_raw_value;
 use windmill_common::{utils::StripPath, DB};
@@ -47,7 +48,7 @@ use axum::{
     extract::Path, http::Request, middleware::Next, response::Response, routing::get, Json, Router,
 };
 use rmcp::transport::streamable_http_server::{
-    session::local::LocalSessionManager, SessionManager, StreamableHttpService,
+    session::local::LocalSessionManager, StreamableHttpService,
 };
 use windmill_common::error::JsonResult;
 
@@ -126,6 +127,7 @@ impl Runner {
                 idempotent_hint: Some(false), // Are not guaranteed to be idempotent
                 open_world_hint: Some(true),  // Can interact with external services
             }),
+            meta: None,
         })
     }
 }
@@ -471,7 +473,7 @@ impl ServerHandler for Runner {
             );
         }
 
-        Ok(ListToolsResult { tools, next_cursor: None })
+        Ok(ListToolsResult { tools, next_cursor: None, meta: None })
     }
 
     fn get_info(&self) -> ServerInfo {
@@ -498,7 +500,7 @@ impl ServerHandler for Runner {
         _request: Option<PaginatedRequestParam>,
         _context: RequestContext<RoleServer>,
     ) -> Result<ListResourcesResult, ErrorData> {
-        Ok(ListResourcesResult { resources: vec![], next_cursor: None })
+        Ok(ListResourcesResult { resources: vec![], next_cursor: None, meta: None })
     }
 
     async fn list_prompts(
@@ -530,11 +532,13 @@ pub async fn extract_and_store_workspace_id(
 }
 
 /// Setup the MCP server with HTTP transport
-pub async fn setup_mcp_server() -> anyhow::Result<(Router, Arc<LocalSessionManager>)> {
+pub async fn setup_mcp_server() -> anyhow::Result<(Router, CancellationToken)> {
+    let cancellation_token = CancellationToken::new();
     let session_manager = Arc::new(LocalSessionManager::default());
     let service_config = StreamableHttpServerConfig {
         sse_keep_alive: Some(Duration::from_secs(15)),
         stateful_mode: false,
+        cancellation_token: cancellation_token.clone(),
     };
     let service = StreamableHttpService::new(
         || Ok(Runner::new()),
@@ -543,34 +547,7 @@ pub async fn setup_mcp_server() -> anyhow::Result<(Router, Arc<LocalSessionManag
     );
 
     let router = axum::Router::new().nest_service("/", service);
-    Ok((router, session_manager))
-}
-
-/// Shutdown the MCP server gracefully by closing all active sessions
-pub async fn shutdown_mcp_server(session_manager: Arc<LocalSessionManager>) {
-    let session_ids_to_close = {
-        let sessions_map = session_manager.sessions.read().await;
-        sessions_map.keys().cloned().collect::<Vec<_>>()
-    };
-
-    if !session_ids_to_close.is_empty() {
-        tracing::info!(
-            "Closing {} active MCP session(s)...",
-            session_ids_to_close.len()
-        );
-        let close_futures = session_ids_to_close
-            .iter()
-            .map(|session_id| {
-                let manager_clone = session_manager.clone();
-                async move {
-                    if let Err(_) = manager_clone.close_session(session_id).await {
-                        tracing::warn!("Error closing MCP session");
-                    }
-                }
-            })
-            .collect::<Vec<_>>();
-        futures::future::join_all(close_futures).await;
-    }
+    Ok((router, cancellation_token))
 }
 
 /// HTTP handler to list MCP tools as JSON

backend/windmill-api/src/mcp/server.rs

@@ -61,6 +61,7 @@ pub fn endpoint_tool_to_mcp_tool(tool: &EndpointTool) -> Tool {
         output_schema: None,
         icons: None,
         annotations: Some(annotations),
+        meta: None,
     }
 }
 

backend/windmill-api/src/mcp/tools/endpoint_tools.rs

@@ -108,7 +108,7 @@ tempfile.workspace = true
 systemstat.workspace = true
 size.workspace = true
 globset.workspace = true
-rmcp = { version = "0.8.1", features = ["client", "transport-streamable-http-client", "transport-streamable-http-client-reqwest"] }
+rmcp = { version = "0.12.0", features = ["client", "transport-streamable-http-client", "transport-streamable-http-client-reqwest"] }
 
 opentelemetry-semantic-conventions = { workspace = true, optional = true }
 opentelemetry-otlp = { workspace = true, optional = true }

backend/windmill-common/Cargo.toml

@@ -5,7 +5,7 @@ use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
 use serde_json::{json, Value};
 use std::str::FromStr;
 
-use rmcp::model::Tool as McpTool;
+pub use rmcp::model::Tool as McpTool;
 use rmcp::{
     model::{
         CallToolRequestParam, ClientCapabilities, ClientInfo, Implementation,

backend/windmill-common/src/mcp_client.rs

@@ -57,7 +57,6 @@ windmill-parser-sql.workspace = true
 windmill-parser-graphql.workspace = true
 windmill-parser-php = { workspace = true, optional = true }
 windmill-git-sync.workspace = true
-rmcp = { version = "0.8.1", features = ["client", "transport-streamable-http-client", "transport-streamable-http-client-reqwest"] }
 aws-sdk-bedrockruntime.workspace = true
 aws-config.workspace = true
 aws-credential-types.workspace = true

backend/windmill-worker/Cargo.toml

@@ -20,7 +20,7 @@ use windmill_common::{
 };
 use windmill_common::{
     flows::FlowModuleValue,
-    mcp_client::{McpClient, McpResource, McpToolSource},
+    mcp_client::{McpClient, McpResource, McpTool, McpToolSource},
 };
 use windmill_queue::{flow_status::get_step_of_flow_status, MiniPulledJob};
 
@@ -352,7 +352,7 @@ pub async fn cleanup_mcp_clients(mcp_clients: HashMap<String, Arc<McpClient>>) {
 
 /// Convert raw MCP tools to Windmill Tool format with source tracking
 fn convert_mcp_tools_to_windmill_tools(
-    mcp_tools: &[rmcp::model::Tool],
+    mcp_tools: &[McpTool],
     resource_name: &str,
     resource_path: &str,
 ) -> Result<Vec<Tool>, Error> {