feat: webhook trigger backend api (#24387)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/core/workflow/nodes/enums.py

@@ -25,6 +25,7 @@ class NodeType(StrEnum):
     DOCUMENT_EXTRACTOR = "document-extractor"
     LIST_OPERATOR = "list-operator"
     AGENT = "agent"
+    WEBHOOK = "webhook"
 
 
 class ErrorStrategy(StrEnum):

api/core/workflow/nodes/node_mapping.py

@@ -22,6 +22,7 @@ from core.workflow.nodes.tool import ToolNode
 from core.workflow.nodes.variable_aggregator import VariableAggregatorNode
 from core.workflow.nodes.variable_assigner.v1 import VariableAssignerNode as VariableAssignerNodeV1
 from core.workflow.nodes.variable_assigner.v2 import VariableAssignerNode as VariableAssignerNodeV2
+from core.workflow.nodes.webhook import WebhookNode
 
 LATEST_VERSION = "latest"
 
@@ -132,4 +133,8 @@ NODE_TYPE_CLASSES_MAPPING: Mapping[NodeType, Mapping[str, type[BaseNode]]] = {
         "2": AgentNode,
         "1": AgentNode,
     },
+    NodeType.WEBHOOK: {
+        LATEST_VERSION: WebhookNode,
+        "1": WebhookNode,
+    },
 }

api/core/workflow/nodes/webhook/__init__.py

@@ -0,0 +1,3 @@
+from .node import WebhookNode
+
+__all__ = ["WebhookNode"]

api/core/workflow/nodes/webhook/entities.py

@@ -0,0 +1,61 @@
+from collections.abc import Sequence
+from enum import StrEnum
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from core.workflow.nodes.base import BaseNodeData
+
+
+class Method(StrEnum):
+    GET = "get"
+    POST = "post"
+    HEAD = "head"
+    PATCH = "patch"
+    PUT = "put"
+    DELETE = "delete"
+
+
+class ContentType(StrEnum):
+    JSON = "application/json"
+    FORM_DATA = "multipart/form-data"
+    FORM_URLENCODED = "application/x-www-form-urlencoded"
+    TEXT = "text/plain"
+    FORM = "form"
+
+
+class WebhookParameter(BaseModel):
+    """Parameter definition for headers, query params, or body."""
+
+    name: str
+    required: bool = False
+
+
+class WebhookBodyParameter(BaseModel):
+    """Body parameter with type information."""
+
+    name: str
+    type: Literal["string", "number", "boolean", "object", "array", "file"] = "string"
+    required: bool = False
+
+
+class WebhookData(BaseNodeData):
+    """
+    Webhook Node Data.
+    """
+
+    class SyncMode(StrEnum):
+        SYNC = "async"  # only support
+
+    method: Method = Method.GET
+    content_type: ContentType = Field(alias="content-type", default=ContentType.JSON)
+    headers: Sequence[WebhookParameter] = Field(default_factory=list)
+    params: Sequence[WebhookParameter] = Field(default_factory=list)  # query parameters
+    body: Sequence[WebhookBodyParameter] = Field(default_factory=list)
+
+    status_code: int = 200  # Expected status code for response
+    response_body: str = ""  # Template for response body
+
+    # Webhook specific fields (not from client data, set internally)
+    webhook_id: Optional[str] = None  # Set when webhook trigger is created
+    timeout: int = 30  # Timeout in seconds to wait for webhook response

api/core/workflow/nodes/webhook/exc.py

@@ -0,0 +1,25 @@
+from core.workflow.nodes.base.exc import BaseNodeError
+
+
+class WebhookNodeError(BaseNodeError):
+    """Base webhook node error."""
+
+    pass
+
+
+class WebhookTimeoutError(WebhookNodeError):
+    """Webhook timeout error."""
+
+    pass
+
+
+class WebhookNotFoundError(WebhookNodeError):
+    """Webhook not found error."""
+
+    pass
+
+
+class WebhookConfigError(WebhookNodeError):
+    """Webhook configuration error."""
+
+    pass

