import json from collections import Counter from datetime import datetime import requests import structlog from skyvern import analytics from skyvern.exceptions import ( FailedToSendWebhook, MissingValueForParameter, WorkflowNotFound, WorkflowOrganizationMismatch, WorkflowRunNotFound, ) from skyvern.forge import app from skyvern.forge.sdk.artifact.models import ArtifactType from skyvern.forge.sdk.core import skyvern_context from skyvern.forge.sdk.core.security import generate_skyvern_signature from skyvern.forge.sdk.core.skyvern_context import SkyvernContext from skyvern.forge.sdk.models import Step from skyvern.forge.sdk.schemas.tasks import Task, TaskStatus from skyvern.forge.sdk.workflow.exceptions import WorkflowDefinitionHasDuplicateParameterKeys from skyvern.forge.sdk.workflow.models.block import ( BlockResult, BlockType, BlockTypeVar, CodeBlock, DownloadToS3Block, ForLoopBlock, SendEmailBlock, TaskBlock, TextPromptBlock, UploadToS3Block, ) from skyvern.forge.sdk.workflow.models.parameter import ( AWSSecretParameter, ContextParameter, OutputParameter, Parameter, ParameterType, WorkflowParameter, WorkflowParameterType, ) from skyvern.forge.sdk.workflow.models.workflow import ( Workflow, WorkflowDefinition, WorkflowRequestBody, WorkflowRun, WorkflowRunOutputParameter, WorkflowRunParameter, WorkflowRunStatus, WorkflowRunStatusResponse, ) from skyvern.forge.sdk.workflow.models.yaml import BLOCK_YAML_TYPES, WorkflowCreateYAMLRequest from skyvern.webeye.browser_factory import BrowserState LOG = structlog.get_logger() class WorkflowService: async def setup_workflow_run( self, request_id: str | None, workflow_request: WorkflowRequestBody, workflow_id: str, organization_id: str, max_steps_override: int | None = None, ) -> WorkflowRun: """ Create a workflow run and its parameters. Validate the workflow and the organization. If there are missing parameters with no default value, mark the workflow run as failed. :param request_id: The request id for the workflow run. :param workflow_request: The request body for the workflow run, containing the parameters and the config. :param workflow_id: The workflow id to run. :param organization_id: The organization id for the workflow. :param max_steps_override: The max steps override for the workflow run, if any. :return: The created workflow run. """ # Validate the workflow and the organization workflow = await self.get_workflow(workflow_id=workflow_id) if workflow is None: LOG.error(f"Workflow {workflow_id} not found") raise WorkflowNotFound(workflow_id=workflow_id) if workflow.organization_id != organization_id: LOG.error(f"Workflow {workflow_id} does not belong to organization {organization_id}") raise WorkflowOrganizationMismatch(workflow_id=workflow_id, organization_id=organization_id) # Create the workflow run and set skyvern context workflow_run = await self.create_workflow_run(workflow_request=workflow_request, workflow_id=workflow_id) LOG.info( f"Created workflow run {workflow_run.workflow_run_id} for workflow {workflow.workflow_id}", request_id=request_id, workflow_run_id=workflow_run.workflow_run_id, workflow_id=workflow.workflow_id, proxy_location=workflow_request.proxy_location, ) skyvern_context.set( SkyvernContext( organization_id=organization_id, request_id=request_id, workflow_id=workflow_id, workflow_run_id=workflow_run.workflow_run_id, max_steps_override=max_steps_override, ) ) # Create all the workflow run parameters, AWSSecretParameter won't have workflow run parameters created. all_workflow_parameters = await self.get_workflow_parameters(workflow_id=workflow.workflow_id) workflow_run_parameters = [] for workflow_parameter in all_workflow_parameters: if workflow_request.data and workflow_parameter.key in workflow_request.data: request_body_value = workflow_request.data[workflow_parameter.key] workflow_run_parameter = await self.create_workflow_run_parameter( workflow_run_id=workflow_run.workflow_run_id, workflow_parameter_id=workflow_parameter.workflow_parameter_id, value=request_body_value, ) elif workflow_parameter.default_value is not None: workflow_run_parameter = await self.create_workflow_run_parameter( workflow_run_id=workflow_run.workflow_run_id, workflow_parameter_id=workflow_parameter.workflow_parameter_id, value=workflow_parameter.default_value, ) else: await self.mark_workflow_run_as_failed(workflow_run_id=workflow_run.workflow_run_id) raise MissingValueForParameter( parameter_key=workflow_parameter.key, workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, ) workflow_run_parameters.append(workflow_run_parameter) return workflow_run async def execute_workflow( self, workflow_run_id: str, api_key: str, ) -> WorkflowRun: """Execute a workflow.""" workflow_run = await self.get_workflow_run(workflow_run_id=workflow_run_id) workflow = await self.get_workflow(workflow_id=workflow_run.workflow_id) # Set workflow run status to running, create workflow run parameters await self.mark_workflow_run_as_running(workflow_run_id=workflow_run.workflow_run_id) # Get all tuples wp_wps_tuples = await self.get_workflow_run_parameter_tuples(workflow_run_id=workflow_run.workflow_run_id) workflow_output_parameters = await self.get_workflow_output_parameters(workflow_id=workflow.workflow_id) app.WORKFLOW_CONTEXT_MANAGER.initialize_workflow_run_context( workflow_run_id, wp_wps_tuples, workflow_output_parameters ) # Execute workflow blocks blocks = workflow.workflow_definition.blocks block_result = None for block_idx, block in enumerate(blocks): try: parameters = block.get_all_parameters(workflow_run_id) await app.WORKFLOW_CONTEXT_MANAGER.register_block_parameters_for_workflow_run( workflow_run_id, parameters ) LOG.info( f"Executing root block {block.block_type} at index {block_idx} for workflow run {workflow_run_id}", block_type=block.block_type, workflow_run_id=workflow_run.workflow_run_id, block_idx=block_idx, ) block_result = await block.execute_safe(workflow_run_id=workflow_run_id) if not block_result.success: LOG.error( f"Block with type {block.block_type} at index {block_idx} failed for workflow run {workflow_run_id}", block_type=block.block_type, workflow_run_id=workflow_run.workflow_run_id, block_idx=block_idx, block_result=block_result, ) await self.mark_workflow_run_as_failed(workflow_run_id=workflow_run.workflow_run_id) break except Exception as e: LOG.exception( f"Error while executing workflow run {workflow_run.workflow_run_id}", workflow_run_id=workflow_run.workflow_run_id, exc_info=True, ) await self.mark_workflow_run_as_failed(workflow_run_id=workflow_run.workflow_run_id) raise e tasks = await self.get_tasks_by_workflow_run_id(workflow_run.workflow_run_id) await self.mark_workflow_run_as_completed(workflow_run_id=workflow_run.workflow_run_id) await self.send_workflow_response( workflow=workflow, workflow_run=workflow_run, tasks=tasks, # TODO: We don't persist the block result for now, but we should in the case the users want to get it later last_block_result=block_result, api_key=api_key, ) return workflow_run async def handle_workflow_status(self, workflow_run: WorkflowRun, tasks: list[Task]) -> WorkflowRun: task_counts_by_status = Counter(task.status for task in tasks) # Create a mapping of status to (action, log_func, log_message) status_action_mapping = { TaskStatus.running: (None, LOG.error, "has running tasks, this should not happen"), TaskStatus.terminated: ( self.mark_workflow_run_as_terminated, LOG.warning, "has terminated tasks, marking as terminated", ), TaskStatus.failed: (self.mark_workflow_run_as_failed, LOG.warning, "has failed tasks, marking as failed"), TaskStatus.completed: ( self.mark_workflow_run_as_completed, LOG.info, "tasks are completed, marking as completed", ), } for status, (action, log_func, log_message) in status_action_mapping.items(): if task_counts_by_status.get(status, 0) > 0: if action is not None: await action(workflow_run_id=workflow_run.workflow_run_id) if log_func and log_message: log_func( f"Workflow run {workflow_run.workflow_run_id} {log_message}", workflow_run_id=workflow_run.workflow_run_id, task_counts_by_status=task_counts_by_status, ) return workflow_run # Handle unexpected state LOG.error( f"Workflow run {workflow_run.workflow_run_id} has tasks in an unexpected state, marking as failed", workflow_run_id=workflow_run.workflow_run_id, task_counts_by_status=task_counts_by_status, ) await self.mark_workflow_run_as_failed(workflow_run_id=workflow_run.workflow_run_id) return workflow_run async def create_workflow( self, organization_id: str, title: str, workflow_definition: WorkflowDefinition, description: str | None = None, ) -> Workflow: return await app.DATABASE.create_workflow( organization_id=organization_id, title=title, description=description, workflow_definition=workflow_definition.model_dump() if workflow_definition else None, ) async def get_workflow(self, workflow_id: str) -> Workflow: workflow = await app.DATABASE.get_workflow(workflow_id=workflow_id) if not workflow: raise WorkflowNotFound(workflow_id) return workflow async def update_workflow( self, workflow_id: str, title: str | None = None, description: str | None = None, workflow_definition: WorkflowDefinition | None = None, ) -> Workflow: if workflow_definition: workflow_definition.validate() return await app.DATABASE.update_workflow( workflow_id=workflow_id, title=title, description=description, workflow_definition=workflow_definition.model_dump() if workflow_definition else None, ) async def create_workflow_run(self, workflow_request: WorkflowRequestBody, workflow_id: str) -> WorkflowRun: return await app.DATABASE.create_workflow_run( workflow_id=workflow_id, proxy_location=workflow_request.proxy_location, webhook_callback_url=workflow_request.webhook_callback_url, ) async def mark_workflow_run_as_completed(self, workflow_run_id: str) -> None: LOG.info( f"Marking workflow run {workflow_run_id} as completed", workflow_run_id=workflow_run_id, status="completed" ) await app.DATABASE.update_workflow_run( workflow_run_id=workflow_run_id, status=WorkflowRunStatus.completed, ) async def mark_workflow_run_as_failed(self, workflow_run_id: str) -> None: LOG.info(f"Marking workflow run {workflow_run_id} as failed", workflow_run_id=workflow_run_id, status="failed") await app.DATABASE.update_workflow_run( workflow_run_id=workflow_run_id, status=WorkflowRunStatus.failed, ) async def mark_workflow_run_as_running(self, workflow_run_id: str) -> None: LOG.info( f"Marking workflow run {workflow_run_id} as running", workflow_run_id=workflow_run_id, status="running" ) await app.DATABASE.update_workflow_run( workflow_run_id=workflow_run_id, status=WorkflowRunStatus.running, ) async def mark_workflow_run_as_terminated(self, workflow_run_id: str) -> None: LOG.info( f"Marking workflow run {workflow_run_id} as terminated", workflow_run_id=workflow_run_id, status="terminated", ) await app.DATABASE.update_workflow_run( workflow_run_id=workflow_run_id, status=WorkflowRunStatus.terminated, ) async def get_workflow_runs(self, workflow_id: str) -> list[WorkflowRun]: return await app.DATABASE.get_workflow_runs(workflow_id=workflow_id) async def get_workflow_run(self, workflow_run_id: str) -> WorkflowRun: workflow_run = await app.DATABASE.get_workflow_run(workflow_run_id=workflow_run_id) if not workflow_run: raise WorkflowRunNotFound(workflow_run_id) return workflow_run async def create_workflow_parameter( self, workflow_id: str, workflow_parameter_type: WorkflowParameterType, key: str, default_value: bool | int | float | str | dict | list | None = None, description: str | None = None, ) -> WorkflowParameter: return await app.DATABASE.create_workflow_parameter( workflow_id=workflow_id, workflow_parameter_type=workflow_parameter_type, key=key, description=description, default_value=default_value, ) async def create_aws_secret_parameter( self, workflow_id: str, aws_key: str, key: str, description: str | None = None ) -> AWSSecretParameter: return await app.DATABASE.create_aws_secret_parameter( workflow_id=workflow_id, aws_key=aws_key, key=key, description=description ) async def create_bitwarden_login_credential_parameter( self, workflow_id: str, bitwarden_client_id_aws_secret_key: str, bitwarden_client_secret_aws_secret_key: str, bitwarden_master_password_aws_secret_key: str, url_parameter_key: str, key: str, description: str | None = None, ) -> Parameter: return await app.DATABASE.create_bitwarden_login_credential_parameter( workflow_id=workflow_id, bitwarden_client_id_aws_secret_key=bitwarden_client_id_aws_secret_key, bitwarden_client_secret_aws_secret_key=bitwarden_client_secret_aws_secret_key, bitwarden_master_password_aws_secret_key=bitwarden_master_password_aws_secret_key, url_parameter_key=url_parameter_key, key=key, description=description, ) async def create_output_parameter( self, workflow_id: str, key: str, description: str | None = None ) -> OutputParameter: return await app.DATABASE.create_output_parameter(workflow_id=workflow_id, key=key, description=description) async def get_workflow_parameters(self, workflow_id: str) -> list[WorkflowParameter]: return await app.DATABASE.get_workflow_parameters(workflow_id=workflow_id) async def create_workflow_run_parameter( self, workflow_run_id: str, workflow_parameter_id: str, value: bool | int | float | str | dict | list, ) -> WorkflowRunParameter: return await app.DATABASE.create_workflow_run_parameter( workflow_run_id=workflow_run_id, workflow_parameter_id=workflow_parameter_id, value=json.dumps(value) if isinstance(value, (dict, list)) else value, ) async def get_workflow_run_parameter_tuples( self, workflow_run_id: str ) -> list[tuple[WorkflowParameter, WorkflowRunParameter]]: return await app.DATABASE.get_workflow_run_parameters(workflow_run_id=workflow_run_id) @staticmethod async def get_workflow_output_parameters(workflow_id: str) -> list[OutputParameter]: return await app.DATABASE.get_workflow_output_parameters(workflow_id=workflow_id) @staticmethod async def get_workflow_run_output_parameters( workflow_run_id: str, ) -> list[WorkflowRunOutputParameter]: return await app.DATABASE.get_workflow_run_output_parameters(workflow_run_id=workflow_run_id) @staticmethod async def get_output_parameter_workflow_run_output_parameter_tuples( workflow_id: str, workflow_run_id: str, ) -> list[tuple[OutputParameter, WorkflowRunOutputParameter]]: workflow_run_output_parameters = await app.DATABASE.get_workflow_run_output_parameters( workflow_run_id=workflow_run_id ) output_parameters = await app.DATABASE.get_workflow_output_parameters(workflow_id=workflow_id) return [ (output_parameter, workflow_run_output_parameter) for output_parameter in output_parameters for workflow_run_output_parameter in workflow_run_output_parameters if output_parameter.output_parameter_id == workflow_run_output_parameter.output_parameter_id ] async def get_last_task_for_workflow_run(self, workflow_run_id: str) -> Task | None: return await app.DATABASE.get_last_task_for_workflow_run(workflow_run_id=workflow_run_id) async def get_tasks_by_workflow_run_id(self, workflow_run_id: str) -> list[Task]: return await app.DATABASE.get_tasks_by_workflow_run_id(workflow_run_id=workflow_run_id) async def build_workflow_run_status_response( self, workflow_id: str, workflow_run_id: str, last_block_result: BlockResult | None, organization_id: str ) -> WorkflowRunStatusResponse: workflow = await self.get_workflow(workflow_id=workflow_id) if workflow is None: LOG.error(f"Workflow {workflow_id} not found") raise WorkflowNotFound(workflow_id=workflow_id) if workflow.organization_id != organization_id: LOG.error(f"Workflow {workflow_id} does not belong to organization {organization_id}") raise WorkflowOrganizationMismatch(workflow_id=workflow_id, organization_id=organization_id) workflow_run = await self.get_workflow_run(workflow_run_id=workflow_run_id) workflow_run_tasks = await app.DATABASE.get_tasks_by_workflow_run_id(workflow_run_id=workflow_run_id) screenshot_urls = [] # get the last screenshot for the last 3 tasks of the workflow run for task in workflow_run_tasks[::-1]: screenshot_artifact = await app.DATABASE.get_latest_artifact( task_id=task.task_id, artifact_types=[ArtifactType.SCREENSHOT_ACTION, ArtifactType.SCREENSHOT_FINAL], organization_id=organization_id, ) if screenshot_artifact: screenshot_url = await app.ARTIFACT_MANAGER.get_share_link(screenshot_artifact) if screenshot_url: screenshot_urls.append(screenshot_url) if len(screenshot_urls) >= 3: break recording_url = None recording_artifact = await app.DATABASE.get_artifact_for_workflow_run( workflow_run_id=workflow_run_id, artifact_type=ArtifactType.RECORDING, organization_id=organization_id ) if recording_artifact: recording_url = await app.ARTIFACT_MANAGER.get_share_link(recording_artifact) workflow_parameter_tuples = await app.DATABASE.get_workflow_run_parameters(workflow_run_id=workflow_run_id) parameters_with_value = {wfp.key: wfrp.value for wfp, wfrp in workflow_parameter_tuples} output_parameter_tuples: list[ tuple[OutputParameter, WorkflowRunOutputParameter] ] = await self.get_output_parameter_workflow_run_output_parameter_tuples( workflow_id=workflow_id, workflow_run_id=workflow_run_id ) if output_parameter_tuples: payload = { output_parameter.key: wfrp.value for output_parameter, wfrp in output_parameter_tuples if wfrp.value is not None } else: payload = { task.task_id: { "title": task.title, "extracted_information": task.extracted_information, "navigation_payload": task.navigation_payload, "errors": await app.agent.get_task_errors(task=task), } for task in workflow_run_tasks } return WorkflowRunStatusResponse( workflow_id=workflow_id, workflow_run_id=workflow_run_id, status=workflow_run.status, proxy_location=workflow_run.proxy_location, webhook_callback_url=workflow_run.webhook_callback_url, created_at=workflow_run.created_at, modified_at=workflow_run.modified_at, parameters=parameters_with_value, screenshot_urls=screenshot_urls, recording_url=recording_url, payload=payload, output=last_block_result, ) async def send_workflow_response( self, workflow: Workflow, workflow_run: WorkflowRun, tasks: list[Task], last_block_result: BlockResult | None, api_key: str | None = None, close_browser_on_completion: bool = True, ) -> None: analytics.capture("skyvern-oss-agent-workflow-status", {"status": workflow_run.status}) all_workflow_task_ids = [task.task_id for task in tasks] browser_state = await app.BROWSER_MANAGER.cleanup_for_workflow_run( workflow_run.workflow_run_id, all_workflow_task_ids, close_browser_on_completion ) if browser_state: await self.persist_video_data(browser_state, workflow, workflow_run) await self.persist_debug_artifacts(browser_state, tasks[-1], workflow, workflow_run) await app.ARTIFACT_MANAGER.wait_for_upload_aiotasks_for_tasks(all_workflow_task_ids) workflow_run_status_response = await self.build_workflow_run_status_response( workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, last_block_result=last_block_result, organization_id=workflow.organization_id, ) LOG.info("Built workflow run status response", workflow_run_status_response=workflow_run_status_response) if not workflow_run.webhook_callback_url: LOG.warning( "Workflow has no webhook callback url. Not sending workflow response", workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, ) return if not api_key: LOG.warning( "Request has no api key. Not sending workflow response", workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, ) return # send task_response to the webhook callback url # TODO: use async requests (httpx) timestamp = str(int(datetime.utcnow().timestamp())) payload = workflow_run_status_response.model_dump_json() signature = generate_skyvern_signature( payload=payload, api_key=api_key, ) headers = { "x-skyvern-timestamp": timestamp, "x-skyvern-signature": signature, "Content-Type": "application/json", } LOG.info( "Sending webhook run status to webhook callback url", workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, webhook_callback_url=workflow_run.webhook_callback_url, payload=payload, headers=headers, ) try: resp = requests.post(workflow_run.webhook_callback_url, data=payload, headers=headers) if resp.ok: LOG.info( "Webhook sent successfully", workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, resp_code=resp.status_code, resp_text=resp.text, ) else: LOG.info( "Webhook failed", workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, resp=resp, resp_code=resp.status_code, resp_text=resp.text, resp_json=resp.json(), ) except Exception as e: raise FailedToSendWebhook( workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id ) from e async def persist_video_data( self, browser_state: BrowserState, workflow: Workflow, workflow_run: WorkflowRun ) -> None: # Create recording artifact after closing the browser, so we can get an accurate recording video_data = await app.BROWSER_MANAGER.get_video_data( workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, browser_state=browser_state, ) if video_data: await app.ARTIFACT_MANAGER.update_artifact_data( artifact_id=browser_state.browser_artifacts.video_artifact_id, organization_id=workflow.organization_id, data=video_data, ) async def persist_har_data( self, browser_state: BrowserState, last_step: Step, workflow: Workflow, workflow_run: WorkflowRun ) -> None: har_data = await app.BROWSER_MANAGER.get_har_data( workflow_id=workflow.workflow_id, workflow_run_id=workflow_run.workflow_run_id, browser_state=browser_state ) if har_data: await app.ARTIFACT_MANAGER.create_artifact( step=last_step, artifact_type=ArtifactType.HAR, data=har_data, ) async def persist_tracing_data( self, browser_state: BrowserState, last_step: Step, workflow_run: WorkflowRun ) -> None: if browser_state.browser_context is None or browser_state.browser_artifacts.traces_dir is None: return trace_path = f"{browser_state.browser_artifacts.traces_dir}/{workflow_run.workflow_run_id}.zip" await app.ARTIFACT_MANAGER.create_artifact(step=last_step, artifact_type=ArtifactType.TRACE, path=trace_path) async def persist_debug_artifacts( self, browser_state: BrowserState, last_task: Task, workflow: Workflow, workflow_run: WorkflowRun ) -> None: last_step = await app.DATABASE.get_latest_step( task_id=last_task.task_id, organization_id=last_task.organization_id ) if not last_step: return await self.persist_har_data(browser_state, last_step, workflow, workflow_run) await self.persist_tracing_data(browser_state, last_step, workflow_run) async def create_workflow_from_request(self, organization_id: str, request: WorkflowCreateYAMLRequest) -> Workflow: LOG.info("Creating workflow from request", organization_id=organization_id, title=request.title) try: workflow = await self.create_workflow( organization_id=organization_id, title=request.title, description=request.description, workflow_definition=WorkflowDefinition(blocks=[]), ) # Create parameters from the request parameters = {} duplicate_parameter_keys = set() # We're going to process context parameters after other parameters since they depend on the other parameters context_parameter_yamls = [] for parameter in request.workflow_definition.parameters: if parameter.key in parameters: LOG.error(f"Duplicate parameter key {parameter.key}") duplicate_parameter_keys.add(parameter.key) continue if parameter.parameter_type == ParameterType.AWS_SECRET: parameters[parameter.key] = await self.create_aws_secret_parameter( workflow_id=workflow.workflow_id, aws_key=parameter.aws_key, key=parameter.key, description=parameter.description, ) elif parameter.parameter_type == ParameterType.BITWARDEN_LOGIN_CREDENTIAL: parameters[parameter.key] = await self.create_bitwarden_login_credential_parameter( workflow_id=workflow.workflow_id, bitwarden_client_id_aws_secret_key=parameter.bitwarden_client_id_aws_secret_key, bitwarden_client_secret_aws_secret_key=parameter.bitwarden_client_secret_aws_secret_key, bitwarden_master_password_aws_secret_key=parameter.bitwarden_master_password_aws_secret_key, url_parameter_key=parameter.url_parameter_key, key=parameter.key, description=parameter.description, ) elif parameter.parameter_type == ParameterType.WORKFLOW: parameters[parameter.key] = await self.create_workflow_parameter( workflow_id=workflow.workflow_id, workflow_parameter_type=parameter.workflow_parameter_type, key=parameter.key, default_value=parameter.default_value, description=parameter.description, ) elif parameter.parameter_type == ParameterType.OUTPUT: parameters[parameter.key] = await self.create_output_parameter( workflow_id=workflow.workflow_id, key=parameter.key, description=parameter.description, ) elif parameter.parameter_type == ParameterType.CONTEXT: context_parameter_yamls.append(parameter) else: LOG.error(f"Invalid parameter type {parameter.parameter_type}") # Now we can process the context parameters since all other parameters have been created for context_parameter in context_parameter_yamls: parameters[context_parameter.key] = ContextParameter( key=context_parameter.key, description=context_parameter.description, source=parameters[context_parameter.source_workflow_parameter_key], # Context parameters don't have a default value, the value always depends on the source parameter value=None, ) if duplicate_parameter_keys: raise WorkflowDefinitionHasDuplicateParameterKeys(duplicate_keys=duplicate_parameter_keys) # Create blocks from the request block_label_mapping = {} blocks = [] for block_yaml in request.workflow_definition.blocks: block = await self.block_yaml_to_block(block_yaml, parameters) blocks.append(block) block_label_mapping[block.label] = block # Set the blocks for the workflow definition workflow_definition = WorkflowDefinition(blocks=blocks) workflow = await self.update_workflow( workflow_id=workflow.workflow_id, workflow_definition=workflow_definition, ) LOG.info( f"Created workflow from request, title: {request.title}", parameter_keys=[parameter.key for parameter in parameters.values()], block_labels=[block.label for block in blocks], organization_id=organization_id, title=request.title, workflow_id=workflow.workflow_id, ) return workflow except Exception as e: LOG.exception(f"Failed to create workflow from request, title: {request.title}") raise e @staticmethod async def block_yaml_to_block(block_yaml: BLOCK_YAML_TYPES, parameters: dict[str, Parameter]) -> BlockTypeVar: output_parameter = parameters.get(block_yaml.output_parameter_key) if block_yaml.output_parameter_key else None if block_yaml.block_type == BlockType.TASK: task_block_parameters = ( [parameters[parameter_key] for parameter_key in block_yaml.parameter_keys] if block_yaml.parameter_keys else [] ) return TaskBlock( label=block_yaml.label, url=block_yaml.url, title=block_yaml.title, parameters=task_block_parameters, output_parameter=output_parameter, navigation_goal=block_yaml.navigation_goal, data_extraction_goal=block_yaml.data_extraction_goal, data_schema=block_yaml.data_schema, error_code_mapping=block_yaml.error_code_mapping, max_retries=block_yaml.max_retries, ) elif block_yaml.block_type == BlockType.FOR_LOOP: loop_blocks = [ await WorkflowService.block_yaml_to_block(loop_block, parameters) for loop_block in block_yaml.loop_blocks ] loop_over_parameter = parameters[block_yaml.loop_over_parameter_key] return ForLoopBlock( label=block_yaml.label, loop_over=loop_over_parameter, loop_blocks=loop_blocks, output_parameter=output_parameter, ) elif block_yaml.block_type == BlockType.CODE: return CodeBlock( label=block_yaml.label, code=block_yaml.code, parameters=[parameters[parameter_key] for parameter_key in block_yaml.parameter_keys] if block_yaml.parameter_keys else [], output_parameter=output_parameter, ) elif block_yaml.block_type == BlockType.TEXT_PROMPT: return TextPromptBlock( label=block_yaml.label, llm_key=block_yaml.llm_key, prompt=block_yaml.prompt, parameters=[parameters[parameter_key] for parameter_key in block_yaml.parameter_keys] if block_yaml.parameter_keys else [], json_schema=block_yaml.json_schema, output_parameter=output_parameter, ) elif block_yaml.block_type == BlockType.DOWNLOAD_TO_S3: return DownloadToS3Block( label=block_yaml.label, output_parameter=output_parameter, url=block_yaml.url, ) elif block_yaml.block_type == BlockType.UPLOAD_TO_S3: return UploadToS3Block( label=block_yaml.label, output_parameter=output_parameter, path=block_yaml.path, ) elif block_yaml.block_type == BlockType.SEND_EMAIL: return SendEmailBlock( label=block_yaml.label, output_parameter=output_parameter, smtp_host=parameters[block_yaml.smtp_host_secret_parameter_key], smtp_port=parameters[block_yaml.smtp_port_secret_parameter_key], smtp_username=parameters[block_yaml.smtp_username_secret_parameter_key], smtp_password=parameters[block_yaml.smtp_password_secret_parameter_key], sender=block_yaml.sender, recipients=block_yaml.recipients, subject=block_yaml.subject, body=block_yaml.body, file_attachments=block_yaml.file_attachments or [], ) raise ValueError(f"Invalid block type {block_yaml.block_type}")