feat: notify MCP clients when their workspace's tool list changes

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

backend/migrations/20260924065755_mcp_tools_change_notify.down.sql

@@ -0,0 +1,7 @@
+DROP TRIGGER IF EXISTS script_mcp_tools_update_trigger ON script;
+DROP TRIGGER IF EXISTS script_mcp_tools_delete_trigger ON script;
+DROP TRIGGER IF EXISTS flow_mcp_tools_update_trigger ON flow;
+DROP TRIGGER IF EXISTS flow_mcp_tools_delete_trigger ON flow;
+DROP TRIGGER IF EXISTS favorite_mcp_tools_insert_trigger ON favorite;
+DROP TRIGGER IF EXISTS favorite_mcp_tools_delete_trigger ON favorite;
+DROP FUNCTION IF EXISTS notify_mcp_tools_change();

backend/migrations/20260924065755_mcp_tools_change_notify.up.sql

@@ -0,0 +1,52 @@
+-- Deploys already emit `notify_runnable_version_change`; these cover the other changes to the
+-- scripts and flows the MCP server lists as tools. One event per workspace per transaction:
+-- archiving a path touches every version of it and deleting a workspace every script, while the
+-- poll loop reads 1000 events per tick, so a row-level burst would delay every other channel.
+CREATE OR REPLACE FUNCTION notify_mcp_tools_change()
+RETURNS TRIGGER AS $$
+DECLARE
+    ws TEXT := COALESCE(NEW.workspace_id, OLD.workspace_id);
+    seen TEXT := COALESCE(current_setting('windmill.mcp_tools_notified', true), '');
+BEGIN
+    IF position(',' || ws || ',' IN seen) = 0 THEN
+        INSERT INTO notify_event (channel, payload) VALUES ('notify_mcp_tools_change', ws);
+        PERFORM set_config('windmill.mcp_tools_notified', seen || ',' || ws || ',', true);
+    END IF;
+    RETURN NULL;
+END;
+$$ LANGUAGE plpgsql SECURITY DEFINER;
+
+CREATE TRIGGER script_mcp_tools_update_trigger
+AFTER UPDATE OF archived, deleted ON script
+FOR EACH ROW
+WHEN (OLD.archived IS DISTINCT FROM NEW.archived OR OLD.deleted IS DISTINCT FROM NEW.deleted)
+EXECUTE FUNCTION notify_mcp_tools_change();
+
+CREATE TRIGGER script_mcp_tools_delete_trigger
+AFTER DELETE ON script
+FOR EACH ROW
+EXECUTE FUNCTION notify_mcp_tools_change();
+
+CREATE TRIGGER flow_mcp_tools_update_trigger
+AFTER UPDATE OF archived ON flow
+FOR EACH ROW
+WHEN (OLD.archived IS DISTINCT FROM NEW.archived)
+EXECUTE FUNCTION notify_mcp_tools_change();
+
+CREATE TRIGGER flow_mcp_tools_delete_trigger
+AFTER DELETE ON flow
+FOR EACH ROW
+EXECUTE FUNCTION notify_mcp_tools_change();
+
+-- Favorites are the tool list of an `mcp:favorites` token.
+CREATE TRIGGER favorite_mcp_tools_insert_trigger
+AFTER INSERT ON favorite
+FOR EACH ROW
+WHEN (NEW.favorite_kind IN ('script', 'flow'))
+EXECUTE FUNCTION notify_mcp_tools_change();
+
+CREATE TRIGGER favorite_mcp_tools_delete_trigger
+AFTER DELETE ON favorite
+FOR EACH ROW
+WHEN (OLD.favorite_kind IN ('script', 'flow'))
+EXECUTE FUNCTION notify_mcp_tools_change();

backend/src/main.rs

@@ -1876,10 +1876,14 @@ async fn process_notify_event(
             );
             windmill_common::workspaces::PUBLIC_APP_RATE_LIMIT_CACHE.remove(payload);
         }