api/core/workflow/nodes/webhook/node.py

@@ -0,0 +1,118 @@
+from collections.abc import Mapping
+from typing import Any, Optional
+
+from core.workflow.entities.node_entities import NodeRunResult
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionStatus
+from core.workflow.nodes.base import BaseNode
+from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig
+from core.workflow.nodes.enums import ErrorStrategy, NodeType
+
+from .entities import WebhookData
+
+
+class WebhookNode(BaseNode):
+    _node_type = NodeType.WEBHOOK
+
+    _node_data: WebhookData
+
+    def init_node_data(self, data: Mapping[str, Any]) -> None:
+        self._node_data = WebhookData.model_validate(data)
+
+    def _get_error_strategy(self) -> Optional[ErrorStrategy]:
+        return self._node_data.error_strategy
+
+    def _get_retry_config(self) -> RetryConfig:
+        return self._node_data.retry_config
+
+    def _get_title(self) -> str:
+        return self._node_data.title
+
+    def _get_description(self) -> Optional[str]:
+        return self._node_data.desc
+
+    def _get_default_value_dict(self) -> dict[str, Any]:
+        return self._node_data.default_value_dict
+
+    def get_base_node_data(self) -> BaseNodeData:
+        return self._node_data
+
+    @classmethod
+    def get_default_config(cls, filters: Optional[dict[str, Any]] = None) -> dict:
+        return {
+            "type": "webhook",
+            "config": {
+                "method": "get",
+                "content-type": "application/json",
+                "headers": [],
+                "params": [],
+                "body": [],
+                "async_mode": True,
+                "status_code": 200,
+                "response_body": "",
+                "timeout": 30,
+            },
+        }
+
+    @classmethod
+    def version(cls) -> str:
+        return "1"
+
+    def _run(self) -> NodeRunResult:
+        """
+        Run the webhook node.
+
+        Like the start node, this simply takes the webhook data from the variable pool
+        and makes it available to downstream nodes. The actual webhook handling
+        happens in the trigger controller.
+        """
+        # Get webhook data from variable pool (injected by Celery task)
+        webhook_inputs = dict(self.graph_runtime_state.variable_pool.user_inputs)
+
+        # Extract webhook-specific outputs based on node configuration
+        outputs = self._extract_configured_outputs(webhook_inputs)
+
+        return NodeRunResult(
+            status=WorkflowNodeExecutionStatus.SUCCEEDED,
+            inputs=webhook_inputs,
+            outputs=outputs,
+        )
+
+    def _extract_configured_outputs(self, webhook_inputs: dict[str, Any]) -> dict[str, Any]:
+        """Extract outputs based on node configuration from webhook inputs."""
+        outputs = {}
+
+        # Get the raw webhook data (should be injected by Celery task)
+        webhook_data = webhook_inputs.get("webhook_data", {})
+
+        # Extract configured headers (case-insensitive)
+        webhook_headers = webhook_data.get("headers", {})
+        webhook_headers_lower = {k.lower(): v for k, v in webhook_headers.items()}
+
+        for header in self._node_data.headers:
+            header_name = header.name
+            # Try exact match first, then case-insensitive match
+            value = webhook_headers.get(header_name) or webhook_headers_lower.get(header_name.lower())
+            outputs[header_name] = value
+
+        # Extract configured query parameters
+        for param in self._node_data.params:
+            param_name = param.name
+            outputs[param_name] = webhook_data.get("query_params", {}).get(param_name)
+
+        # Extract configured body parameters
+        for body_param in self._node_data.body:
+            param_name = body_param.name
+            param_type = body_param.type
+
+            if param_type == "file":
+                # Get File object (already processed by webhook controller)
+                file_obj = webhook_data.get("files", {}).get(param_name)
+                outputs[param_name] = file_obj
+            else:
+                # Get regular body parameter
+                outputs[param_name] = webhook_data.get("body", {}).get(param_name)
+
+        # Include raw webhook data for debugging/advanced use
+        outputs["_webhook_raw"] = webhook_data
+
+        return outputs