mirror of
https://github.com/johndoe6345789/metabuilder.git
synced 2026-04-25 06:14:59 +00:00
johndoe6345789
3d6ae4cbf7
Add full Python workflow execution engine with: Core Executor: - engine.py: WorkflowEngine for running n8n configs - n8n_executor.py: N8N-style workflow execution with connections - node_executor.py: Individual node execution with plugin dispatch - loop_executor.py: Loop node execution with iteration control - execution_order.py: Topological sort for node ordering Schema & Validation: - n8n_schema.py: N8N workflow schema types and validation - n8n_converter.py: Legacy to n8n schema conversion Plugin System: - plugin_loader.py: Dynamic plugin loading - plugin_registry.py: Plugin discovery and registration - plugin_map.json: 116 plugin type mappings Runtime & Context: - runtime.py: Workflow runtime container - input_resolver.py: Binding and coercion resolution - value_helpers.py: Value normalization helpers - workflow_context_builder.py: Runtime context assembly - workflow_config_loader.py: Configuration loading - workflow_engine_builder.py: Engine assembly with dependencies Utilities: - tool_calls_handler.py: LLM tool call handling - tool_runner.py: Tool execution with logging - notification_helpers.py: Slack/Discord notifications - workflow_adapter.py: N8N format handling - workflow_graph.py: Node/edge graph for visualization Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
"""N8N workflow schema types and validation."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Literal, Optional, Union
|
|
|
|
|
|
class N8NPosition:
|
|
"""Canvas position [x, y]."""
|
|
|
|
@staticmethod
|
|
def validate(value: Any) -> bool:
|
|
return (
|
|
isinstance(value, list) and
|
|
len(value) == 2 and
|
|
all(isinstance(v, (int, float)) for v in value)
|
|
)
|
|
|
|
|
|
class N8NConnectionTarget:
|
|
"""Connection target specification."""
|
|
|
|
@staticmethod
|
|
def validate(value: Any) -> bool:
|
|
if not isinstance(value, dict):
|
|
return False
|
|
return (
|
|
"node" in value and isinstance(value["node"], str) and
|
|
"type" in value and isinstance(value["type"], str) and
|
|
"index" in value and isinstance(value["index"], int) and value["index"] >= 0
|
|
)
|
|
|
|
|
|
class N8NNode:
|
|
"""N8N workflow node specification."""
|
|
|
|
@staticmethod
|
|
def validate(value: Any) -> bool:
|
|
if not isinstance(value, dict):
|
|
return False
|
|
required = ["id", "name", "type", "typeVersion", "position"]
|
|
if not all(key in value for key in required):
|
|
return False
|
|
if not isinstance(value["id"], str) or not value["id"]:
|
|
return False
|
|
if not isinstance(value["name"], str) or not value["name"]:
|
|
return False
|
|
if not isinstance(value["type"], str) or not value["type"]:
|
|
return False
|
|
if not isinstance(value["typeVersion"], (int, float)) or value["typeVersion"] < 1:
|
|
return False
|
|
if not N8NPosition.validate(value["position"]):
|
|
return False
|
|
return True
|
|
|
|
|
|
class N8NTrigger:
|
|
"""N8N workflow trigger specification."""
|
|
|
|
VALID_KINDS = ["webhook", "schedule", "queue", "email", "poll", "manual", "other"]
|
|
|
|
@staticmethod
|
|
def validate(value: Any) -> bool:
|
|
if not isinstance(value, dict):
|
|
return False
|
|
required = ["nodeId", "kind"]
|
|
if not all(key in value for key in required):
|
|
return False
|
|
if not isinstance(value["nodeId"], str) or not value["nodeId"]:
|
|
return False
|
|
if not isinstance(value["kind"], str) or value["kind"] not in N8NTrigger.VALID_KINDS:
|
|
return False
|
|
if "enabled" in value and not isinstance(value["enabled"], bool):
|
|
return False
|
|
if "meta" in value and not isinstance(value["meta"], dict):
|
|
return False
|
|
return True
|
|
|
|
|
|
class N8NWorkflow:
|
|
"""N8N workflow specification."""
|
|
|
|
@staticmethod
|
|
def validate(value: Any) -> bool:
|
|
if not isinstance(value, dict):
|
|
return False
|
|
required = ["name", "nodes", "connections"]
|
|
if not all(key in value for key in required):
|
|
return False
|
|
if not isinstance(value["name"], str) or not value["name"]:
|
|
return False
|
|
if not isinstance(value["nodes"], list) or len(value["nodes"]) < 1:
|
|
return False
|
|
if not isinstance(value["connections"], dict):
|
|
return False
|
|
if not all(N8NNode.validate(node) for node in value["nodes"]):
|
|
return False
|
|
# Validate triggers array if present
|
|
if "triggers" in value:
|
|
if not isinstance(value["triggers"], list):
|
|
return False
|
|
if not all(N8NTrigger.validate(trigger) for trigger in value["triggers"]):
|
|
return False
|
|
return True
|