+        #[cfg(feature = "mcp")]
+        "notify_mcp_tools_change" => windmill_api::notify_mcp_tools_changed(payload),
         "notify_runnable_version_change" => {
             tracing::info!("Runnable version change detected: {}", payload);
             match payload.split(':').collect::<Vec<&str>>().as_slice() {
                 [workspace_id, source_type, path, kind] => {
+                    #[cfg(feature = "mcp")]
+                    windmill_api::notify_mcp_tools_changed(workspace_id);
                     let key = (workspace_id.to_string(), path.to_string());
                     match *source_type {
                         "script" => {

backend/windmill-api/src/lib.rs

@@ -77,6 +77,8 @@ mod ai_shared_artifacts;
 mod apps;
 mod apps_raw_bundle;
 pub use apps::invalidate_app_policy_cache;
+#[cfg(feature = "mcp")]
+pub use windmill_mcp::server::notify_tools_changed as notify_mcp_tools_changed;
 pub mod args;
 mod audit;
 pub mod auth;

backend/windmill-common/tests/notify_events.rs

@@ -1083,3 +1083,44 @@ async fn test_two_server_processes_both_receive_event() {
     let _ = server_a.child.kill();
     let _ = server_b.child.kill();
 }
+
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_trigger_notify_mcp_tools_change_once_per_transaction(db: Pool<Postgres>) {
+    let script_path = format!("f/test/script_{}", uuid::Uuid::new_v4());
+    for _ in 0..3 {
+        sqlx::query(
+            "INSERT INTO script (workspace_id, hash, path, summary, description, content, created_by, language, kind)
+             VALUES ('test-workspace', $1, $2, 'test', 'test', 'def main(): pass', 'test-user', 'python3', 'script')",
+        )
+        .bind(rand::random::<i64>().abs())
+        .bind(&script_path)
+        .execute(&db)
+        .await
+        .expect("Failed to insert script");
+    }
+
+    let before_id = get_latest_event_id(&db).await.unwrap();
+    let mut tx = db.begin().await.unwrap();
+    sqlx::query("UPDATE script SET archived = true WHERE path = $1")
+        .bind(&script_path)
+        .execute(&mut *tx)
+        .await
+        .unwrap();
+    sqlx::query("DELETE FROM script WHERE path = $1")
+        .bind(&script_path)
+        .execute(&mut *tx)
+        .await
+        .unwrap();
+    tx.commit().await.unwrap();
+
+    // Archiving a path touches every version and deleting a workspace every script; one event
+    // per row would back up the poll loop that every other channel shares.
+    let events: Vec<_> = poll_notify_events(&db, before_id)
+        .await
+        .unwrap()
+        .into_iter()
+        .filter(|e| e.channel == "notify_mcp_tools_change")
+        .collect();
+    assert_eq!(events.len(), 1, "one event per workspace per transaction");
+    assert_eq!(events[0].payload, "test-workspace");
+}

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

@@ -17,7 +17,7 @@ pub use endpoints::{
     endpoint_tool_to_mcp_tool, endpoint_tool_to_mcp_tool_multi, is_endpoint_read_only,
     list_workspaces_tool, non_empty_body_fields, EndpointTool,
 };
-pub use runner::{has_endpoint_path_policy, Runner};
+pub use runner::{has_endpoint_path_policy, notify_tools_changed, Runner};
 pub use tools::create_tool_from_item;
 
 // Re-export rmcp types for convenience

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

@@ -22,14 +22,16 @@ use rmcp::model::{
     CacheScope, CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock,
     Implementation, InitializeResult, ListPromptsResult, ListResourceTemplatesResult,
     ListResourcesResult, ListToolsResult, PaginatedRequestParams, ProtocolVersion,
-    ServerCapabilities, ServerInfo,
+    ServerCapabilities, ServerInfo, ServerNotification, 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 std::sync::{Arc, LazyLock};
+use std::time::Duration;
+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,29 @@ 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 `list_changed`
+/// only reaches clients holding a `subscriptions/listen` stream, a poll interval
+/// after the change.
 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;
 
+/// Workspaces whose listed scripts or flows changed. Fed from the notify-event
+/// poll loop, so a change committed through any replica reaches the listen
+/// streams held by every replica.
+static TOOLS_CHANGED: LazyLock<broadcast::Sender<String>> =
+    LazyLock::new(|| broadcast::channel(1024).0);
+
+/// A poll batch delivers all of its events back to back; waiting this long before
+/// notifying folds a bulk change into one `list_changed` per stream.
+const TOOLS_CHANGED_COALESCE: Duration = Duration::from_secs(1);
+
+pub fn notify_tools_changed(workspace_id: &str) {
+    let _ = TOOLS_CHANGED.send(workspace_id.to_string());
+}
+
 // Re-export from http crate for extracting request parts
 use http::request::Parts as HttpParts;
 
@@ -428,12 +444,17 @@ impl<B: McpBackend> ServerHandler for Runner<B> {
             .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 +465,41 @@ impl<B: McpBackend> ServerHandler for Runner<B> {
         Cow::Borrowed(SUPPORTED_PROTOCOL_VERSIONS)
     }
 
+    fn accepted_subscription_filter(
+        &self,
+        _requested: &SubscriptionFilter,
+    ) -> Option<SubscriptionFilter> {
+        Some(SubscriptionFilter::builder().tools_list_changed().build())
+    }
+
+    async fn listen(&self, context: SubscriptionContext) -> Result<(), ErrorData> {
+        let McpContext { mode, .. } = Self::extract_context(context.request_context())?;
+        // A multi-workspace listing holds only the fixed endpoint tools.
+        let McpMode::Single(workspace_id) = mode else {
+            context.cancelled().await;
+            return Ok(());
+        };
+        let mut changes = TOOLS_CHANGED.subscribe();
+        loop {
+            tokio::select! {
+                _ = context.cancelled() => return Ok(()),
+                // `Lagged` falls through: some of the missed events may have been ours.
+                changed = changes.recv() => if matches!(changed, Ok(w_id) if w_id != workspace_id) {
+                    continue;
+                },
+            }
+            tokio::time::sleep(TOOLS_CHANGED_COALESCE).await;
+            while !matches!(
+                changes.try_recv(),
+                Err(broadcast::error::TryRecvError::Empty)
+            ) {}
+            let notification = ServerNotification::ToolListChangedNotification(Default::default());
+            if context.sink().send(notification).await.is_err() {
+                return Ok(());
+            }
+        }
+    }
+
     async fn list_tools(
         &self,
         _request: Option<PaginatedRequestParams>,