Skip to main content

Type

workflow_executor

Formats

JSON, Python DSL, Agent-based
Purpose: The Workflow Executor skill enables agents to run sophisticated multi-step workflows with support for conditional branching, parallel execution, and orchestration across multiple systems.

Common Use Cases

Multi-step deployment pipelinesBuild, test, and deploy applications across multiple environments

Complex orchestration tasksCoordinate operations across multiple services and systems

Dependent task executionExecute tasks with dependencies and data passing between steps

Conditional workflow branchingMake decisions based on runtime conditions and previous step outcomes

Workflow Formats

FormatBest ForKey Features
JSONSimple, linear workflowsDeclarative syntax, version control friendly, schema validation
Python DSLComplex logic, dynamic workflowsFull Python capabilities, dynamic step generation, complex conditionals
Agent-basedMulti-agent orchestrationDelegate to specialized agents, hierarchical execution
Format Selection: Start with JSON for simple workflows, use Python DSL for complex logic, Agent-based for multi-agent coordination.

Configuration

Example Configuration:
{
  "validation_enabled": true,
  "timeout": 1800,
  "allowed_executors": ["shell", "python", "docker"],
  "max_parallel_steps": 5,
  "on_failure": "stop"
}
ParameterTypeDefaultDescription
validation_enabledbooleantrueValidate workflows before execution
timeoutnumber1800Global workflow timeout (seconds)
allowed_executorsarray[“shell”, “python”]Permitted step executors
max_parallel_stepsnumber5Maximum concurrent steps
parallel_executionbooleantrueEnable parallel execution
on_failurestring”stop”Failure strategy (stop, continue, retry)
max_retriesnumber3Maximum retry attempts per step
retry_delaynumber5Delay between retries (seconds)
enable_context_sharingbooleantrueShare data between steps

Quick Start

# Create skill
kubiya skill create --name "CI/CD Executor" --type workflow_executor --enabled

# Associate with agent
kubiya skill associate agent <agent-id> <skill-id>

View Complete Examples

See full deployment pipelines, ETL workflows, and multi-agent orchestration examples

Workflow Features

Conditional Execution

Execute steps based on runtime conditions:
{
  "id": "deploy-to-prod",
  "type": "shell",
  "command": "deploy-production.sh",
  "condition": "test_results.passed && approval.status == 'approved'"
}

Parallel Execution

Run independent steps concurrently:
{
  "parallel_group": "testing",
  "steps": [
    {"id": "unit-tests", "type": "shell", "command": "npm run test:unit"},
    {"id": "integration-tests", "type": "shell", "command": "npm run test:integration"},
    {"id": "e2e-tests", "type": "shell", "command": "npm run test:e2e"}
  ]
}

Data Passing Between Steps

Share outputs between workflow steps:
{
  "steps": [
    {
      "id": "get-version",
      "type": "shell",
      "command": "cat version.txt",
      "output_variable": "app_version"
    },
    {
      "id": "build-image",
      "type": "docker",
      "command": "build -t myapp:{{app_version}} .",
      "depends_on": ["get-version"]
    }
  ]
}

Error Handling and Retries

Configure retry behavior for resilient workflows:
{
  "id": "api-call",
  "type": "shell",
  "command": "curl https://api.example.com/data",
  "retries": 3,
  "retry_delay": 5,
  "on_error": "continue"
}

Example Workflow

Deployment Pipeline

{
  "name": "deployment-pipeline",
  "steps": [
    {
      "id": "build",
      "type": "docker",
      "command": "build -t myapp:latest .",
      "timeout": 300
    },
    {
      "parallel_group": "testing",
      "depends_on": ["build"],
      "steps": [
        {"id": "unit-tests", "type": "shell", "command": "npm test"},
        {"id": "lint", "type": "shell", "command": "npm run lint"}
      ]
    },
    {
      "id": "deploy",
      "type": "shell",
      "command": "kubectl apply -f k8s/",
      "depends_on": ["testing"],
      "condition": "unit_tests.status == 'success'"
    }
  ]
}

Security Best Practices

Always enable validation_enabled to catch errors before execution.
"validation_enabled": true  # Validate workflow syntax and dependencies
Limit allowed_executors to only necessary execution methods.
"allowed_executors": ["shell", "python"]  # No docker or agent execution
Configure timeouts to prevent runaway workflows.
"timeout": 1800  # Global 30-minute timeout
Best Practice: Start with on_failure: "stop" for production workflows to catch issues early.

Solutions:
  • Check JSON syntax is valid
  • Verify all depends_on references exist
  • Ensure no circular dependencies
  • Validate executor types are in allowed_executors
Solutions:
  • Increase step-specific timeout
  • Increase global workflow timeout
  • Optimize step execution
  • Break into smaller steps
Solutions:
  • Verify enable_context_sharing: true
  • Check output_variable is set on source step
  • Ensure variable name matches in dependent step