feat: start node support json schema (#29053)

api/core/app/app_config/entities.py

@@ -2,6 +2,7 @@ from collections.abc import Sequence
 from enum import StrEnum, auto
 from typing import Any, Literal
 
+from jsonschema import Draft7Validator, SchemaError
 from pydantic import BaseModel, Field, field_validator
 
 from core.file import FileTransferMethod, FileType, FileUploadConfig
@@ -98,6 +99,7 @@ class VariableEntityType(StrEnum):
     FILE = "file"
     FILE_LIST = "file-list"
     CHECKBOX = "checkbox"
+    JSON_OBJECT = "json_object"
 
 
 class VariableEntity(BaseModel):
@@ -118,6 +120,7 @@ class VariableEntity(BaseModel):
     allowed_file_types: Sequence[FileType] | None = Field(default_factory=list)
     allowed_file_extensions: Sequence[str] | None = Field(default_factory=list)
     allowed_file_upload_methods: Sequence[FileTransferMethod] | None = Field(default_factory=list)
+    json_schema: dict[str, Any] | None = Field(default=None)
 
     @field_validator("description", mode="before")
     @classmethod
@@ -129,6 +132,17 @@ class VariableEntity(BaseModel):
     def convert_none_options(cls, v: Any) -> Sequence[str]:
         return v or []
 
+    @field_validator("json_schema")
+    @classmethod
+    def validate_json_schema(cls, schema: dict[str, Any] | None) -> dict[str, Any] | None:
+        if schema is None:
+            return None
+        try:
+            Draft7Validator.check_schema(schema)
+        except SchemaError as e:
+            raise ValueError(f"Invalid JSON schema: {e.message}")
+        return schema
+
 
 class RagPipelineVariableEntity(VariableEntity):
     """

api/core/workflow/nodes/start/start_node.py

@@ -1,3 +1,8 @@
+from typing import Any
+
+from jsonschema import Draft7Validator, ValidationError
+
+from core.app.app_config.entities import VariableEntityType
 from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.enums import NodeExecutionType, NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
@@ -15,6 +20,7 @@ class StartNode(Node[StartNodeData]):
 
     def _run(self) -> NodeRunResult:
         node_inputs = dict(self.graph_runtime_state.variable_pool.user_inputs)
+        self._validate_and_normalize_json_object_inputs(node_inputs)
         system_inputs = self.graph_runtime_state.variable_pool.system_variables.to_dict()
 
         # TODO: System variables should be directly accessible, no need for special handling
@@ -24,3 +30,27 @@ class StartNode(Node[StartNodeData]):
         outputs = dict(node_inputs)
 
         return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED, inputs=node_inputs, outputs=outputs)
+
+    def _validate_and_normalize_json_object_inputs(self, node_inputs: dict[str, Any]) -> None:
+        for variable in self.node_data.variables:
+            if variable.type != VariableEntityType.JSON_OBJECT:
+                continue
+
+            key = variable.variable
+            value = node_inputs.get(key)
+
+            if value is None and variable.required:
+                raise ValueError(f"{key} is required in input form")
+
+            if not isinstance(value, dict):
+                raise ValueError(f"{key} must be a JSON object")
+
+            schema = variable.json_schema
+            if not schema:
+                continue
+
+            try:
+                Draft7Validator(schema).validate(value)
+            except ValidationError as e:
+                raise ValueError(f"JSON object for '{key}' does not match schema: {e.message}")
+            node_inputs[key] = value