Skip to content

Workflow Authoring Guide

Complete guide to writing Ploston workflows.

Workflow Structure

Every workflow has this structure:

# Metadata
name: my-workflow
version: "1.0"
description: What this workflow does

# Optional: Package configuration
packages:
  profile: standard
  additional: []

# Optional: Default settings
defaults:
  timeout: 30
  on_error: fail

# Inputs (array format)
inputs:
  - input_name:
      type: string
      required: true
      default: "value"
      description: Input description

# Steps
steps:
  - id: step1
    code: |
      result = "output"

  - id: step2
    tool: tool_name
    params:
      key: "{{ steps.step1.output }}"

# Output
output: "{{ steps.step2.output }}"

Metadata

Required Fields

Field Description
name Unique workflow identifier (alphanumeric, hyphens)
version Semantic version string (e.g., "1.0", "2.1.3")

Optional Fields

Field Description
description Human-readable description
packages Python package configuration
defaults Default step settings

Inputs

Define workflow inputs with validation:

inputs:
  # Simple required string input (just the name)
  - api_key

  # String input with default (makes it optional)
  - name: "World"

  # Full definition with all options
  - count:
      type: integer
      minimum: 1
      maximum: 100
      default: 10

  # Enum (restricted values)
  - format:
      type: string
      enum: ["json", "csv", "xml"]
      default: "json"

  # Pattern validation
  - email:
      type: string
      pattern: "^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+$"

Input Types

Type Description
string Text value
integer Whole number
number Decimal number
boolean true/false
array List of values
object Key-value pairs

Steps

Steps are the building blocks of workflows. Each step is either a code step or a tool step.

Code Steps

Execute Python code:

steps:
  - id: process
    code: |
      data = "{{ inputs.data }}"
      result = data.upper()

See Code Steps Guide for details.

Tool Steps

Call MCP tools:

steps:
  - id: fetch
    tool: http_get
    params:
      url: "{{ inputs.url }}"
      headers:
        Authorization: "Bearer {{ inputs.token }}"

Step Dependencies

Control execution order with depends_on:

steps:
  - id: step1
    code: |
      result = "first"

  - id: step2
    depends_on: [step1]  # Waits for step1
    code: |
      previous = "{{ steps.step1.output }}"
      result = f"after {previous}"

  - id: step3
    depends_on: [step1, step2]  # Waits for both
    code: |
      result = "last"

Without depends_on, steps execute in YAML order.

Step Options

steps:
  - id: risky_step
    tool: external_api
    mcp: my-server               # MCP server hosting the tool
    params:
      data: "{{ inputs.data }}"

    # Conditional execution — step is skipped if falsy
    when: "{{ inputs.run_risky }}"

    # Timeout (seconds)
    timeout: 60

    # Error handling: fail | continue | retry
    on_error: retry

    # Retry configuration
    retry:
      max_attempts: 3
      initial_delay: 1.0
      max_delay: 30.0
      backoff_multiplier: 2.0

    # Skip if tool not available (instead of failing)
    on_missing_tool: skip

Templates

Use Jinja2 templates to reference values:

Accessing Inputs

code: |
  name = "{{ inputs.name }}"

Accessing Step Outputs

code: |
  data = {{ steps.previous_step.output }}        # Step result value
  status = "{{ steps.previous_step.status }}"     # "completed", "failed", or "skipped"
  err = {{ steps.previous_step.error }}           # Error object (code + message) or null
  logs = {{ steps.previous_step.debug_log }}      # List of strings from context.log()

Template Filters

code: |
  # Convert to JSON string
  json_str = '{{ steps.data.output | tojson }}'

  # Default value
  value = "{{ inputs.optional | default('fallback') }}"

Output

Define workflow output:

# Single output
output: "{{ steps.final.output }}"

# Multiple outputs
outputs:
  - name: result
    from_path: steps.process.output.data
  - name: count
    value: "{{ steps.count.output }}"

Complete Example

name: data-transform
version: "1.0"
description: Transform and validate data

inputs:
  - data:
      type: string
      description: JSON data to transform
  - format:
      type: string
      enum: ["json", "csv"]
      default: "json"

defaults:
  timeout: 30
  on_error: fail

steps:
  - id: parse
    code: |
      import json
      data = json.loads('{{ inputs.data }}')
      result = data

  - id: transform
    depends_on: [parse]
    code: |
      data = {{ steps.parse.output }}
      result = {k.upper(): v for k, v in data.items()}

  - id: format
    depends_on: [transform]
    code: |
      import json
      data = {{ steps.transform.output }}
      format_type = "{{ inputs.format }}"
      if format_type == "json":
          result = json.dumps(data, indent=2)
      else:
          result = ",".join(f"{k}={v}" for k, v in data.items())

output: "{{ steps.format.output }}"

Next Steps