diff --git a/backend/windmill-api/src/mcp/core.rs b/backend/windmill-api/src/mcp/core.rs index 86fbec1806..472c7bbec9 100644 --- a/backend/windmill-api/src/mcp/core.rs +++ b/backend/windmill-api/src/mcp/core.rs @@ -386,6 +386,12 @@ impl McpBackend for WindmillBackend { }) } + async fn runnable_list_fingerprint(&self, workspace_id: &str) -> BackendResult { + windmill_common::runnables::runnable_list_fingerprint(&self.db, workspace_id) + .await + .map_err(|e| ErrorData::internal_error(e.to_string(), None)) + } + fn all_endpoint_tools(&self) -> Vec { all_tools() } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 3e74d8e17f..1015948882 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -118,6 +118,7 @@ pub mod queue; pub mod queue_metrics; pub mod result_stream; pub mod runnable_settings; +pub mod runnables; pub mod schedule; pub mod schema; pub mod scripts; diff --git a/backend/windmill-common/src/runnables.rs b/backend/windmill-common/src/runnables.rs new file mode 100644 index 0000000000..579e5e248e --- /dev/null +++ b/backend/windmill-common/src/runnables.rs @@ -0,0 +1,22 @@ +//! Queries spanning both scripts and flows. + +use crate::{error::Error, DB}; + +/// An opaque value that changes whenever a workspace's live scripts or flows are added, +/// removed, moved or redeployed: a redeploy is a new script hash or a new last flow +/// version, so schema and description changes register too. Not filtered by caller, +/// so it reveals nothing beyond whether it changed. +pub async fn runnable_list_fingerprint(db: &DB, workspace_id: &str) -> Result { + Ok(sqlx::query_scalar::<_, String>( + "SELECT COALESCE(md5(string_agg(k, ',' ORDER BY k)), '') FROM ( + SELECT 's' || path || ':' || hash FROM script + WHERE workspace_id = $1 AND archived = false + UNION ALL + SELECT 'f' || path || ':' || COALESCE(versions[array_upper(versions, 1)], 0) FROM flow + WHERE workspace_id = $1 AND archived = false + ) t(k)", + ) + .bind(workspace_id) + .fetch_one(db) + .await?) +} diff --git a/backend/windmill-common/tests/runnable_list_fingerprint.rs b/backend/windmill-common/tests/runnable_list_fingerprint.rs new file mode 100644 index 0000000000..d7438f0008 --- /dev/null +++ b/backend/windmill-common/tests/runnable_list_fingerprint.rs @@ -0,0 +1,54 @@ +use sqlx::{Pool, Postgres}; +use windmill_common::runnables::runnable_list_fingerprint; + +/// MCP clients refresh their tool list when this fingerprint moves, so it must move on every +/// change to what the listing shows and stay put on writes it does not show. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn fingerprint_tracks_the_live_runnable_list(db: Pool) { + let fp = || async { + runnable_list_fingerprint(&db, "test-workspace") + .await + .unwrap() + }; + let exec = |sql: &'static str| { + let db = db.clone(); + async move { sqlx::query(sql).execute(&db).await.unwrap() } + }; + + let mut last = fp().await; + let mut assert_moved = async |what: &str, moved: bool| { + let now = fp().await; + assert_eq!(now != last, moved, "{what}"); + last = now; + }; + + exec( + "INSERT INTO script (workspace_id, hash, path, summary, description, content, created_by, language, kind) + VALUES ('test-workspace', 1, 'f/test/a', '', '', 'def main(): pass', 'test-user', 'python3', 'script')", + ) + .await; + assert_moved("deploy", true).await; + + exec("UPDATE script SET lock = '' WHERE hash = 1").await; + assert_moved("lock update", false).await; + + exec("UPDATE script SET path = 'f/test/b' WHERE hash = 1").await; + assert_moved("path move", true).await; + + exec("UPDATE script SET archived = true WHERE hash = 1").await; + assert_moved("archive", true).await; + + exec( + "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, schema, versions) + VALUES ('test-workspace', 'f/test/flow', '', '', '{}'::jsonb, 'test-user', '{}'::json, ARRAY[1]::bigint[])", + ) + .await; + assert_moved("flow create", true).await; + + exec("UPDATE flow SET versions = array_append(versions, 2::bigint) WHERE path = 'f/test/flow'") + .await; + assert_moved("flow redeploy", true).await; + + exec("DELETE FROM flow WHERE path = 'f/test/flow'").await; + assert_moved("flow delete", true).await; +} diff --git a/backend/windmill-mcp/src/server/backend.rs b/backend/windmill-mcp/src/server/backend.rs index 7ca1eb08a1..4801bfaa99 100644 --- a/backend/windmill-mcp/src/server/backend.rs +++ b/backend/windmill-mcp/src/server/backend.rs @@ -208,6 +208,11 @@ pub trait McpBackend: Send + Sync + Clone + 'static { workspace_id: &str, ) -> BackendResult; + /// An opaque value that changes whenever the workspace's live scripts or flows are + /// added, removed, moved or redeployed, compared across polls to detect tool-list + /// changes. + async fn runnable_list_fingerprint(&self, workspace_id: &str) -> BackendResult; + // ───────────────────────────────────────────────────────────────── // Endpoint Tools // ───────────────────────────────────────────────────────────────── diff --git a/backend/windmill-mcp/src/server/list_changes.rs b/backend/windmill-mcp/src/server/list_changes.rs new file mode 100644 index 0000000000..8470cff630 --- /dev/null +++ b/backend/windmill-mcp/src/server/list_changes.rs @@ -0,0 +1,104 @@ +//! Detects changes to a workspace's scripts and flows for `subscriptions/listen`. +//! +//! Each process polls a fingerprint of every workspace one of its listeners watches, +//! so a change made on any replica, through any path (UI, CLI, git sync, a user +//! rename), reaches every listener within one poll interval with no cross-replica +//! signalling. + +use crate::server::backend::McpBackend; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::sync::{broadcast, Notify}; + +const POLL_INTERVAL: Duration = Duration::from_secs(30); + +struct Watcher { + changes: broadcast::Sender<()>, + poll_now: Arc, +} + +#[derive(Default)] +pub(crate) struct ListChanges { + watchers: Mutex>, +} + +impl ListChanges { + /// Subscribe to `workspace_id`'s changes, starting its poller if this is the + /// first listener for it in this process. + pub(crate) fn subscribe( + self: &Arc, + backend: &Arc, + workspace_id: &str, + ) -> broadcast::Receiver<()> { + let mut watchers = self.watchers.lock().unwrap(); + if let Some(watcher) = watchers.get(workspace_id) { + return watcher.changes.subscribe(); + } + let (changes, rx) = broadcast::channel(16); + let poll_now = Arc::new(Notify::new()); + watchers.insert( + workspace_id.to_string(), + Watcher { changes: changes.clone(), poll_now: poll_now.clone() }, + ); + tokio::spawn(self.clone().poll( + backend.clone(), + workspace_id.to_string(), + changes, + poll_now, + )); + rx + } + + /// Poll `workspace_id` now rather than at its next interval, after this process + /// changed it: the client that made the change is the one waiting on it. Polling + /// rather than signalling directly keeps the poller's baseline current, so the + /// same change is not reported again on the next tick. + pub(crate) fn poll_now(&self, workspace_id: &str) { + if let Some(watcher) = self.watchers.lock().unwrap().get(workspace_id) { + watcher.poll_now.notify_one(); + } + } + + async fn poll( + self: Arc, + backend: Arc, + workspace_id: String, + changes: broadcast::Sender<()>, + poll_now: Arc, + ) { + let mut interval = tokio::time::interval(POLL_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut last = None; + loop { + tokio::select! { + _ = interval.tick() => {} + _ = poll_now.notified() => {} + } + { + // Checked under the lock `subscribe` takes, so a listener cannot join a + // watcher that is about to stop. + let mut watchers = self.watchers.lock().unwrap(); + if changes.receiver_count() == 0 { + watchers.remove(&workspace_id); + return; + } + } + match backend.runnable_list_fingerprint(&workspace_id).await { + Ok(fingerprint) => { + // The first fingerprint also notifies: the listener that started this + // watcher has already sent its opening notification, and a change landing + // before this baseline would otherwise never be announced. + if last.as_ref() != Some(&fingerprint) { + let _ = changes.send(()); + } + last = Some(fingerprint); + } + Err(e) => tracing::warn!( + "MCP tool-list poll failed for workspace {workspace_id}: {}", + e.message + ), + } + } + } +} diff --git a/backend/windmill-mcp/src/server/mod.rs b/backend/windmill-mcp/src/server/mod.rs index b97e374e98..ad7c14d320 100644 --- a/backend/windmill-mcp/src/server/mod.rs +++ b/backend/windmill-mcp/src/server/mod.rs @@ -7,6 +7,7 @@ pub mod backend; pub mod endpoints; +mod list_changes; pub mod runner; pub mod tools; diff --git a/backend/windmill-mcp/src/server/runner.rs b/backend/windmill-mcp/src/server/runner.rs index e27eaa8470..177989a82e 100644 --- a/backend/windmill-mcp/src/server/runner.rs +++ b/backend/windmill-mcp/src/server/runner.rs @@ -16,20 +16,22 @@ use crate::server::backend::{McpAuth, McpBackend, McpRequest, PathFilter}; use crate::server::endpoints::{ endpoint_tool_to_mcp_tool, endpoint_tool_to_mcp_tool_multi, list_workspaces_tool, EndpointTool, }; +use crate::server::list_changes::ListChanges; use crate::server::tools::create_tool_from_item; use rmcp::handler::server::ServerHandler; use rmcp::model::{ CacheScope, CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, - Implementation, InitializeResult, ListPromptsResult, ListResourceTemplatesResult, - ListResourcesResult, ListToolsResult, PaginatedRequestParams, ProtocolVersion, - ServerCapabilities, ServerInfo, + Implementation, InitializeRequestParams, InitializeResult, ListPromptsResult, + ListResourceTemplatesResult, ListResourcesResult, ListToolsResult, PaginatedRequestParams, + ProtocolVersion, ServerCapabilities, ServerInfo, SubscriptionFilter, }; -use rmcp::service::{RequestContext, RoleServer}; +use rmcp::service::{RequestContext, RoleServer, SubscriptionContext}; use rmcp::ErrorData; use serde_json::Value; use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use tokio::sync::broadcast; /// Protocol revisions this server is willing to speak. `2026-07-28` is served /// statelessly with per-request metadata; the older revisions keep the @@ -46,15 +48,21 @@ const SUPPORTED_PROTOCOL_VERSIONS: &[ProtocolVersion] = &[ /// leaves them unset, and a strict client rejects the response without them. /// /// Zero because nothing here is cacheable: the listing is rebuilt from the -/// workspace's scripts and flows, which change at any time, and this server -/// advertises no `listChanged` capability, so a client that cached a stale list -/// would have no way to learn it had gone stale. +/// workspace's scripts and flows, which change at any time, and `listChanged` +/// covers only some of what the listing depends on (not resources, favorites or +/// permissions), lags by up to a poll interval, and reaches only clients that hold a +/// `subscriptions/listen` stream. const LIST_TTL_MS: u64 = 0; + /// Every listing is filtered by the caller's token scopes and workspace /// membership, so no two callers necessarily see the same tools — a shared /// cache entry would leak one token's view to another. const LIST_CACHE_SCOPE: CacheScope = CacheScope::Private; +/// How long a `subscriptions/listen` stream waits after a change before notifying, +/// collecting the rest of the burst into the same notification. +const LIST_CHANGE_COALESCE: std::time::Duration = std::time::Duration::from_millis(500); + // Re-export from http crate for extracting request parts use http::request::Parts as HttpParts; @@ -83,11 +91,12 @@ fn truncate_tool_result(text: String) -> String { /// to perform the actual operations (database queries, job execution, etc.) pub struct Runner { backend: Arc, + list_changes: Arc, } impl Clone for Runner { fn clone(&self) -> Self { - Self { backend: self.backend.clone() } + Self { backend: self.backend.clone(), list_changes: self.list_changes.clone() } } } @@ -113,7 +122,7 @@ struct McpContext { impl Runner { /// Create a new Runner with the given backend pub fn new(backend: B) -> Self { - Self { backend: Arc::new(backend) } + Self { backend: Arc::new(backend), list_changes: Arc::default() } } /// Extract authentication, the workspace mode and the HTTP request itself @@ -232,6 +241,22 @@ fn endpoint_path_policy(endpoint_name: &str) -> Option { } } +/// Endpoint tools that change which scripts/flows a workspace exposes, or their +/// schemas. The process serving the call polls the workspace right away rather than +/// at its next interval, so a client that just deployed a script can call it. +fn changes_runnable_list(endpoint_name: &str) -> bool { + matches!( + endpoint_name, + "createScript" + | "updateScript" + | "deleteScriptByPath" + | "deleteScriptByHash" + | "createFlow" + | "updateFlow" + | "deleteFlowByPath" + ) +} + /// Whether the token restricts `kind` ("script"/"flow") to specific paths. A /// `*` pattern grants every path (see `is_resource_allowed`), so it does not /// count as confinement. @@ -428,12 +453,17 @@ impl ServerHandler for Runner { .with_title("Windmill") .with_website_url("https://windmill.dev"); - InitializeResult::new(ServerCapabilities::builder().enable_tools().build()) - .with_server_info(server_info) - .with_instructions( - "This server provides a list of scripts and flows the user can run on Windmill. \ + InitializeResult::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .build(), + ) + .with_server_info(server_info) + .with_instructions( + "This server provides a list of scripts and flows the user can run on Windmill. \ Each flow and script is a tool callable with their respective arguments.", - ) + ) } /// Pinned rather than left to rmcp's default (every version the SDK knows), so a @@ -444,6 +474,84 @@ impl ServerHandler for Runner { Cow::Borrowed(SUPPORTED_PROTOCOL_VERSIONS) } + /// Only clients below `2026-07-28` send `initialize`, and those receive server + /// notifications on a session's standalone stream. The transport is sessionless, so + /// there is none: `listChanged` is withheld from them rather than promised and + /// never delivered. `2026-07-28` clients learn it from `server/discover` and + /// receive it on `subscriptions/listen` (see `listen`). + async fn initialize( + &self, + request: InitializeRequestParams, + context: RequestContext, + ) -> Result { + context.peer.set_peer_info(request.clone()); + let mut info = self.get_info(); + if let Some(tools) = info.capabilities.tools.as_mut() { + tools.list_changed = None; + } + if SUPPORTED_PROTOCOL_VERSIONS.contains(&request.protocol_version) { + info.protocol_version = request.protocol_version; + } + Ok(info) + } + + /// `None` for a request that asks for nothing this server sends, so the client is + /// told the subscription is unsupported rather than acked with an empty filter. + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + (requested.tools_list_changed == Some(true)) + .then(|| SubscriptionFilter::builder().tools_list_changed().build()) + } + + async fn listen(&self, subscription: SubscriptionContext) -> Result<(), ErrorData> { + let McpContext { auth, mode, .. } = Self::extract_context(subscription.request_context())?; + // Multi-workspace mode and read-only tokens list no scripts or flows, so their + // tool list never changes. + let workspace_id = match mode { + McpMode::Single(workspace_id) if !auth.read_only() => workspace_id, + _ => { + subscription.cancelled().await; + return Ok(()); + } + }; + + let mut changes = self.list_changes.subscribe(&self.backend, &workspace_id); + // Every stream opens with one notification: a change the watcher recorded between + // this client's `tools/list` and this subscription is never broadcast again. + let mut opening = true; + loop { + if !std::mem::take(&mut opening) { + tokio::select! { + _ = subscription.cancelled() => return Ok(()), + change = changes.recv() => match change { + Ok(()) | Err(broadcast::error::RecvError::Lagged(_)) => {} + Err(broadcast::error::RecvError::Closed) => return Ok(()), + }, + } + } + // A client creating several scripts in a row: collapse the burst into one + // notification, so it re-lists once. + tokio::select! { + _ = subscription.cancelled() => return Ok(()), + _ = tokio::time::sleep(LIST_CHANGE_COALESCE) => {} + } + while !matches!( + changes.try_recv(), + Err(broadcast::error::TryRecvError::Empty | broadcast::error::TryRecvError::Closed) + ) {} + if subscription + .sink() + .notify_tool_list_changed() + .await + .is_err() + { + return Ok(()); + } + } + } + async fn list_tools( &self, _request: Option, @@ -714,6 +822,9 @@ impl Runner { .backend .call_endpoint(auth, workspace_id, endpoint_tool, args) .await?; + if changes_runnable_list(&endpoint_tool.name) { + self.list_changes.poll_now(workspace_id); + } return Ok(CallToolResult::success(vec![ContentBlock::text( truncate_tool_result( @@ -969,6 +1080,9 @@ impl Runner { .backend .call_endpoint(&resolved_auth, &workspace_id, endpoint_tool, args) .await?; + if changes_runnable_list(&endpoint_tool.name) { + self.list_changes.poll_now(&workspace_id); + } Ok(CallToolResult::success(vec![ContentBlock::text( truncate_tool_result(