feat: refresh MCP tools when scripts and flows change (#11337)

* feat: notify MCP clients when a workspace's scripts or flows change

Advertise tools.listChanged and implement subscriptions/listen, so a
2026-07-28 client refreshes its tool list when scripts or flows are
deployed, archived, renamed or deleted. Changes reach every replica via
new statement-level notify_event triggers; MCP-originated changes also
signal the serving replica inline.

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

* fix: signal each workspace once per notify-event poll batch

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

* fix: signal script/flow path moves, keep unrelated updates trigger-free

Row-level UPDATE OF archived/deleted/path triggers with a WHEN guard
replace the statement-level ones, which built transition tables for
every UPDATE on script and flow. Path moves from username changes and
offboarding are now signalled. subscriptions/listen is refused when the
client asks for nothing this server sends.

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

* fix: signal path moves only for unarchived versions

A username change or offboarding rewrites the path of every version,
archived ones included, which would queue one notify_event per version.

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

* refactor: detect MCP tool-list changes by polling a workspace fingerprint

Replace the notify_event triggers and poller hook with a per-process,
per-workspace poll of a hash over the live scripts and flows, run only
while a subscriptions/listen stream watches that workspace. Every write
path (UI, CLI, git sync, user renames, workspace moves) is covered with
no migration; an MCP-originated change polls at once.

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

* fix: announce the first fingerprint so a change before the baseline is not lost

Also poll immediately after a script/flow mutation through a
multi-workspace token, and pin the fingerprint test on a lock update,
which is the unrelated write that actually happens.

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

* fix: open every tool-list subscription with one notification

A subscriber joining an existing watcher missed a change the watcher
recorded before it subscribed. Also move the fingerprint query to a
runnables module, since it spans scripts and flows.

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

---------

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-25 09:38:41 +00:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 1fd729ac1a
commit 0bac766756
8 changed files with 321 additions and 14 deletions
+6
View File
@@ -386,6 +386,12 @@ impl McpBackend for WindmillBackend {
})
}
async fn runnable_list_fingerprint(&self, workspace_id: &str) -> BackendResult<String> {
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<EndpointTool> {
all_tools()
}
+1
View File
@@ -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;
+22
View File
@@ -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<String, Error> {
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?)
}
@@ -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<Postgres>) {
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;
}
@@ -208,6 +208,11 @@ pub trait McpBackend: Send + Sync + Clone + 'static {
workspace_id: &str,
) -> BackendResult<Self::Auth>;
/// 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<String>;
// ─────────────────────────────────────────────────────────────────
// Endpoint Tools
// ─────────────────────────────────────────────────────────────────
@@ -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<Notify>,
}
#[derive(Default)]
pub(crate) struct ListChanges {
watchers: Mutex<HashMap<String, Watcher>>,
}
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<B: McpBackend>(
self: &Arc<Self>,
backend: &Arc<B>,
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<B: McpBackend>(
self: Arc<Self>,
backend: Arc<B>,
workspace_id: String,
changes: broadcast::Sender<()>,
poll_now: Arc<Notify>,
) {
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
),
}
}
}
}
+1
View File
@@ -7,6 +7,7 @@
pub mod backend;
pub mod endpoints;
mod list_changes;
pub mod runner;
pub mod tools;
+128 -14
View File
@@ -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<B: McpBackend> {
backend: Arc<B>,
list_changes: Arc<ListChanges>,
}
impl<B: McpBackend> Clone for Runner<B> {
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<A> {
impl<B: McpBackend> Runner<B> {
/// 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<EndpointPathPolicy> {
}
}
/// 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<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 +474,84 @@ impl<B: McpBackend> ServerHandler for Runner<B> {
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<RoleServer>,
) -> Result<InitializeResult, ErrorData> {
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<SubscriptionFilter> {
(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<PaginatedRequestParams>,
@@ -714,6 +822,9 @@ impl<B: McpBackend> Runner<B> {
.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<B: McpBackend> Runner<B> {
.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(