diff --git a/backend/.sqlx/query-80858777f29eacf087a399eca67c398e598763051b286edcbe05112ae70521c9.json b/backend/.sqlx/query-80858777f29eacf087a399eca67c398e598763051b286edcbe05112ae70521c9.json deleted file mode 100644 index 8ad744a0db..0000000000 --- a/backend/.sqlx/query-80858777f29eacf087a399eca67c398e598763051b286edcbe05112ae70521c9.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT\n (flow_status->>'memory_id')::uuid as memory_id,\n (flow_status->>'chat_input_enabled')::boolean as chat_input_enabled\n FROM v2_job_status\n WHERE id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "memory_id", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "chat_input_enabled", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [ - null, - null - ] - }, - "hash": "80858777f29eacf087a399eca67c398e598763051b286edcbe05112ae70521c9" -} diff --git a/backend/.sqlx/query-dd89d652154748d6d7e625e31778f6885d0ee62d29a4b8894a4b459dd215a103.json b/backend/.sqlx/query-dd89d652154748d6d7e625e31778f6885d0ee62d29a4b8894a4b459dd215a103.json new file mode 100644 index 0000000000..4a82795a3c --- /dev/null +++ b/backend/.sqlx/query-dd89d652154748d6d7e625e31778f6885d0ee62d29a4b8894a4b459dd215a103.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n j.args as \"args: Json>>\",\n js.flow_status as \"flow_status: Json\"\n FROM v2_job_status js\n INNER JOIN v2_job j ON j.id = js.id\n WHERE js.id = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "args: Json>>", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "flow_status: Json", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "dd89d652154748d6d7e625e31778f6885d0ee62d29a4b8894a4b459dd215a103" +} diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 1f52565182..fc45ed08cc 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -2,21 +2,25 @@ use crate::ai::providers::openai::OpenAIToolCall; use crate::ai::query_builder::StreamEventProcessor; use crate::ai::types::*; use crate::ai::utils::{ - add_message_to_conversation, execute_mcp_tool, get_flow_chat_settings, get_step_name_from_flow, + add_message_to_conversation, execute_mcp_tool, get_step_name_from_flow, update_flow_status_module_with_actions, update_flow_status_module_with_actions_success, - FlowChatSettings, + FlowContext, }; use crate::common::{error_to_value, OccupancyMetrics}; use crate::result_processor::handle_non_flow_job_error; -use crate::worker_flow::{raw_script_to_payload, script_to_payload, JobPayloadWithTag}; +use crate::worker_flow::{ + evaluate_input_transform, raw_script_to_payload, script_to_payload, JobPayloadWithTag, +}; use crate::{ create_job_dir, handle_queued_job, JobCompletedReceiver, JobCompletedSender, SendResult, SendResultPayload, }; use anyhow::Context; +use mappable_rc::Marc; use serde_json::value::RawValue; use std::{collections::HashMap, sync::Arc}; use uuid::Uuid; +use windmill_common::flows::InputTransform; use windmill_common::jobs::JobPayload; use windmill_common::mcp_client::{McpClient, McpToolSource}; use windmill_common::{ @@ -26,7 +30,7 @@ use windmill_common::{ flow_conversations::MessageType, flow_status::AgentAction, flows::FlowModuleValue, - worker::Connection, + worker::{to_raw_value, Connection}, }; use windmill_queue::{ get_mini_pulled_job, push, JobCompleted, MiniPulledJob, PushArgs, PushIsolationLevel, @@ -57,7 +61,9 @@ pub struct ToolExecutionContext<'a> { // Optional streaming & chat pub stream_event_processor: Option<&'a StreamEventProcessor>, - pub chat_settings: &'a mut Option, + pub flow_context: &'a mut FlowContext, + pub previous_result: &'a Option>, + pub id_context: &'a Option, } /// Execute all tool calls from an AI response @@ -257,10 +263,7 @@ async fn execute_windmill_tool( ) -> Result<(), Error> { // Regular Windmill tools must have a module let tool_module = tool.module.as_ref().ok_or_else(|| { - Error::internal_err(format!( - "Tool {} has no module (MCP tools should be handled above)", - tool_call.function.name - )) + Error::internal_err(format!("Tool {} has no module", tool_call.function.name)) })?; let job_id = ulid::Ulid::new().into(); @@ -278,7 +281,7 @@ async fn execute_windmill_tool( tool_call.function.arguments.clone() }; - let tool_call_args = serde_json::from_str::>>( + let mut tool_call_args = serde_json::from_str::>>( &raw_tool_call_args, ) .with_context(|| { @@ -288,6 +291,54 @@ async fn execute_windmill_tool( ) })?; + // Get input transforms given by the user and merge them with AI given args + let input_transforms = match tool_module.get_value()? { + FlowModuleValue::Script { input_transforms, .. } => input_transforms, + FlowModuleValue::RawScript { input_transforms, .. } => input_transforms, + FlowModuleValue::FlowScript { input_transforms, .. } => input_transforms, + _ => { + return Err(Error::internal_err(format!( + "Unsupported tool: {}", + tool_call.function.name + ))); + } + }; + + // Prepare context for transform evaluation + let last_result = Arc::new( + ctx.previous_result + .as_ref() + .cloned() + .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)), + ); + + let flow_inputs = ctx + .flow_context + .flow_inputs + .as_ref() + .map(|args| Marc::new(args.clone())); + + // Evaluate each input transform and merge with AI-provided args + for (key, transform) in input_transforms.iter() { + // We skip static empty / null values, those are the one the AI will fill in + if let InputTransform::Static { value } = transform { + let val = value.get().trim(); + if val.is_empty() || val == "null" { + continue; + } + } + let result = evaluate_input_transform::>( + transform, + last_result.clone(), + flow_inputs.clone(), + Some(ctx.client), + ctx.id_context.as_ref(), + ) + .await?; + + tool_call_args.insert(key.clone(), result); + } + let job_payload = match tool_module.get_value()? { FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => { script_to_payload( @@ -659,18 +710,19 @@ async fn add_tool_message_to_chat( content: &str, success: bool, ) { - if ctx.chat_settings.is_none() { - *ctx.chat_settings = Some(get_flow_chat_settings(ctx.db, ctx.job).await); - } - let chat_enabled = ctx - .chat_settings + .flow_context + .flow_status .as_ref() - .map(|s| s.chat_input_enabled) + .and_then(|fs| fs.chat_input_enabled) .unwrap_or(false); - if chat_enabled { - if let Some(mid) = ctx.chat_settings.as_ref().and_then(|s| s.memory_id) { + if let Some(memory_id) = ctx + .flow_context + .flow_status + .as_ref() + .and_then(|fs| fs.memory_id) + { let db_clone = ctx.db.clone(); let step_name = get_step_name_from_flow(ctx.summary.as_deref(), ctx.job.flow_step_id.as_deref()); @@ -680,7 +732,7 @@ async fn add_tool_message_to_chat( tokio::spawn(async move { if let Err(e) = add_message_to_conversation( &db_clone, - &mid, + &memory_id, tool_job_id, &content, MessageType::Tool, @@ -689,7 +741,11 @@ async fn add_tool_message_to_chat( ) .await { - tracing::warn!("Failed to add tool message to conversation {}: {}", mid, e); + tracing::warn!( + "Failed to add tool message to conversation {}: {}", + memory_id, + e + ); } }); } diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index 70c022577a..6f70c02c84 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -1,20 +1,27 @@ use crate::ai::types::{ToolDef, ToolDefFunction}; use anyhow::Context; use serde_json::value::RawValue; -use std::{collections::HashMap, sync::Arc}; +use sqlx::types::Json; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; use uuid::Uuid; -use windmill_common::mcp_client::{McpClient, McpResource, McpToolSource}; use windmill_common::{ ai_providers::AIProvider, db::DB, error::Error, flow_conversations::{add_message_to_conversation_tx, MessageType}, flow_status::AgentAction, - flows::Step, + flows::{InputTransform, Step}, jobs::JobKind, scripts::{ScriptHash, ScriptLang}, worker::to_raw_value, }; +use windmill_common::{ + flows::FlowModuleValue, + mcp_client::{McpClient, McpResource, McpToolSource}, +}; use windmill_queue::{flow_status::get_step_of_flow_status, MiniPulledJob}; use crate::{ai::types::*, parse_sig_of_lang}; @@ -51,6 +58,66 @@ pub fn parse_raw_script_schema( Ok(to_raw_value(&schema)) } +/// Filters out properties from a JSON schema that have completed input transforms. +/// This allows AI agents to only see and fill parameters that don't have user-configured values. +pub fn filter_schema_by_input_transforms( + schema: Box, + input_transforms: &HashMap, +) -> Result, Error> { + // Parse the schema JSON + let mut schema_value: serde_json::Value = serde_json::from_str(schema.get()) + .context("Failed to parse schema JSON") + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + // Collect keys to remove (parameters with completed input transforms) + let keys_to_remove: HashSet = input_transforms + .iter() + .filter_map(|(key, transform)| { + let is_completed = match transform { + InputTransform::Static { value } => { + let val = value.get().trim(); + !val.is_empty() && val != "null" + } + InputTransform::Javascript { expr } => !expr.trim().is_empty(), + }; + if is_completed { + Some(key.clone()) + } else { + None + } + }) + .collect(); + + if !keys_to_remove.is_empty() { + // Remove completed parameters from properties + if let Some(properties) = schema_value + .get_mut("properties") + .and_then(|p| p.as_object_mut()) + { + for key in &keys_to_remove { + properties.remove(key); + } + } + + // Also remove from required array + if let Some(required) = schema_value + .get_mut("required") + .and_then(|r| r.as_array_mut()) + { + required.retain(|item| { + if let Some(key) = item.as_str() { + !keys_to_remove.contains(key) + } else { + true + } + }); + } + } + + // Convert back to RawValue + Ok(to_raw_value(&schema_value)) +} + pub struct FlowJobRunnableIdAndRawFlow { pub runnable_id: Option, pub raw_flow: Option>>, @@ -72,45 +139,51 @@ pub async fn get_flow_job_runnable_and_raw_flow( } #[derive(Debug, Clone, Default)] -pub struct FlowChatSettings { - pub memory_id: Option, - pub chat_input_enabled: bool, +pub struct FlowContext { + pub flow_inputs: Option>>, + pub flow_status: Option, } -/// Get chat settings (memory_id and chat_input_enabled) from root flow's flow_status -pub async fn get_flow_chat_settings(db: &DB, job: &MiniPulledJob) -> FlowChatSettings { +/// Get flow context (chat settings + args + flow_status) from root flow's job data +pub async fn get_flow_context(db: &DB, job: &MiniPulledJob) -> FlowContext { let root_job_id = job .root_job .or(job.flow_innermost_root_job) .or(job.parent_job); let Some(root_job_id) = root_job_id else { - return FlowChatSettings::default(); + return FlowContext::default(); }; match sqlx::query!( - "SELECT - (flow_status->>'memory_id')::uuid as memory_id, - (flow_status->>'chat_input_enabled')::boolean as chat_input_enabled - FROM v2_job_status - WHERE id = $1", + r#" + SELECT + j.args as "args: Json>>", + js.flow_status as "flow_status: Json" + FROM v2_job_status js + INNER JOIN v2_job j ON j.id = js.id + WHERE js.id = $1 + "#, root_job_id ) .fetch_optional(db) .await { - Ok(Some(row)) => FlowChatSettings { - memory_id: row.memory_id, - chat_input_enabled: row.chat_input_enabled.unwrap_or(false), + Ok(Some(row)) => FlowContext { + flow_inputs: row.args.map(|j| j.0), + flow_status: row.flow_status.map(|j| j.0), }, - Ok(None) => FlowChatSettings::default(), - Err(e) => { + Ok(None) => { tracing::warn!( - "Failed to get chat settings from flow status for job {}: {}", - job.id, - e + "No flow context found for root job {} (agent job {}), returning default", + root_job_id, + job.id ); - FlowChatSettings::default() + FlowContext::default() + } + Err(e) => { + tracing::error!("Failed to get flow context for job {}: {}", job.id, e); + FlowContext::default() } } } @@ -463,3 +536,28 @@ pub async fn execute_mcp_tool( Ok(result) } + +/// Check if any tool's input transforms reference previous_result +pub fn any_tool_needs_previous_result(tools: &[Tool]) -> bool { + tools.iter().any(|tool| { + if let Some(module) = &tool.module { + if let Ok(module_value) = module.get_value() { + let input_transforms = match module_value { + FlowModuleValue::Script { input_transforms, .. } => input_transforms, + FlowModuleValue::RawScript { input_transforms, .. } => input_transforms, + FlowModuleValue::FlowScript { input_transforms, .. } => input_transforms, + _ => return false, + }; + + return input_transforms.iter().any(|(_, transform)| { + if let windmill_common::flows::InputTransform::Javascript { expr } = transform { + expr.contains("previous_result") + } else { + false + } + }); + } + } + false + }) +} diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 6b6596f9cb..9032a0ad07 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -1,12 +1,13 @@ use crate::ai::tools::{execute_tool_calls, ToolExecutionContext}; use crate::ai::utils::{ - add_message_to_conversation, cleanup_mcp_clients, find_unique_tool_name, - get_flow_chat_settings, get_flow_job_runnable_and_raw_flow, get_step_name_from_flow, - is_anthropic_provider, load_mcp_tools, parse_raw_script_schema, - update_flow_status_module_with_actions, update_flow_status_module_with_actions_success, - FlowChatSettings, + add_message_to_conversation, any_tool_needs_previous_result, cleanup_mcp_clients, + filter_schema_by_input_transforms, find_unique_tool_name, get_flow_context, + get_flow_job_runnable_and_raw_flow, get_step_name_from_flow, is_anthropic_provider, + load_mcp_tools, parse_raw_script_schema, update_flow_status_module_with_actions, + update_flow_status_module_with_actions_success, }; use crate::memory_oss::{read_from_memory, write_to_memory}; +use crate::worker_flow::{get_previous_job_result, get_transform_context}; use async_recursion::async_recursion; use regex::Regex; use serde_json::value::RawValue; @@ -161,70 +162,69 @@ pub async fn handle_ai_agent_job( ))); }; - let schema = match &t.get_value() { - Ok(FlowModuleValue::Script { + // Extract schema and input_transforms from the module value + let module_value = t.get_value()?; + let (schema, input_transforms) = match &module_value { + FlowModuleValue::Script { hash, path, tag_override, input_transforms, is_trigger, pass_flow_input_directly, - }) => match hash { - Some(hash) => { - let (_, metadata) = cache::script::fetch(conn, hash.clone()).await?; - Ok::<_, Error>( - metadata - .schema - .clone() - .map(|s| RawValue::from_string(s).ok()) - .flatten(), - ) - } - None => { - if path.starts_with("hub/") { - let hub_script = get_full_hub_script_by_path( - StripPath(path.to_string()), - &HTTP_CLIENT, - None, + } => { + let schema = match hash { + Some(hash) => { + let (_, metadata) = cache::script::fetch(conn, hash.clone()).await?; + Ok::<_, Error>( + metadata + .schema + .clone() + .map(|s| RawValue::from_string(s).ok()) + .flatten(), ) - .await?; - Ok(Some(hub_script.schema)) - } else { - let hash = get_latest_hash_for_path( - db, - &job.workspace_id, - path.as_str(), - true, - ) - .await? - .0; - // update module definition to use a fixed hash so all tool calls match the same schema - t.value = to_raw_value(&FlowModuleValue::Script { - hash: Some(hash), - path: path.clone(), - tag_override: tag_override.clone(), - input_transforms: input_transforms.clone(), - is_trigger: *is_trigger, - pass_flow_input_directly: *pass_flow_input_directly, - }); - let (_, metadata) = cache::script::fetch(conn, hash).await?; - Ok(metadata - .schema - .clone() - .map(|s| RawValue::from_string(s).ok()) - .flatten()) } - } - }, - Ok(FlowModuleValue::RawScript { content, language, .. }) => { - Ok(Some(parse_raw_script_schema(&content, &language)?)) + None => { + if path.starts_with("hub/") { + let hub_script = get_full_hub_script_by_path( + StripPath(path.to_string()), + &HTTP_CLIENT, + None, + ) + .await?; + Ok(Some(hub_script.schema)) + } else { + let hash = get_latest_hash_for_path( + db, + &job.workspace_id, + path.as_str(), + true, + ) + .await? + .0; + // update module definition to use a fixed hash so all tool calls match the same schema + t.value = to_raw_value(&FlowModuleValue::Script { + hash: Some(hash), + path: path.clone(), + tag_override: tag_override.clone(), + input_transforms: input_transforms.clone(), + is_trigger: *is_trigger, + pass_flow_input_directly: *pass_flow_input_directly, + }); + let (_, metadata) = cache::script::fetch(conn, hash).await?; + Ok(metadata + .schema + .clone() + .map(|s| RawValue::from_string(s).ok()) + .flatten()) + } + } + }?; + (schema, input_transforms) } - Err(e) => { - return Err(Error::internal_err(format!( - "Invalid tool {}: {}", - summary, - e.to_string() - ))); + FlowModuleValue::RawScript { content, language, input_transforms, .. } => { + let schema = Some(parse_raw_script_schema(&content, &language)?); + (schema, input_transforms) } _ => { return Err(Error::internal_err(format!( @@ -232,7 +232,14 @@ pub async fn handle_ai_agent_job( summary ))); } - }?; + }; + + // Filter schema based on user given input transforms + let schema = if let Some(s) = schema { + Some(filter_schema_by_input_transforms(s, input_transforms)?) + } else { + None + }; Ok(Tool { def: ToolDef { @@ -358,15 +365,18 @@ pub async fn run_agent( vec![] }; - let mut chat_settings: Option = None; + // Fetch flow context for input transforms context, chat and memory + let mut flow_context = get_flow_context(db, job).await; // Load previous messages from memory for text output mode (only if context length is set) if matches!(output_type, OutputType::Text) { if let Some(context_length) = args.messages_context_length.filter(|&n| n > 0) { if let Some(step_id) = job.flow_step_id.as_deref() { - // Fetch chat settings from root flow - chat_settings = Some(get_flow_chat_settings(db, job).await); - if let Some(memory_id) = chat_settings.as_ref().and_then(|s| s.memory_id) { + if let Some(memory_id) = flow_context + .flow_status + .as_ref() + .and_then(|fs| fs.memory_id) + { // Read messages from memory match read_from_memory(&job.workspace_id, memory_id, step_id).await { Ok(Some(loaded_messages)) => { @@ -393,6 +403,37 @@ pub async fn run_agent( } } + // Extract previous step result only if any tool needs it + let previous_result = { + if any_tool_needs_previous_result(&tools) { + if let Some(ref flow_status) = flow_context.flow_status { + get_previous_job_result(db, &job.workspace_id, flow_status) + .await + .ok() + .flatten() + } else { + None + } + } else { + None + } + }; + + // Build IdContext for results.stepId syntax + let id_context = { + if let Some(ref flow_status) = flow_context.flow_status { + // Get the step ID from the AI agent's flow step + let previous_id = job + .flow_step_id + .clone() + .unwrap_or_else(|| "unknown".to_string()); + + Some(get_transform_context(job, &previous_id, flow_status).await?) + } else { + None + } + }; + // Create user message with optional images let mut parts = vec![ContentPart::Text { text: args.user_message.clone() }]; if let Some(images) = &args.user_images { @@ -563,16 +604,16 @@ pub async fn run_agent( content = Some(OpenAIContent::Text(response_content.clone())); // Add assistant message to conversation if chat_input_enabled - if chat_settings.is_none() { - chat_settings = Some(get_flow_chat_settings(db, job).await); - } - let chat_enabled = chat_settings + let chat_enabled = flow_context + .flow_status .as_ref() - .map(|s| s.chat_input_enabled) + .and_then(|fs| fs.chat_input_enabled) .unwrap_or(false); - if chat_enabled && !response_content.is_empty() { - if let Some(mid) = chat_settings.as_ref().and_then(|s| s.memory_id) + if let Some(memory_id) = flow_context + .flow_status + .as_ref() + .and_then(|fs| fs.memory_id) { let agent_job_id = job.id; let db_clone = db.clone(); @@ -586,7 +627,7 @@ pub async fn run_agent( tokio::spawn(async move { if let Err(e) = add_message_to_conversation( &db_clone, - &mid, + &memory_id, Some(agent_job_id), &message_content, MessageType::Assistant, @@ -595,7 +636,7 @@ pub async fn run_agent( ) .await { - tracing::warn!("Failed to add assistant message to conversation {}: {}", mid, e); + tracing::warn!("Failed to add assistant message to conversation {}: {}", memory_id, e); } }); } @@ -633,7 +674,9 @@ pub async fn run_agent( job_completed_tx, killpill_rx, stream_event_processor: stream_event_processor.as_ref(), - chat_settings: &mut chat_settings, + flow_context: &mut flow_context, + previous_result: &previous_result, + id_context: &id_context, }; let (tool_messages, tool_content, tool_used_structured_output) = @@ -724,7 +767,7 @@ pub async fn run_agent( let start_idx = all_messages.len().saturating_sub(context_length); let messages_to_persist = all_messages[start_idx..].to_vec(); - if let Some(memory_id) = chat_settings.as_ref().and_then(|s| s.memory_id) { + if let Some(memory_id) = flow_context.flow_status.and_then(|fs| fs.memory_id) { if let Err(e) = write_to_memory( &job.workspace_id, memory_id, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index c566c99a07..9e956dbfb8 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -4574,7 +4574,7 @@ pub async fn script_to_payload( }) } -async fn get_transform_context( +pub async fn get_transform_context( flow_job: &MiniPulledJob, previous_id: &str, status: &FlowStatus, @@ -4641,7 +4641,7 @@ fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(Suspend, Uuid) } // returns the result of the previous step of a running flow (if the job was successful) -async fn get_previous_job_result( +pub async fn get_previous_job_result( db: &sqlx::Pool, w_id: &str, flow_status: &FlowStatus, diff --git a/frontend/src/lib/components/InputTransformForm.svelte b/frontend/src/lib/components/InputTransformForm.svelte index 4c7a56825e..0f18081134 100644 --- a/frontend/src/lib/components/InputTransformForm.svelte +++ b/frontend/src/lib/components/InputTransformForm.svelte @@ -40,6 +40,9 @@ import S3ArrayHelperButton from './S3ArrayHelperButton.svelte' import { inputBorderClass } from './text_input/TextInput.svelte' + // We add 'ai' for ai agent tools. 'ai' means the field will be filled by the AI agent dynamically. + type PropertyType = InputTransform['type'] | 'ai' + interface Props { schema: Schema | { properties?: Record; required?: string[] } arg: InputTransform | any @@ -62,6 +65,7 @@ editor?: SimpleEditor | undefined otherArgs?: Record helperScript?: DynamicInputTypes.HelperScript | undefined + isAgentTool?: boolean } let { @@ -85,7 +89,8 @@ class: className = '', editor = $bindable(undefined), otherArgs = {}, - helperScript = undefined + helperScript = undefined, + isAgentTool = false }: Props = $props() let monaco: SimpleEditor | undefined = $state(undefined) @@ -151,8 +156,13 @@ }) } - function getPropertyType(arg: InputTransform | any): 'static' | 'javascript' { - let type: 'static' | 'javascript' = arg?.type ?? 'static' + function getPropertyType(arg: InputTransform | any): PropertyType { + // For agent tools, if static with undefined/empty value, treat as 'ai', meaning the field will be filled by the AI agent dynamically. + if (isAgentTool && arg?.type === 'static' && arg?.value === undefined) { + return 'ai' + } + + let type: PropertyType = arg?.type ?? 'static' if ( type == 'javascript' && @@ -373,7 +383,7 @@ function updateStaticInput( inputCat: InputCat, - propertyType: 'static' | 'javascript', + propertyType: PropertyType, arg: InputTransform | any ) { if (!isStaticTemplate(inputCat)) { @@ -550,7 +560,15 @@ if (e.detail == propertyType) return const staticTemplate = isStaticTemplate(inputCat) - if (e.detail === 'javascript') { + if (e.detail === 'ai') { + // Switch to AI mode: static with no value + if (arg) { + arg.type = 'static' + arg.value = undefined + arg.expr = undefined + } + propertyType = 'ai' + } else if (e.detail === 'javascript') { if (arg.expr == undefined) { arg.expr = getDefaultExpr( argName, @@ -600,6 +618,16 @@ }} > {#snippet children({ item })} + {#if isAgentTool} + + {/if} + {#if isStaticTemplate(inputCat)}
- {#if isStaticTemplate(inputCat) && propertyType == 'static' && !noDynamicToggle} + {#if propertyType === 'ai'} +
+ + + This field will be filled by the AI agent dynamically + +
+ {#if argName && schema?.properties?.[argName]?.description} +
+ 
{schema.properties[argName].description}
+
+ {/if} + {:else if isStaticTemplate(inputCat) && propertyType == 'static' && !noDynamicToggle}
{#if argName && schema?.properties?.[argName]?.description} 
-										{schema.properties[argName].description}
-
+ {schema.properties[argName].description} +
{/if} + {#if arg} key !== argName) )} diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index 529f7f90d6..cb3be1bb23 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -126,7 +126,7 @@ shellcheck: false }) - let selected = $state(preprocessorModule || isAgentTool ? 'test' : 'inputs') + let selected = $state(preprocessorModule ? 'test' : 'inputs') let advancedSelected = $state('retries') let advancedRuntimeSelected = $state('concurrency') let s3Kind = $state('s3_client') @@ -553,7 +553,7 @@
- {#if !preprocessorModule && !isAgentTool} + {#if !preprocessorModule} {/if} @@ -597,6 +597,7 @@ } extraLib={stepPropPicker.extraLib} {enableAi} + {isAgentTool} helperScript={retrieveDynCodeAndLang(flowModule.value)} /> diff --git a/frontend/src/lib/components/flows/content/FlowModuleWrapper.svelte b/frontend/src/lib/components/flows/content/FlowModuleWrapper.svelte index 95d1ce7cd8..8af071e7f5 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleWrapper.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleWrapper.svelte @@ -294,16 +294,17 @@ {/if} {/each} {:else if flowModule.value.type === 'aiagent'} - {@const toolIndex = flowModule.value.tools.findIndex((t) => t.id === $selectedId)} - {#if toolIndex !== -1} - - {/if} + {#each flowModule.value.tools as tool, toolIndex (toolIndex)} + {#if $selectedId === tool.id} + + {/if} + {/each} {/if}