diff --git a/backend/migrations/20260924065755_mcp_tools_change_notify.down.sql b/backend/migrations/20260924065755_mcp_tools_change_notify.down.sql index a151eaacdb..e56f6e8be5 100644 --- a/backend/migrations/20260924065755_mcp_tools_change_notify.down.sql +++ b/backend/migrations/20260924065755_mcp_tools_change_notify.down.sql @@ -1,6 +1,9 @@ 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_insert_trigger ON flow; DROP TRIGGER IF EXISTS flow_mcp_tools_update_trigger ON flow; +DROP TRIGGER IF EXISTS resource_mcp_tools_insert_delete_trigger ON resource; +DROP TRIGGER IF EXISTS resource_mcp_tools_rename_trigger ON resource; 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; diff --git a/backend/migrations/20260924065755_mcp_tools_change_notify.up.sql b/backend/migrations/20260924065755_mcp_tools_change_notify.up.sql index 83c2d1dfcc..1461473db6 100644 --- a/backend/migrations/20260924065755_mcp_tools_change_notify.up.sql +++ b/backend/migrations/20260924065755_mcp_tools_change_notify.up.sql @@ -1,5 +1,5 @@ --- 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: +-- Deploys already emit `notify_runnable_version_change`; these cover the other changes to what +-- 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() @@ -27,6 +27,14 @@ AFTER DELETE ON script FOR EACH ROW EXECUTE FUNCTION notify_mcp_tools_change(); +-- A create inserts the row with no versions and appends one later, which the versions trigger +-- sees. Restore from trash, rename and fork insert the row with its versions already set. +CREATE TRIGGER flow_mcp_tools_insert_trigger +AFTER INSERT ON flow +FOR EACH ROW +WHEN (cardinality(NEW.versions) > 0) +EXECUTE FUNCTION notify_mcp_tools_change(); + CREATE TRIGGER flow_mcp_tools_update_trigger AFTER UPDATE OF archived ON flow FOR EACH ROW @@ -38,6 +46,18 @@ AFTER DELETE ON flow FOR EACH ROW EXECUTE FUNCTION notify_mcp_tools_change(); +-- A tool's input schema lists the workspace's resources of each resource type it takes. +CREATE TRIGGER resource_mcp_tools_insert_delete_trigger +AFTER INSERT OR DELETE ON resource +FOR EACH ROW +EXECUTE FUNCTION notify_mcp_tools_change(); + +CREATE TRIGGER resource_mcp_tools_rename_trigger +AFTER UPDATE OF path ON resource +FOR EACH ROW +WHEN (OLD.path IS DISTINCT FROM NEW.path) +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 diff --git a/backend/src/main.rs b/backend/src/main.rs index 9752a2d672..35c7dfae97 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -1876,8 +1876,10 @@ 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_mcp_tools_change" => { + #[cfg(feature = "mcp")] + windmill_api::notify_mcp_tools_changed(payload); + } "notify_runnable_version_change" => { tracing::info!("Runnable version change detected: {}", payload); match payload.split(':').collect::>().as_slice() { diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 1feb6d5c02..2bae774470 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -77,8 +77,6 @@ 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; @@ -211,6 +209,8 @@ mod mcp_tools; #[cfg(feature = "mcp")] mod mcp; +#[cfg(feature = "mcp")] +pub use windmill_mcp::server::notify_tools_changed as notify_mcp_tools_changed; #[cfg(all(feature = "mcp", feature = "private"))] mod mcp_oauth_ee; #[cfg(feature = "mcp")] diff --git a/backend/windmill-mcp/src/server/runner.rs b/backend/windmill-mcp/src/server/runner.rs index e067ee648f..52918b0ee4 100644 --- a/backend/windmill-mcp/src/server/runner.rs +++ b/backend/windmill-mcp/src/server/runner.rs @@ -57,16 +57,19 @@ const LIST_TTL_MS: u64 = 0; /// 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. +/// Workspaces whose tool listing changed. Fed from the notify-event poll loop, so +/// a change committed through any replica reaches the listen streams held by +/// every replica. Sized well past the poll loop's 1000-event batch: a lagged +/// receiver notifies, and every stream on the instance would re-list at once. static TOOLS_CHANGED: LazyLock> = - LazyLock::new(|| broadcast::channel(1024).0); + LazyLock::new(|| broadcast::channel(8192).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); +/// Wake the `subscriptions/listen` streams of `workspace_id` on this process. +/// Fire-and-forget: with no stream open, the event is dropped. pub fn notify_tools_changed(workspace_id: &str) { let _ = TOOLS_CHANGED.send(workspace_id.to_string()); } @@ -465,10 +468,10 @@ impl ServerHandler for Runner { Cow::Borrowed(SUPPORTED_PROTOCOL_VERSIONS) } - /// Only the pre-2026-07-28 revisions reach `initialize`, and they have no stream - /// to receive `list_changed` on (legacy sessions are off and - /// `subscriptions/listen` is 2026-07-28 only), so the capability is withdrawn - /// here. It stays in `get_info`, which bounds what a listen filter may accept. + /// A pre-2026-07-28 revision has no stream to receive `list_changed` on (legacy + /// sessions are off and `subscriptions/listen` is 2026-07-28 only), so the + /// capability is withdrawn when one is negotiated. It stays in `get_info`, which + /// bounds what a listen filter may accept. async fn initialize( &self, request: InitializeRequestParams, @@ -479,8 +482,10 @@ impl ServerHandler for Runner { if SUPPORTED_PROTOCOL_VERSIONS.contains(&request.protocol_version) { info.protocol_version = request.protocol_version; } - if let Some(tools) = info.capabilities.tools.as_mut() { - tools.list_changed = None; + if info.protocol_version != ProtocolVersion::V_2026_07_28 { + if let Some(tools) = info.capabilities.tools.as_mut() { + tools.list_changed = None; + } } Ok(info) }