feat(engine): add interpolatable field support for dynamic type resolution

Introduces a new interpolation pattern that allows block input fields to accept
both their strict types (enums, booleans, integers) and string variables that
get resolved at execution time. This enables dynamic workflow behavior while
maintaining type safety through two-phase validation:

1. Load time: Accept Union[StrictType, str], validate interpolation syntax
2. Execution time: Resolve variables, validate against strict type

Key additions:
- New interpolation.py module (588 lines) with reusable validators and resolvers
- Interpolatable enum, boolean, and numeric field support
- Updated executors (Shell, CreateFile, HttpCall, LLMCall, State) with interpolation
- LLMProvider enum type for better type safety
- Comprehensive validation with clear error messages

This change enables workflows to use variables in strictly-typed fields:
  provider: "{{inputs.llm_provider}}"  # Resolves to LLMProvider enum
  timeout: "{{blocks.config.outputs.timeout}}"  # Resolves to int
  verify_ssl: "{{inputs.ssl_enabled}}"  # Resolves to bool

Author: ibacalu

schema.json

@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import Any, ClassVar
 
-from pydantic import Field
+from pydantic import Field, field_validator
 
 from .block import BlockInput, BlockOutput
 from .context_vars import block_custom_outputs
@@ -17,6 +17,10 @@
     ExecutorCapabilities,
     ExecutorSecurityLevel,
 )
+from .interpolation import (
+    interpolatable_numeric_validator,
+    resolve_interpolatable_numeric,
+)
 
 # ============================================================================
 # Shell Executor
@@ -36,7 +40,9 @@ class ShellInput(BlockInput):
 
     command: str = Field(description="Shell command to execute")
     working_dir: str = Field(default="", description="Working directory (empty = current dir)")
-    timeout: int = Field(default=120, description="Timeout in seconds")
+    timeout: int | str = Field(
+        default=120, description="Timeout in seconds (or interpolation string)"
+    )
     env: dict[str, str] = Field(default_factory=dict, description="Environment variables")
     capture_output: bool = Field(default=True, description="Capture stdout/stderr")
     shell: bool = Field(default=True, description="Execute via shell")
@@ -46,6 +52,11 @@ class ShellInput(BlockInput):
         exclude=True,
     )
 
+    # Validator for numeric field with interpolation support
+    _validate_timeout = field_validator("timeout", mode="before")(
+        interpolatable_numeric_validator(int, ge=1, le=3600)
+    )
+
 
 class ShellOutput(BlockOutput):
     """Output model for Shell executor.
@@ -361,6 +372,9 @@ async def execute(  # type: ignore[override]
             TimeoutError: If command times out
             Exception: For other execution failures
         """
+        # Resolve interpolatable fields to their actual types
+        timeout = resolve_interpolatable_numeric(inputs.timeout, int, "timeout", ge=1, le=3600)
+
         # Prepare working directory
         cwd = Path(inputs.working_dir) if inputs.working_dir else Path.cwd()
         if not cwd.exists():
@@ -408,14 +422,12 @@ async def execute(  # type: ignore[override]
         # Wait for completion with timeout
         try:
             stdout_bytes, stderr_bytes = await asyncio.wait_for(
-                process.communicate(), timeout=inputs.timeout
+                process.communicate(), timeout=timeout
             )
         except TimeoutError:
             process.kill()
             await process.wait()
-            raise TimeoutError(
-                f"Command timed out after {inputs.timeout} seconds: {inputs.command}"
-            )
+            raise TimeoutError(f"Command timed out after {timeout} seconds: {inputs.command}")
 
         # Decode output
         stdout = stdout_bytes.decode("utf-8") if stdout_bytes else ""

src/workflows_mcp/engine/executors_core.py

@@ -12,7 +12,7 @@
 from typing import Any, ClassVar
 
 from jinja2 import Environment, StrictUndefined
-from pydantic import Field
+from pydantic import Field, field_validator
 
 from .block import BlockInput, BlockOutput
 from .block_utils import FileOperations, PathResolver
@@ -22,6 +22,10 @@
     ExecutorCapabilities,
     ExecutorSecurityLevel,
 )
