Files
metabuilder/workflow/executor/python/input_resolver.py
johndoe6345789 3d6ae4cbf7 feat: Add complete Python workflow executor from AutoMetabuilder
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>
2026-01-21 16:42:30 +00:00

23 lines
758 B
Python

"""Resolve workflow bindings and coercions."""
from .value_helpers import ValueHelpers
class InputResolver:
"""Resolve bindings in workflow inputs."""
def __init__(self, store: dict):
self.store = store
def resolve_inputs(self, inputs: dict) -> dict:
"""Resolve bindings for every input."""
return {key: self.resolve_binding(value) for key, value in (inputs or {}).items()}
def resolve_binding(self, value):
"""Resolve a single binding value."""
if isinstance(value, str) and value.startswith("$"):
return self.store.get(value[1:])
return value
def coerce_bool(self, value) -> bool:
"""Coerce values into booleans."""
return ValueHelpers.coerce_bool(value)