return clear 400 for missing required workflow params instead of 500 (#4380)

skyvern/forge/sdk/workflow/service.py

@@ -464,9 +464,13 @@ class WorkflowService:
         # 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)
         try:
+            missing_parameters: list[str] = []
             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]
+                    if self._is_missing_required_value(workflow_parameter, request_body_value):
+                        missing_parameters.append(workflow_parameter.key)
+                        continue
                     if workflow_parameter.workflow_parameter_type == WorkflowParameterType.CREDENTIAL_ID:
                         await self._validate_credential_id(str(request_body_value), organization)
                     try:
@@ -499,11 +503,15 @@ class WorkflowService:
                             reason=self._format_parameter_persistence_error(parameter_error),
                         ) from parameter_error
                 else:
-                    raise MissingValueForParameter(
-                        parameter_key=workflow_parameter.key,
-                        workflow_id=workflow.workflow_permanent_id,
-                        workflow_run_id=workflow_run.workflow_run_id,
-                    )
+                    missing_parameters.append(workflow_parameter.key)
+
+            if missing_parameters:
+                missing_list = ", ".join(sorted(missing_parameters))
+                raise MissingValueForParameter(
+                    parameter_key=missing_list,
+                    workflow_id=workflow.workflow_permanent_id,
+                    workflow_run_id=workflow_run.workflow_run_id,
+                )
         except Exception as e:
             LOG.exception(
                 f"Error while setting up workflow run {workflow_run.workflow_run_id}",
@@ -527,6 +535,53 @@ class WorkflowService:
             return "value cannot be null"
         return "database error while saving parameter value"
 
+    @staticmethod
+    def _is_missing_required_value(workflow_parameter: WorkflowParameter, value: Any) -> bool:
+        """
+        Determine if a provided value should be treated as missing for a required parameter.
+
+        Rules:
+        - None/null is always missing.
+        - String parameters may be empty strings (per UI behavior).
+        - JSON parameters treat empty/whitespace-only strings as missing.
+        - Boolean/integer/float parameters treat empty strings as missing.
+        - File URL treats empty strings, empty dicts, or dicts with empty s3uri as missing.
+        - Credential ID treats empty/whitespace-only strings as missing.
+        """
+
+        if value is None:
+            return True
+
+        param_type = workflow_parameter.workflow_parameter_type
+
+        if param_type == WorkflowParameterType.STRING:
+            return False
+
+        if param_type == WorkflowParameterType.JSON:
+            return isinstance(value, str) and value.strip() == ""
+
+        if param_type in (
+            WorkflowParameterType.BOOLEAN,
+            WorkflowParameterType.INTEGER,
+            WorkflowParameterType.FLOAT,
+        ):
+            return isinstance(value, str) and value.strip() == ""
+
+        if param_type == WorkflowParameterType.FILE_URL:
+            if isinstance(value, str):
+                return value.strip() == ""
+            if isinstance(value, dict):
+                if not value:
+                    return True
+                if "s3uri" in value:
+                    return not bool(value.get("s3uri"))
+            return False
+
+        if param_type == WorkflowParameterType.CREDENTIAL_ID:
+            return isinstance(value, str) and value.strip() == ""
+
+        return False
+
     async def auto_create_browser_session_if_needed(
         self,
         organization_id: str,