+from .interpolation import (
+    interpolatable_boolean_validator,
+    resolve_interpolatable_boolean,
+)
 
 # ============================================================================
 # CreateFile Executor
@@ -38,8 +42,21 @@ class CreateFileInput(BlockInput):
         default=None,
         description="File permissions (Unix only, e.g., 0o644, 644, or '644')",
     )
-    overwrite: bool = Field(default=True, description="Whether to overwrite existing file")
-    create_parents: bool = Field(default=True, description="Create parent directories if missing")
+    overwrite: bool | str = Field(
+        default=True, description="Whether to overwrite existing file (or interpolation string)"
+    )
+    create_parents: bool | str = Field(
+        default=True,
+        description="Create parent directories if missing (or interpolation string)",
+    )
+
+    # Validators for boolean fields with interpolation support
+    _validate_overwrite = field_validator("overwrite", mode="before")(
+        interpolatable_boolean_validator()
+    )
+    _validate_create_parents = field_validator("create_parents", mode="before")(
+        interpolatable_boolean_validator()
+    )
 
 
 class CreateFileOutput(BlockOutput):
@@ -101,6 +118,10 @@ async def execute(  # type: ignore[override]
             FileExistsError: File exists and overwrite=False
             Exception: Other I/O errors
         """
+        # Resolve interpolatable fields to their actual types
+        overwrite = resolve_interpolatable_boolean(inputs.overwrite, "overwrite")
+        create_parents = resolve_interpolatable_boolean(inputs.create_parents, "create_parents")
+
         # Resolve path with security validation
         path_result = PathResolver.resolve_and_validate(inputs.path, allow_traversal=True)
         if not path_result.is_success:
@@ -112,7 +133,7 @@ async def execute(  # type: ignore[override]
 
         # Check overwrite protection
         file_existed = file_path.exists()
-        if file_existed and not inputs.overwrite:
+        if file_existed and not overwrite:
             raise FileExistsError(f"File exists and overwrite=False: {file_path}")
 
         # Convert mode to integer if it's a string
@@ -136,7 +157,7 @@ async def execute(  # type: ignore[override]
             content=inputs.content,
             encoding=inputs.encoding,
             mode=mode_int,
-            create_parents=inputs.create_parents,
+            create_parents=create_parents,
         )
 
         if not write_result.is_success:
@@ -161,9 +182,17 @@ class ReadFileInput(BlockInput):
 
     path: str = Field(description="File path to read (absolute or relative)")
     encoding: str = Field(default="utf-8", description="Text encoding")
-    required: bool = Field(
+    required: bool | str = Field(
         default=True,
-        description="If False, missing file returns empty content instead of error",
+        description=(
+            "If False, missing file returns empty content instead of error "
+            "(or interpolation string)"
+        ),
+    )
+
+    # Validator for boolean field with interpolation support
+    _validate_required = field_validator("required", mode="before")(
+        interpolatable_boolean_validator()
     )
 
 
@@ -222,6 +251,9 @@ async def execute(  # type: ignore[override]
             FileNotFoundError: File not found and required=True
             Exception: Other I/O errors
         """
+        # Resolve interpolatable fields to their actual types
+        required = resolve_interpolatable_boolean(inputs.required, "required")
+
         # Resolve path
         path_result = PathResolver.resolve_and_validate(inputs.path, allow_traversal=True)
         if not path_result.is_success:
@@ -233,7 +265,7 @@ async def execute(  # type: ignore[override]
 
         # Check if file exists
         if not file_path.exists():
-            if inputs.required:
+            if required:
                 raise FileNotFoundError(f"File not found: {file_path}")
             else:
                 # Graceful: return empty content

src/workflows_mcp/engine/executors_file.py

@@ -22,7 +22,7 @@
 from typing import TYPE_CHECKING, Any, ClassVar
 
 import httpx
-from pydantic import Field, model_validator
+from pydantic import Field, field_validator, model_validator
 
 if TYPE_CHECKING:
     from typing import Self
@@ -34,6 +34,12 @@
     ExecutorCapabilities,
     ExecutorSecurityLevel,
 )
+from .interpolation import (
+    interpolatable_boolean_validator,
+    interpolatable_numeric_validator,
+    resolve_interpolatable_boolean,
+    resolve_interpolatable_numeric,
+)
 
 # ============================================================================
 # HttpCall Executor
@@ -71,14 +77,27 @@ class HttpCallInput(BlockInput):
             "Matches httpx parameter name."
         ),
     )
