diff --git a/services/flow-engine/app/nodes/validation.py b/services/flow-engine/app/nodes/validation.py
new file mode 100644
index 0000000..777a39a
--- /dev/null
+++ b/services/flow-engine/app/nodes/validation.py
@@ -0,0 +1,82 @@
+import re
+from typing import Optional, Any
+from datetime import datetime
+from app.nodes.base import NodeExecutor
+from app.context import FlowContext
+
+
+class ValidateEmailExecutor(NodeExecutor):
+    async def execute(self, config: dict, context: FlowContext, session: Any) -> Optional[str]:
+        variable = config.get("variable", "")
+        value = context.get(variable) or context.message.get("content", "")
+        email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
+        if re.match(email_pattern, str(value).strip()):
+            return "valid"
+        return "invalid"
+
+
+class ValidatePhoneExecutor(NodeExecutor):
+    async def execute(self, config: dict, context: FlowContext, session: Any) -> Optional[str]:
+        variable = config.get("variable", "")
+        value = context.get(variable) or context.message.get("content", "")
+        cleaned = re.sub(r'[\s\-\(\)\+]', '', str(value))
+        if cleaned.isdigit() and 8 <= len(cleaned) <= 15:
+            return "valid"
+        return "invalid"
+
+
+class ValidateNumberExecutor(NodeExecutor):
+    async def execute(self, config: dict, context: FlowContext, session: Any) -> Optional[str]:
+        variable = config.get("variable", "")
+        value = context.get(variable) or context.message.get("content", "")
+        min_val = config.get("min")
+        max_val = config.get("max")
+        try:
+            num = float(str(value).strip())
+            if min_val is not None and num < float(min_val):
+                return "invalid"
+            if max_val is not None and num > float(max_val):
+                return "invalid"
+            return "valid"
+        except ValueError:
+            return "invalid"
+
+
+class ValidateDateExecutor(NodeExecutor):
+    async def execute(self, config: dict, context: FlowContext, session: Any) -> Optional[str]:
+        variable = config.get("variable", "")
+        value = context.get(variable) or context.message.get("content", "")
+        date_format = config.get("format", "%Y-%m-%d")
+        try:
+            datetime.strptime(str(value).strip(), date_format)
+            return "valid"
+        except ValueError:
+            return "invalid"
+
+
+class ValidateRegexExecutor(NodeExecutor):
+    async def execute(self, config: dict, context: FlowContext, session: Any) -> Optional[str]:
+        variable = config.get("variable", "")
+        value = context.get(variable) or context.message.get("content", "")
+        pattern = config.get("pattern", ".*")
+        try:
+            if re.match(pattern, str(value).strip()):
+                return "valid"
+            return "invalid"
+        except re.error:
+            return "invalid"
+
+
+class ValidateOptionsExecutor(NodeExecutor):
+    async def execute(self, config: dict, context: FlowContext, session: Any) -> Optional[str]:
+        variable = config.get("variable", "")
+        value = context.get(variable) or context.message.get("content", "")
+        options = config.get("options", [])
+        case_sensitive = config.get("case_sensitive", False)
+        value_str = str(value).strip()
+        if not case_sensitive:
+            value_str = value_str.lower()
+            options = [str(o).lower() for o in options]
+        if value_str in options:
+            return "valid"
+        return "invalid"