diff --git a/backend/migrations/20260924065755_mcp_tools_change_notify.down.sql b/backend/migrations/20260924065755_mcp_tools_change_notify.down.sql new file mode 100644 index 0000000000..a151eaacdb --- /dev/null +++ b/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(); diff --git a/backend/migrations/20260924065755_mcp_tools_change_notify.up.sql b/backend/migrations/20260924065755_mcp_tools_change_notify.up.sql new file mode 100644 index 0000000000..83c2d1dfcc --- /dev/null +++ b/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(); diff --git a/backend/src/main.rs b/backend/src/main.rs index 9e7a93cfcd..9752a2d672 100644 --- a/backend/src/main.rs +++ b/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::>().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" => { diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index d6b29a8501..1feb6d5c02 100644 --- a/backend/windmill-api/src/lib.rs +++ b/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; diff --git a/backend/windmill-common/tests/notify_events.rs b/backend/windmill-common/tests/notify_events.rs index 285086308e..2759c0db5c 100644 --- a/backend/windmill-common/tests/notify_events.rs +++ b/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) { + 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::().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"); +} diff --git a/backend/windmill-mcp/src/server/mod.rs b/backend/windmill-mcp/src/server/mod.rs index b97e374e98..0514d18109 100644 --- a/backend/windmill-mcp/src/server/mod.rs +++ b/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 diff --git a/backend/windmill-mcp/src/server/runner.rs b/backend/windmill-mcp/src/server/runner.rs index e27eaa8470..85e4e0a92c 100644 --- a/backend/windmill-mcp/src/server/runner.rs +++ b/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> = + 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 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 +465,41 @@ impl ServerHandler for Runner { Cow::Borrowed(SUPPORTED_PROTOCOL_VERSIONS) } + fn accepted_subscription_filter( + &self, + _requested: &SubscriptionFilter, + ) -> Option { + 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,