-    timeout: int = Field(
+    timeout: int | str = Field(
         default=30,
-        description="Request timeout in seconds",
-        ge=1,
-        le=1800,
+        description="Request timeout in seconds (or interpolation string)",
+    )
+    follow_redirects: bool | str = Field(
+        default=True, description="Whether to follow HTTP redirects (or interpolation string)"
+    )
+    verify_ssl: bool | str = Field(
+        default=True, description="Whether to verify SSL certificates (or interpolation string)"
+    )
+
+    # Validators for numeric and boolean fields with interpolation support
+    _validate_timeout = field_validator("timeout", mode="before")(
+        interpolatable_numeric_validator(int, ge=1, le=1800)
+    )
+    _validate_follow_redirects = field_validator("follow_redirects", mode="before")(
+        interpolatable_boolean_validator()
+    )
+    _validate_verify_ssl = field_validator("verify_ssl", mode="before")(
+        interpolatable_boolean_validator()
     )
-    follow_redirects: bool = Field(default=True, description="Whether to follow HTTP redirects")
-    verify_ssl: bool = Field(default=True, description="Whether to verify SSL certificates")
 
     @model_validator(mode="after")
     def validate_body_exclusive(self) -> Self:
@@ -163,6 +182,13 @@ async def execute(  # type: ignore[override]
             httpx.HTTPStatusError: HTTP error responses (can be caught)
             Exception: Other HTTP client errors
         """
+        # Resolve interpolatable fields to their actual types
+        timeout = resolve_interpolatable_numeric(inputs.timeout, int, "timeout", ge=1, le=1800)
+        follow_redirects = resolve_interpolatable_boolean(
+            inputs.follow_redirects, "follow_redirects"
+        )
+        verify_ssl = resolve_interpolatable_boolean(inputs.verify_ssl, "verify_ssl")
+
         # Substitute environment variables in URL and headers
         url = self._substitute_env_vars(inputs.url)
         headers = {key: self._substitute_env_vars(value) for key, value in inputs.headers.items()}
@@ -174,9 +200,9 @@ async def execute(  # type: ignore[override]
 
         # Create HTTP client with configuration
         async with httpx.AsyncClient(
-            timeout=inputs.timeout,
-            follow_redirects=inputs.follow_redirects,
-            verify=inputs.verify_ssl,
+            timeout=timeout,
+            follow_redirects=follow_redirects,
+            verify=verify_ssl,
         ) as client:
             # Make HTTP request (parameters pass through directly to httpx)
             try:
@@ -188,9 +214,7 @@ async def execute(  # type: ignore[override]
                     content=inputs.content,  # Direct passthrough - type-safe!
                 )
             except httpx.TimeoutException as e:
-                raise httpx.TimeoutException(
-                    f"Request timeout after {inputs.timeout}s: {url}"
-                ) from e
+                raise httpx.TimeoutException(f"Request timeout after {timeout}s: {url}") from e
             except httpx.NetworkError as e:
                 raise httpx.NetworkError(f"Network error for {url}: {e}") from e