fix: add base64 dependency to windmill-ai for bedrock PDF support

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

backend/Cargo.lock

@@ -15804,6 +15804,7 @@ dependencies = [
  "tracing-subscriber",
  "url",
  "uuid",
+ "windmill-ai",
  "windmill-api",
  "windmill-api-agent-workers",
  "windmill-api-auth",
@@ -15835,6 +15836,30 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "windmill-ai"
+version = "1.665.0"
+dependencies = [
+ "async-trait",
+ "aws-config",
+ "aws-credential-types",
+ "aws-sdk-bedrockruntime",
+ "aws-smithy-types",
+ "base64 0.22.1",
+ "lazy_static",
+ "reqwest 0.13.1",
+ "serde",
+ "serde_json",
+ "sqlx",
+ "tokio",
+ "tracing",
+ "uuid",
+ "windmill-common",
+ "windmill-mcp",
+ "windmill-parser",
+ "windmill-types",
+]
+
 [[package]]
 name = "windmill-alerting"
 version = "1.665.0"
@@ -15934,6 +15959,7 @@ dependencies = [
  "url",
  "urlencoding",
  "uuid",
+ "windmill-ai",
  "windmill-alerting",
  "windmill-api-agent-workers",
  "windmill-api-assets",
@@ -16359,6 +16385,7 @@ dependencies = [
  "tokio",
  "tracing",
  "uuid",
+ "windmill-ai",
  "windmill-alerting",
  "windmill-api-auth",
  "windmill-common",
@@ -17567,6 +17594,7 @@ dependencies = [
  "url",
  "urlencoding",
  "uuid",
+ "windmill-ai",
  "windmill-audit",
  "windmill-common",
  "windmill-dep-map",

backend/Cargo.toml

@@ -7,6 +7,7 @@ edition.workspace = true
 [workspace]
 resolver = "2"
 members = [
+    "./windmill-ai",
     "./windmill-object-store",
     "./windmill-api",
     "./windmill-api-scripts",
@@ -132,8 +133,8 @@ dind = ["windmill-worker/dind"]
 websocket = ["windmill-api/websocket"]
 http_trigger = ["windmill-api/http_trigger"]
 postgres_trigger = ["windmill-api/postgres_trigger"]
-mcp = ["windmill-api/mcp", "windmill-worker/mcp"]
-bedrock = ["windmill-api/bedrock", "windmill-worker/bedrock", "windmill-common/bedrock"]
+mcp = ["windmill-ai/mcp", "windmill-api/mcp", "windmill-worker/mcp"]
+bedrock = ["windmill-ai/bedrock", "windmill-api/bedrock", "windmill-worker/bedrock", "windmill-common/bedrock"]
 mqtt_trigger = ["windmill-api/mqtt_trigger"]
 native_trigger = ["windmill-api/native_trigger"]
 sqs_trigger = ["windmill-api/sqs_trigger", "windmill-common/aws_auth", "windmill-api/openidconnect"]
@@ -206,6 +207,7 @@ anyhow.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
 dotenv.workspace = true
+windmill-ai = { workspace = true, default-features = false }
 windmill-queue.workspace = true
 windmill-common = { workspace = true, default-features = false }
 windmill-object-store.workspace = true
@@ -284,6 +286,7 @@ aws-credential-types.workspace = true
 
 
 [workspace.dependencies]
+windmill-ai = { path = "./windmill-ai", default-features = false }
 windmill-api = { path = "./windmill-api", default-features = false  }
 windmill-queue = { path = "./windmill-queue" }
 windmill-worker = { path = "./windmill-worker" }

backend/src/main.rs

@@ -34,9 +34,9 @@ use windmill_common::ee_oss::{
     maybe_renew_license_key_on_start, LICENSE_KEY_ID, LICENSE_KEY_VALID,
 };
 
+use windmill_ai::ai_cache::bump_instance_ai_config_revision;
 use windmill_common::{
     agent_workers::AgentConfig,
-    ai_cache::bump_instance_ai_config_revision,
     global_settings::{
         AI_CONFIG_SETTING, APP_WORKSPACED_ROUTE_SETTING, AUDIT_LOG_RETENTION_DAYS_SETTING,
         BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING,

backend/windmill-ai/Cargo.toml

@@ -0,0 +1,37 @@
+[package]
+name = "windmill-ai"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+
+[features]
+default = []
+bedrock = ["dep:aws-sdk-bedrockruntime", "dep:aws-credential-types", "dep:aws-smithy-types", "dep:aws-config"]
+mcp = ["dep:windmill-mcp"]
+
+[lib]
+name = "windmill_ai"
+path = "src/lib.rs"
+
+[dependencies]
+windmill-common = { workspace = true, default-features = false }
+windmill-types.workspace = true
+windmill-parser.workspace = true
+windmill-mcp = { workspace = true, optional = true }
+
+async-trait.workspace = true
+base64.workspace = true
+reqwest.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+sqlx = { workspace = true, features = ["postgres"] }
+uuid.workspace = true
+lazy_static.workspace = true
+tracing.workspace = true
+tokio.workspace = true
+
+# Bedrock (optional)
+aws-config = { workspace = true, optional = true }
+aws-credential-types = { workspace = true, optional = true }
+aws-smithy-types = { workspace = true, optional = true }
+aws-sdk-bedrockruntime = { workspace = true, optional = true }

backend/windmill-ai/src/ai_bedrock.rs

@@ -18,7 +18,7 @@ use aws_sdk_bedrockruntime::types::{
 use aws_sdk_bedrockruntime::Client as BedrockRuntimeClient;
 use serde::{Deserialize, Serialize};
 
-use crate::error::Error;
+use windmill_common::error::Error;
 
 use crate::ai_types::{
     ContentPart, OpenAIContent, OpenAIFunction, OpenAIMessage, OpenAIToolCall, ToolDef,

backend/windmill-ai/src/ai_google.rs

@@ -13,7 +13,7 @@ use crate::ai_types::{
     ContentPart, ExtraContent, GoogleExtraContent, OpenAIContent, OpenAIMessage, ToolDef,
     UrlCitation,
 };
-use crate::error::Error;
+use windmill_common::error::Error;
 
 // ============================================================================
 // Request / Content Types

backend/windmill-ai/src/ai_providers.rs

@@ -2,8 +2,8 @@
  * This file contains shared AI provider utilities used by both the API and worker.
  */
 
-use crate::db::DB;
-use crate::error::{Error, Result};
+use windmill_common::db::DB;
+use windmill_common::error::{Error, Result};
 use serde::{Deserialize, Deserializer, Serialize};
 
 /// Deserializes an Option<String> where empty strings become None.

backend/windmill-ai/src/ai_types.rs

@@ -6,7 +6,7 @@
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
 
-use crate::flow_status::AgentAction;
+use windmill_common::flow_status::AgentAction;
 use windmill_types::s3::S3Object;
 
 // ============================================================================

backend/windmill-ai/src/lib.rs

@@ -0,0 +1,8 @@
+#[cfg(feature = "bedrock")]
+pub mod ai_bedrock;
+pub mod ai_cache;
+pub mod ai_google;
+pub mod ai_providers;
+pub mod ai_types;
+pub mod query_builder;
+pub mod types;

backend/windmill-ai/src/query_builder.rs

@@ -0,0 +1,99 @@
+use async_trait::async_trait;
+use windmill_common::{client::AuthedClient, error::Error};
+use windmill_types::s3::S3Object;
+
+use crate::ai_types::OpenAIToolCall;
+use crate::types::*;
+
+/// Arguments for building an AI request
+pub struct BuildRequestArgs<'a> {
+    pub messages: &'a [OpenAIMessage],
+    pub tools: Option<&'a [ToolDef]>,
+    pub model: &'a str,
+    pub temperature: Option<f32>,
+    pub max_tokens: Option<u32>,
+    pub output_schema: Option<&'a OpenAPISchema>,
+    pub output_type: &'a OutputType,
+    pub system_prompt: Option<&'a str>,
+    pub user_message: &'a str,
+    pub attachments: Option<&'a [S3Object]>,
+    pub has_websearch: bool,
+}
+
+/// Response from AI provider
+pub enum ParsedResponse {
+    Text {
+        content: Option<String>,
+        tool_calls: Vec<OpenAIToolCall>,
+        events_str: Option<String>,
+        annotations: Vec<UrlCitation>,
+        used_websearch: bool,
+        usage: Option<TokenUsage>,
+    },
+    Image {
+        base64_data: String,
+    },
+}
+
+/// Trait for streaming AI events to a sink (e.g., database persistence).
+/// Implemented by the worker's StreamEventProcessor.
+#[async_trait]
+pub trait StreamEventSink: Send + Sync {
+    async fn send(&self, event: StreamingEvent, events_str: &mut String) -> Result<(), Error>;
+}
+
+/// Trait for building provider-specific AI requests
+#[async_trait]
+pub trait QueryBuilder: Send + Sync {
+    /// Check if this provider supports tools with the given output type
+    fn supports_tools_with_output_type(&self, output_type: &OutputType) -> bool;
+
+    /// Build the request body for the provider
+    async fn build_request(
+        &self,
+        args: &BuildRequestArgs<'_>,
+        client: &AuthedClient,
+        workspace_id: &str,
+    ) -> Result<String, Error>;
+
+    /// Build the request body without usage tracking (for retry on incompatible providers)
+    /// Default implementation just calls build_request (most providers don't need this)
+    async fn build_request_without_usage(
+        &self,
+        args: &BuildRequestArgs<'_>,
+        client: &AuthedClient,
+        workspace_id: &str,
+    ) -> Result<String, Error> {
+        self.build_request(args, client, workspace_id).await
+    }
+
+    /// Whether this provider supports retry without usage tracking
+    /// Only OtherQueryBuilder (OpenAI-compatible providers) needs this
+    fn supports_retry_without_usage(&self) -> bool {
+        false
+    }
+
+    /// Parse the image response from the provider
+    async fn parse_image_response(
+        &self,
+        response: reqwest::Response,
+    ) -> Result<ParsedResponse, Error>;
+
+    /// Parse streaming response from the provider
+    async fn parse_streaming_response(
+        &self,
+        response: reqwest::Response,
+        stream_event_sink: Box<dyn StreamEventSink>,
+    ) -> Result<ParsedResponse, Error>;
+
+    /// Get the API endpoint for this provider
+    fn get_endpoint(&self, base_url: &str, model: &str, output_type: &OutputType) -> String;
+
+    /// Get the authentication headers for this provider
+    fn get_auth_headers(
+        &self,
+        api_key: &str,
+        base_url: &str,
+        output_type: &OutputType,
+    ) -> Vec<(&'static str, String)>;
+}

backend/windmill-api-settings/Cargo.toml

@@ -16,6 +16,7 @@ parquet = ["windmill-common/parquet", "windmill-object-store/parquet"]
 license = ["dep:rsa"]
 
 [dependencies]
+windmill-ai = { workspace = true, default-features = false }
 windmill-alerting.workspace = true
 windmill-api-auth.workspace = true
 windmill-common = { workspace = true, default-features = false }

backend/windmill-api-settings/src/lib.rs

@@ -37,8 +37,8 @@ use serde::{Deserialize, Serialize};
 use windmill_common::ee_oss::{send_critical_alert, CriticalAlertKind, CriticalErrorChannel};
 #[cfg(all(feature = "private", feature = "enterprise"))]
 use windmill_common::secret_backend::{SecretMigrationReport, VaultSettings};
+use windmill_ai::ai_cache::bump_instance_ai_config_revision;
 use windmill_common::{
-    ai_cache::bump_instance_ai_config_revision,
     email_oss::send_email_plain_text,
     error::{self, JsonResult, Result},
     get_database_url,

backend/windmill-api/Cargo.toml

@@ -39,12 +39,13 @@ sqs_trigger = ["dep:windmill-trigger-sqs", "windmill-store/sqs_trigger"]
 gcp_trigger = ["dep:windmill-trigger-gcp", "windmill-store/gcp_trigger"]
 cloud = ["windmill-common/cloud", "windmill-api-auth/cloud", "windmill-store/cloud", "windmill-api-workspaces/cloud"]
 mcp = ["dep:windmill-mcp", "windmill-mcp/server", "windmill-mcp/auth", "windmill-api-auth/mcp", "windmill-store/mcp"]
-bedrock = ["dep:aws-sdk-bedrock", "dep:aws-sdk-bedrockruntime", "windmill-common/bedrock", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
+bedrock = ["windmill-ai/bedrock", "dep:aws-sdk-bedrock", "dep:aws-sdk-bedrockruntime", "windmill-common/bedrock", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
 python = ["windmill-dep-map/python", "dep:windmill-parser-py", "dep:windmill-parser-py-imports", "windmill-api-scripts/python", "windmill-api-configs/python", "windmill-api-agent-workers?/python", "windmill-trigger/python", "windmill-common/python"]
 no_auth = ["windmill-api-auth/no_auth", "windmill-store/no_auth", "windmill-api-users/no_auth"]
 quickjs = ["windmill-jseval/quickjs"]
 
 [dependencies]
+windmill-ai = { workspace = true, default-features = false }
 windmill-mcp = { workspace = true, optional = true }
 windmill-api-auth.workspace = true
 windmill-api-scripts.workspace = true

backend/windmill-api/src/ai.rs

@@ -16,8 +16,8 @@ use serde_json::{json, value::RawValue};
 use std::collections::HashMap;
 use std::time::Duration;
 use windmill_audit::{audit_oss::audit_log, ActionKind};
-use windmill_common::ai_cache::current_instance_ai_config_revision;
-use windmill_common::ai_providers::{
+use windmill_ai::ai_cache::current_instance_ai_config_revision;
+use windmill_ai::ai_providers::{
     empty_string_as_none, AIPlatform, AIProvider, ProviderConfig, ProviderModel,
 };
 use windmill_common::error::{to_anyhow, Error, Result};
@@ -624,8 +624,8 @@ pub fn workspaced_service() -> Router {
 async fn check_bedrock_credentials(
     _authed: ApiAuthed,
     Path(_w_id): Path<String>,
-) -> Result<Json<windmill_common::ai_bedrock::BedrockCredentialsCheck>> {
-    let response = windmill_common::ai_bedrock::check_env_credentials().await;
+) -> Result<Json<windmill_ai::ai_bedrock::BedrockCredentialsCheck>> {
+    let response = windmill_ai::ai_bedrock::check_env_credentials().await;
     Ok(Json(response))
 }
 
@@ -945,7 +945,7 @@ async fn proxy(
             let region = request_config
                 .region
                 .as_deref()
-                .unwrap_or(windmill_common::ai_providers::USE_ENV_REGION);
+                .unwrap_or(windmill_ai::ai_providers::USE_ENV_REGION);
 
             // Audit log before making the SDK request
             let mut tx = db.begin().await?;
@@ -1062,8 +1062,8 @@ async fn proxy(
 mod tests {
     use super::*;
     use std::sync::{LazyLock, Mutex};
-    use windmill_common::ai_cache::bump_instance_ai_config_revision;
-    use windmill_common::ai_providers::AIPlatform;
+    use windmill_ai::ai_cache::bump_instance_ai_config_revision;
+    use windmill_ai::ai_providers::AIPlatform;
 
     static TEST_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
 

backend/windmill-api/src/bedrock.rs

@@ -18,13 +18,13 @@
 
 use axum::body::Bytes;
 use serde::Deserialize;
-use windmill_common::ai_bedrock::build_tool_config;
-use windmill_common::ai_bedrock::{
+use windmill_ai::ai_bedrock::build_tool_config;
+use windmill_ai::ai_bedrock::{
     bedrock_stream_event_is_block_stop, bedrock_stream_event_to_text,
     bedrock_stream_event_to_tool_delta, bedrock_stream_event_to_tool_start, format_bedrock_error,
     BedrockClient,
 };
-use windmill_common::ai_types::{
+use windmill_ai::ai_types::{
     OpenAIFunction, OpenAIMessage, OpenAIToolCall, ToolDef, ToolDefFunction,
 };
 use windmill_common::error::{Error, Result};
@@ -182,7 +182,7 @@ async fn create_bedrock_control_client(
     region: &str,
 ) -> Result<aws_sdk_bedrock::Client> {
     use aws_config::BehaviorVersion;
-    use windmill_common::ai_bedrock::BearerTokenProvider;
+    use windmill_ai::ai_bedrock::BearerTokenProvider;
 
     let region_provider = aws_sdk_bedrock::config::Region::new(region.to_string());
 
@@ -366,10 +366,10 @@ pub async fn handle_bedrock_sdk_streaming(
 
     // Convert messages using shared conversion
     let (bedrock_messages, system_prompts) =
-        windmill_common::ai_bedrock::openai_messages_to_bedrock(&openai_req.messages)?;
+        windmill_ai::ai_bedrock::openai_messages_to_bedrock(&openai_req.messages)?;
 
     // Build inference configuration
-    let inference_config = windmill_common::ai_bedrock::create_inference_config(
+    let inference_config = windmill_ai::ai_bedrock::create_inference_config(
         openai_req.temperature,
         openai_req.max_tokens,
     );
@@ -626,10 +626,10 @@ pub async fn handle_bedrock_sdk_non_streaming(
 
     // Convert messages using shared conversion
     let (bedrock_messages, system_prompts) =
-        windmill_common::ai_bedrock::openai_messages_to_bedrock(&openai_req.messages)?;
+        windmill_ai::ai_bedrock::openai_messages_to_bedrock(&openai_req.messages)?;
 
     // Build inference configuration
-    let inference_config = windmill_common::ai_bedrock::create_inference_config(
+    let inference_config = windmill_ai::ai_bedrock::create_inference_config(
         openai_req.temperature,
         openai_req.max_tokens,
     );

backend/windmill-api/src/google.rs

@@ -15,15 +15,15 @@ use eventsource_stream::Eventsource;
 use futures::StreamExt;
 use serde::Deserialize;
 use serde_json::json;
-use windmill_common::{
+use windmill_ai::{
     ai_google::{
         gemini_event_to_openai_sse_chunks, gemini_response_to_openai, openai_messages_to_gemini,
         parse_gemini_response, parse_gemini_sse_event, sanitize_schema_for_google,
         GeminiFunctionDeclaration, GeminiGenerationConfig, GeminiTextRequest, GeminiTool,
     },
     ai_types::OpenAIMessage,
-    error::{Error, Result},
 };
+use windmill_common::error::{Error, Result};
 
 use crate::ai::{inject_keepalives, HTTP_CLIENT, KEEPALIVE_INTERVAL_SECS};
 

backend/windmill-common/src/lib.rs

@@ -27,12 +27,6 @@ use scripts::ScriptLang;
 use sqlx::{Acquire, Postgres};
 
 pub mod agent_workers;
-#[cfg(feature = "bedrock")]
-pub mod ai_bedrock;
-pub mod ai_cache;
-pub mod ai_google;
-pub mod ai_providers;
-pub mod ai_types;
 pub mod apps;
 pub mod assets;
 pub mod audit;

backend/windmill-worker/Cargo.toml

@@ -11,7 +11,7 @@ path = "src/lib.rs"
 [features]
 default = []
 private = ["windmill-worker-volumes/private", "windmill-queue/private"]
-mcp = ["dep:windmill-mcp"]
+mcp = ["windmill-ai/mcp", "dep:windmill-mcp"]
 prometheus = ["dep:prometheus", "windmill-common/prometheus"]
 enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker-volumes/enterprise", "dep:pem", "dep:tokio-util", "dep:opentelemetry-proto", "dep:prost", "dep:hudsucker", "dep:rcgen", "dep:hyper-http-proxy", "dep:hyper-tls", "dep:hyper-util"]
 mssql = ["dep:tiberius"]
@@ -38,9 +38,10 @@ java = ["dep:windmill-parser-java"]
 ruby = ["dep:windmill-parser-ruby"]
 duckdb = ["dep:libloading"]
 quickjs = ["windmill-jseval/quickjs"]
-bedrock = ["dep:aws-sdk-bedrockruntime", "windmill-common/bedrock", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
+bedrock = ["windmill-ai/bedrock", "dep:aws-sdk-bedrockruntime", "windmill-common/bedrock", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
 
 [dependencies]
+windmill-ai = { workspace = true, default-features = false }
 windmill-queue.workspace = true
 windmill-dep-map.workspace = true
 windmill-audit.workspace = true # there isn't really a reason for audit-worth actions to happen in the worker.

backend/windmill-worker/src/ai/image_handler.rs

@@ -80,7 +80,7 @@ pub async fn s3_object_to_content_part(
         download_and_encode_s3_image(s3_object, client, workspace_id).await?;
     let data_url = format!("data:{};base64,{}", mime_type, file_bytes);
 
-    if windmill_common::ai_types::is_document_mime(&mime_type) {
+    if windmill_ai::ai_types::is_document_mime(&mime_type) {
         let filename = s3_object
             .s3
             .rsplit('/')

backend/windmill-worker/src/ai/providers/anthropic.rs

@@ -1,13 +1,12 @@
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
-use windmill_common::{
-    ai_google::parse_data_url, ai_providers::AIProvider, client::AuthedClient, error::Error,
-};
+use windmill_ai::{ai_google::parse_data_url, ai_providers::AIProvider};
+use windmill_common::{client::AuthedClient, error::Error};
 
 use crate::ai::{
     image_handler::prepare_messages_for_api,
-    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventProcessor},
+    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventSink},
     sse::{AnthropicSSEParser, SSEParser},
     types::*,
     utils::{extract_text_content, should_use_structured_output_tool},
@@ -536,9 +535,9 @@ impl QueryBuilder for AnthropicQueryBuilder {
     async fn parse_streaming_response(
         &self,
         response: reqwest::Response,
-        stream_event_processor: StreamEventProcessor,
+        stream_event_sink: Box<dyn StreamEventSink>,
     ) -> Result<ParsedResponse, Error> {
-        let mut anthropic_sse_parser = AnthropicSSEParser::new(stream_event_processor);
+        let mut anthropic_sse_parser = AnthropicSSEParser::new(stream_event_sink);
         anthropic_sse_parser.parse_events(response).await?;
 
         let AnthropicSSEParser {

backend/windmill-worker/src/ai/providers/bedrock.rs

@@ -1,6 +1,6 @@
 //! AWS Bedrock provider for the AI agent.
 //!
-//! Uses shared SDK code from windmill_common::ai_bedrock for:
+//! Uses shared SDK code from windmill_ai::ai_bedrock for:
 //! - BedrockClient (SDK wrapper with auth)
 //! - Message conversion (OpenAI format -> Bedrock format)
 //! - Stream event parsing
@@ -8,7 +8,7 @@
 
 use crate::ai::{
     image_handler::prepare_messages_for_api,
-    query_builder::{ParsedResponse, StreamEventProcessor},
+    query_builder::{ParsedResponse, StreamEventSink},
     types::StreamingEvent,
     types::TokenUsage,
     types::{OpenAIMessage, ToolDef},
@@ -17,13 +17,13 @@ use std::collections::HashMap;
 use windmill_common::{client::AuthedClient, error::Error};
 
 // Re-export from shared module for use by other parts of the worker
-use windmill_common::ai_bedrock::{
+use windmill_ai::ai_bedrock::{
     bedrock_stream_event_is_block_stop, bedrock_stream_event_to_text,
     bedrock_stream_event_to_tool_delta, bedrock_stream_event_to_tool_start, build_tool_config,
     create_inference_config, format_bedrock_error, openai_messages_to_bedrock,
     streaming_tool_calls_to_openai, StreamingToolCall,
 };
-pub use windmill_common::ai_bedrock::{check_env_credentials, BedrockClient};
+pub use windmill_ai::ai_bedrock::{check_env_credentials, BedrockClient};
 
 // ============================================================================
 // Query Builder (Worker-specific orchestration)
@@ -43,7 +43,7 @@ impl BedrockQueryBuilder {
         max_tokens: Option<u32>,
         api_key: &str,
         region: &str,
-        stream_event_processor: Option<StreamEventProcessor>,
+        stream_event_sink: Option<Box<dyn StreamEventSink>>,
         client: &AuthedClient,
         workspace_id: &str,
         structured_output_tool_name: Option<&str>,
@@ -86,7 +86,7 @@ impl BedrockQueryBuilder {
             system_prompts,
             inference_config,
             tool_config,
-            stream_event_processor,
+            stream_event_sink,
         )
         .await
     }
@@ -100,7 +100,7 @@ impl BedrockQueryBuilder {
         system_prompts: Vec<aws_sdk_bedrockruntime::types::SystemContentBlock>,
         inference_config: Option<aws_sdk_bedrockruntime::types::InferenceConfiguration>,
         tool_config: Option<aws_sdk_bedrockruntime::types::ToolConfiguration>,
-        stream_event_processor: Option<StreamEventProcessor>,
+        stream_event_sink: Option<Box<dyn StreamEventSink>>,
     ) -> Result<ParsedResponse, Error> {
         tracing::debug!(
             "Worker Bedrock: executing converse_stream, messages={}, system_prompts={}, has_tools={}",
@@ -159,7 +159,7 @@ impl BedrockQueryBuilder {
                     // Handle text delta using shared parser
                     if let Some(text_delta) = bedrock_stream_event_to_text(&event) {
                         accumulated_text.push_str(&text_delta);
-                        if let Some(processor) = stream_event_processor.as_ref() {
+                        if let Some(processor) = stream_event_sink.as_ref() {
                             processor
                                 .send(
                                     StreamingEvent::TokenDelta { content: text_delta },
@@ -214,7 +214,7 @@ impl BedrockQueryBuilder {
         }
 
         // Send tool call events to stream processor
-        if let Some(processor) = stream_event_processor.as_ref() {
+        if let Some(processor) = stream_event_sink.as_ref() {
             for tool_call in accumulated_tool_calls.values() {
                 processor
                     .send(

backend/windmill-worker/src/ai/providers/google_ai.rs

@@ -1,17 +1,14 @@
 use async_trait::async_trait;
-use windmill_common::{
-    ai_google::{
-        openai_messages_to_gemini, openai_tools_to_gemini, GeminiGenerationConfig,
-        GeminiImageContent, GeminiImageRequest, GeminiImageResponse, GeminiInlineData, GeminiPart,
-        GeminiPredictContent, GeminiTextRequest, GeminiTool,
-    },
-    client::AuthedClient,
-    error::Error,
+use windmill_ai::ai_google::{
+    openai_messages_to_gemini, openai_tools_to_gemini, GeminiGenerationConfig,
+    GeminiImageContent, GeminiImageRequest, GeminiImageResponse, GeminiInlineData, GeminiPart,
+    GeminiPredictContent, GeminiTextRequest, GeminiTool,
 };
+use windmill_common::{client::AuthedClient, error::Error};
 
 use crate::ai::{
     image_handler::{download_and_encode_s3_image, prepare_messages_for_api},
-    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventProcessor},
+    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventSink},
     sse::{GeminiSSEParser, SSEParser},
     types::*,
 };
@@ -222,9 +219,9 @@ impl QueryBuilder for GoogleAIQueryBuilder {
     async fn parse_streaming_response(
         &self,
         response: reqwest::Response,
-        stream_event_processor: StreamEventProcessor,
+        stream_event_sink: Box<dyn StreamEventSink>,
     ) -> Result<ParsedResponse, Error> {
-        let mut gemini_sse_parser = GeminiSSEParser::new(stream_event_processor);
+        let mut gemini_sse_parser = GeminiSSEParser::new(stream_event_sink);
         gemini_sse_parser.parse_events(response).await?;
 
         let GeminiSSEParser {

backend/windmill-worker/src/ai/providers/openai.rs

@@ -1,18 +1,18 @@
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
-use windmill_common::{ai_providers::AIProvider, client::AuthedClient, error::Error};
+use windmill_ai::ai_providers::AIProvider;
+use windmill_ai::ai_types::OpenAIToolCall;
+use windmill_common::{client::AuthedClient, error::Error};
 
 use crate::ai::{
     image_handler::{prepare_messages_for_api, s3_object_to_content_part},
-    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventProcessor},
+    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventSink},
     sse::{OpenAIResponsesSSEParser, SSEParser},
     types::*,
     utils::extract_text_content,
 };
 
-use windmill_common::ai_types::OpenAIToolCall;
-
 // Responses API structures
 #[derive(Deserialize)]
 #[allow(dead_code)]
@@ -486,9 +486,9 @@ impl QueryBuilder for OpenAIQueryBuilder {
     async fn parse_streaming_response(
         &self,
         response: reqwest::Response,
-        stream_event_processor: StreamEventProcessor,
+        stream_event_sink: Box<dyn StreamEventSink>,
     ) -> Result<ParsedResponse, Error> {
-        let mut parser = OpenAIResponsesSSEParser::new(stream_event_processor);
+        let mut parser = OpenAIResponsesSSEParser::new(stream_event_sink);
         parser.parse_events(response).await?;
 
         // Convert OpenAI Responses usage to TokenUsage

backend/windmill-worker/src/ai/providers/openrouter.rs

@@ -1,12 +1,13 @@
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use serde_json;
-use windmill_common::{ai_providers::AIProvider, client::AuthedClient, error::Error};
+use windmill_ai::ai_providers::AIProvider;
+use windmill_common::{client::AuthedClient, error::Error};
 
 use crate::ai::{
     image_handler::prepare_messages_for_api,
     providers::other::OtherQueryBuilder,
-    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventProcessor},
+    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventSink},
     types::*,
 };
 
@@ -129,10 +130,10 @@ impl QueryBuilder for OpenRouterQueryBuilder {
     async fn parse_streaming_response(
         &self,
         response: reqwest::Response,
-        stream_event_processor: StreamEventProcessor,
+        stream_event_sink: Box<dyn StreamEventSink>,
     ) -> Result<ParsedResponse, Error> {
         self.other_builder
-            .parse_streaming_response(response, stream_event_processor)
+            .parse_streaming_response(response, stream_event_sink)
             .await
     }
 

backend/windmill-worker/src/ai/providers/other.rs

@@ -1,11 +1,12 @@
 use async_trait::async_trait;
 use serde::Serialize;
 use serde_json;
-use windmill_common::{ai_providers::AIProvider, client::AuthedClient, error::Error};
+use windmill_ai::ai_providers::AIProvider;
+use windmill_common::{client::AuthedClient, error::Error};
 
 use crate::ai::{
     image_handler::prepare_messages_for_api,
-    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventProcessor},
+    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventSink},
     sse::{OpenAISSEParser, SSEParser},
     types::*,
     utils::should_use_structured_output_tool,
@@ -200,9 +201,9 @@ impl QueryBuilder for OtherQueryBuilder {
     async fn parse_streaming_response(
         &self,
         response: reqwest::Response,
-        stream_event_processor: StreamEventProcessor,
+        stream_event_sink: Box<dyn StreamEventSink>,
     ) -> Result<ParsedResponse, Error> {
-        let mut openai_sse_parser = OpenAISSEParser::new(stream_event_processor);
+        let mut openai_sse_parser = OpenAISSEParser::new(stream_event_sink);
         openai_sse_parser.parse_events(response).await?;
 
         let OpenAISSEParser {

backend/windmill-worker/src/ai/query_builder.rs

@@ -1,7 +1,6 @@
 use async_trait::async_trait;
-use windmill_common::{client::AuthedClient, error::Error, worker::Connection};
+use windmill_common::{error::Error, worker::Connection};
 use windmill_queue::MiniPulledJob;
-use windmill_types::s3::S3Object;
 
 use crate::{
     ai::{
@@ -15,119 +14,32 @@ use crate::{
     job_logger::append_result_stream,
 };
 
-use windmill_common::ai_types::OpenAIToolCall;
-
-/// Arguments for building an AI request
-pub struct BuildRequestArgs<'a> {
-    pub messages: &'a [OpenAIMessage],
-    pub tools: Option<&'a [ToolDef]>,
-    pub model: &'a str,
-    pub temperature: Option<f32>,
-    pub max_tokens: Option<u32>,
-    pub output_schema: Option<&'a OpenAPISchema>,
-    pub output_type: &'a OutputType,
-    pub system_prompt: Option<&'a str>,
-    pub user_message: &'a str,
-    pub attachments: Option<&'a [S3Object]>,
-    pub has_websearch: bool,
-}
-
-use crate::ai::types::TokenUsage;
-
-/// Response from AI provider
-pub enum ParsedResponse {
-    Text {
-        content: Option<String>,
-        tool_calls: Vec<OpenAIToolCall>,
-        events_str: Option<String>,
-        annotations: Vec<UrlCitation>,
-        used_websearch: bool,
-        usage: Option<TokenUsage>,
-    },
-    Image {
-        base64_data: String,
-    },
-}
-
-/// Trait for building provider-specific AI requests
-#[async_trait]
-pub trait QueryBuilder: Send + Sync {
-    /// Check if this provider supports tools with the given output type
-    fn supports_tools_with_output_type(&self, output_type: &OutputType) -> bool;
-
-    /// Build the request body for the provider
-    async fn build_request(
-        &self,
-        args: &BuildRequestArgs<'_>,
-        client: &AuthedClient,
-        workspace_id: &str,
-    ) -> Result<String, Error>;
-
-    /// Build the request body without usage tracking (for retry on incompatible providers)
-    /// Default implementation just calls build_request (most providers don't need this)
-    async fn build_request_without_usage(
-        &self,
-        args: &BuildRequestArgs<'_>,
-        client: &AuthedClient,
-        workspace_id: &str,
-    ) -> Result<String, Error> {
-        self.build_request(args, client, workspace_id).await
-    }
-
-    /// Whether this provider supports retry without usage tracking
-    /// Only OtherQueryBuilder (OpenAI-compatible providers) needs this
-    fn supports_retry_without_usage(&self) -> bool {
-        false
-    }
-
-    /// Parse the image response from the provider
-    async fn parse_image_response(
-        &self,
-        response: reqwest::Response,
-    ) -> Result<ParsedResponse, Error>;
-
-    /// Parse streaming response from the provider
-    async fn parse_streaming_response(
-        &self,
-        _response: reqwest::Response,
-        _stream_event_processor: StreamEventProcessor,
-    ) -> Result<ParsedResponse, Error>;
-
-    /// Get the API endpoint for this provider
-    fn get_endpoint(&self, base_url: &str, model: &str, output_type: &OutputType) -> String;
-
-    /// Get the authentication headers for this provider
-    fn get_auth_headers(
-        &self,
-        api_key: &str,
-        base_url: &str,
-        output_type: &OutputType,
-    ) -> Vec<(&'static str, String)>;
-}
+// Re-export from windmill_ai
+pub use windmill_ai::query_builder::{
+    BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventSink,
+};
 
 /// Factory function to create the appropriate query builder for a provider
 pub fn create_query_builder(provider: &ProviderWithResource) -> Box<dyn QueryBuilder> {
-    use windmill_common::ai_providers::AIProvider;
+    use windmill_ai::ai_providers::AIProvider;
 
     match provider.kind {
-        // Google AI uses the Gemini API (with platform-specific handling for Vertex AI)
         AIProvider::GoogleAI => Box::new(GoogleAIQueryBuilder::new(
             provider.get_platform().clone(),
         )),
-        // OpenAI use the Responses API
         AIProvider::OpenAI => Box::new(OpenAIQueryBuilder::new(provider.kind.clone())),
-        // Anthropic uses its own API format (with platform-specific handling for Vertex AI)
         AIProvider::Anthropic => Box::new(AnthropicQueryBuilder::new(
             provider.kind.clone(),
             provider.get_platform().clone(),
             provider.get_enable_1m_context(),
         )),
         AIProvider::OpenRouter => Box::new(OpenRouterQueryBuilder::new()),
-        // All other providers use the completion endpoint
         _ => Box::new(OtherQueryBuilder::new(provider.kind.clone())),
     }
 }
 
+/// Processes streaming events by persisting them to the database.
+/// Implements StreamEventSink so it can be passed to QueryBuilder methods.
 pub struct StreamEventProcessor {
     tx: Option<tokio::sync::mpsc::Sender<String>>,
     pub handle: Option<tokio::task::JoinHandle<()>>,
@@ -178,7 +90,14 @@ impl StreamEventProcessor {
         Self { tx: None, handle: None }
     }
 
-    pub async fn send(&self, event: StreamingEvent, events_str: &mut String) -> Result<(), Error> {
+    pub fn to_handle(self) -> Option<tokio::task::JoinHandle<()>> {
+        self.handle
+    }
+}
+
+#[async_trait]
+impl StreamEventSink for StreamEventProcessor {
+    async fn send(&self, event: StreamingEvent, events_str: &mut String) -> Result<(), Error> {
         // In silent mode, skip both persistence AND accumulation (no overhead)
         let Some(ref tx) = self.tx else {
             return Ok(());
@@ -208,8 +127,4 @@ impl StreamEventProcessor {
             ))),
         }
     }
-
-    pub fn to_handle(self) -> Option<tokio::task::JoinHandle<()>> {
-        self.handle
-    }
 }

backend/windmill-worker/src/ai/sse.rs

@@ -5,15 +5,17 @@ use reqwest::Response;
 use serde::Deserialize;
 use serde_json;
 use tokio_stream::StreamExt;
-use windmill_common::{
+use windmill_ai::{
     ai_google::{parse_gemini_sse_event, GeminiUsageMetadata},
     ai_types::{ExtraContent, GoogleExtraContent, OpenAIFunction, OpenAIToolCall},
+};
+use windmill_common::{
     error::Error,
     utils::rd_string,
 };
 
 use crate::ai::{
-    query_builder::StreamEventProcessor,
+    query_builder::StreamEventSink,
     types::{StreamingEvent, UrlCitation},
 };
 
@@ -107,13 +109,13 @@ pub struct OpenAISSEParser {
     pub accumulated_content: String,
     pub accumulated_tool_calls: HashMap<i64, OpenAIToolCall>,
     pub events_str: String,
-    pub stream_event_processor: StreamEventProcessor,
+    pub stream_event_processor: Box<dyn StreamEventSink>,
     /// Token usage from final chunk (when stream_options.include_usage is true)
     pub usage: Option<OpenAIChatUsage>,
 }
 
 impl OpenAISSEParser {
-    pub fn new(stream_event_processor: StreamEventProcessor) -> Self {
+    pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
             accumulated_tool_calls: HashMap::new(),
@@ -303,7 +305,7 @@ pub struct AnthropicSSEParser {
     pub accumulated_content: String,
     pub accumulated_tool_calls: HashMap<i64, OpenAIToolCall>,
     pub events_str: String,
-    pub stream_event_processor: StreamEventProcessor,
+    pub stream_event_processor: Box<dyn StreamEventSink>,
     /// Track content block types by index
     content_blocks: HashMap<usize, ContentBlockState>,
     /// Collected URL citation annotations from web search
@@ -315,7 +317,7 @@ pub struct AnthropicSSEParser {
 }
 
 impl AnthropicSSEParser {
-    pub fn new(stream_event_processor: StreamEventProcessor) -> Self {
+    pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
             accumulated_tool_calls: HashMap::new(),
@@ -469,7 +471,7 @@ pub struct GeminiSSEParser {
     pub accumulated_content: String,
     pub accumulated_tool_calls: HashMap<i64, OpenAIToolCall>,
     pub events_str: String,
-    pub stream_event_processor: StreamEventProcessor,
+    pub stream_event_processor: Box<dyn StreamEventSink>,
     tool_call_index: i64,
     pub annotations: Vec<UrlCitation>,
     pub used_websearch: bool,
@@ -477,7 +479,7 @@ pub struct GeminiSSEParser {
 }
 
 impl GeminiSSEParser {
-    pub fn new(stream_event_processor: StreamEventProcessor) -> Self {
+    pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
             accumulated_tool_calls: HashMap::new(),
@@ -659,7 +661,7 @@ pub struct OpenAIResponsesSSEParser {
     /// Maps item_id -> accumulated arguments
     tool_call_arguments: HashMap<String, String>,
     pub events_str: String,
-    pub stream_event_processor: StreamEventProcessor,
+    pub stream_event_processor: Box<dyn StreamEventSink>,
     /// Collected URL citation annotations from web search
     pub annotations: Vec<UrlCitation>,
     /// Whether web search was used in this response
@@ -669,7 +671,7 @@ pub struct OpenAIResponsesSSEParser {
 }
 
 impl OpenAIResponsesSSEParser {
-    pub fn new(stream_event_processor: StreamEventProcessor) -> Self {
+    pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
             accumulated_tool_calls: HashMap::new(),

backend/windmill-worker/src/ai/tools.rs

@@ -1,4 +1,4 @@
-use crate::ai::query_builder::StreamEventProcessor;
+use crate::ai::query_builder::{StreamEventProcessor, StreamEventSink};
 use crate::ai::types::McpToolSource;
 use crate::ai::types::*;
 use crate::ai::utils::{
@@ -20,7 +20,7 @@ use mappable_rc::Marc;
 use serde_json::value::RawValue;
 use std::{collections::HashMap, sync::Arc};
 use uuid::Uuid;
-use windmill_common::ai_types::OpenAIToolCall;
+use windmill_ai::ai_types::OpenAIToolCall;
 use windmill_common::jobs::JobPayload;
 
 #[cfg(feature = "mcp")]

backend/windmill-worker/src/ai/utils.rs

@@ -8,9 +8,9 @@ use std::{
     sync::Arc,
 };
 use uuid::Uuid;
+use windmill_ai::ai_providers::AIProvider;
 use windmill_common::flows::FlowModuleValue;
 use windmill_common::{
-    ai_providers::AIProvider,
     db::DB,
     error::Error,
     flow_conversations::{add_message_to_conversation_tx, MessageType},

backend/windmill-worker/src/ai_executor.rs

@@ -20,8 +20,8 @@ use windmill_mcp::McpClient;
 
 #[cfg(not(feature = "mcp"))]
 use crate::ai::tools::McpClientStub as McpClient;
+use windmill_ai::ai_providers::AIProvider;
 use windmill_common::{
-    ai_providers::AIProvider,
     cache,
     client::AuthedClient,
     db::DB,
@@ -864,7 +864,7 @@ pub async fn run_agent(
                 let region = args
                     .provider
                     .get_region()
-                    .unwrap_or(windmill_common::ai_providers::USE_ENV_REGION);
+                    .unwrap_or(windmill_ai::ai_providers::USE_ENV_REGION);
                 // Use Bedrock SDK via dedicated query builder
                 crate::ai::providers::bedrock::BedrockQueryBuilder::default()
                     .execute_request(
@@ -875,7 +875,7 @@ pub async fn run_agent(
                         args.max_completion_tokens,
                         api_key,
                         region,
-                        stream_event_processor.clone(),
+                        stream_event_processor.as_ref().map(|p| Box::new(p.clone()) as Box<dyn windmill_ai::query_builder::StreamEventSink>),
                         client,
                         &job.workspace_id,
                         structured_output_tool_name.as_deref(),
@@ -1001,7 +1001,7 @@ pub async fn run_agent(
 
             if let Some(ref stream_event_processor) = stream_event_processor {
                 query_builder
-                    .parse_streaming_response(resp, stream_event_processor.clone())
+                    .parse_streaming_response(resp, Box::new(stream_event_processor.clone()))
                     .await?
             } else {
                 query_builder.parse_image_response(resp).await?