Workflow DSL¶

imports:        # optional file/glob includes — merged into this document
  - "./common/error-handlers.yaml"
  - "./routes/*.yaml"

helix:          # engine configuration (see pkg/config.HelixConfig)
  logging:
    level: info
    format: json

error-handlers: # reusable named error handlers (see below)
  - id: api-defaults
    on-error: [...]

connectors:     # named connector instances (used via connector://name)
  slack-prod:
    type: slack
    config:
      token: "${SLACK_TOKEN}"

workflows:      # the routes themselves — required for almost every doc
  - id: my-route
    trigger:
      uri: kafka://orders?brokers=localhost:9092
    steps:
      - target: postgres://orders/insert?dsn=postgres://localhost/db

rest:           # REST DSL blocks — see ./rest.md
  - path: "/api/v1"
    verbs: [...]

agents:         # top-level agents — converted into workflows
pipelines:      # top-level pipelines — converted into workflows
tools:          # tool:// definitions (mandatory I/O schemas)
mcps:           # MCP servers — converted into helix.mcp.servers
skills:         # reusable skill bundles
tests:          # declarative test cases (see testing guide)

imports:
  - "./common/error-handlers.yaml"
  - "./routes/*.yaml"

trigger:
  uri: kafka://orders?brokers=localhost:9092&group=processors

trigger:
  uri: "timer://tick?period=5000&repeatCount=10"
# same as:
trigger:
  uri: "timer://tick"
  params:
    period: "5000"
    repeatCount: "10"

steps:
  - target: log://debug?level=info        # short scalar form
  - target: kafka://orders?brokers=...    # any component URI
  - wiretap: log://audit                  # fire-and-forget copy

steps:
  - log: { message: "received", level: info }
  - transform: { starlang: 'payload + " v2"' }
  - set-header: { name: "X-Route", constant: "alpha" }
  - set-body: { starlang: 'json({ ok: true })' }
  - remove-header: "Authorization"
  - remove-headers: { pattern: "Helix*", except: ["HelixHttpResponseCode"] }
  - set-property: { name: trace_id, starlang: 'uuid()' }
  - remove-property: "tmp"
  - remove-properties: { pattern: "temp.*", except: ["temp.keep"] }
  - convert-body: { type: string, charset: utf-8 }

steps:
  # Content-based router
  - choice:
      when:
        - starlang: 'headers["priority"] == "high"'
          steps:
            - target: direct://fast-lane
        - starlang: 'headers["priority"] == "low"'
          target: seda://batch-queue      # shorthand: no nested steps
      otherwise:
        steps:
          - target: direct://normal

  # Filter — drops exchanges where predicate is false
  - filter:
      starlang: 'len(payload) > 0'
      steps:
        - target: direct://keep

  # Splitter
  - split:
      starlang: 'payload.items'
      parallel: true
      stop-on-exception: false
      aggregation-strategy: "max-total"
      steps:
        - target: direct://process-item

  # Aggregator
  - aggregate:
      correlation: 'headers["customerId"]'
      strategy: "concat-strategy"
      completion-size: 10
      completion-timeout: 30s
      steps:
        - target: postgres://orders/insert

  # Multicast — fan-out to N endpoints
  - multicast:
      uris: ["direct://a", "direct://b", "direct://c"]
      parallel: true
      strategy: "first-wins"

  - recipient-list: { starlang: 'headers["targets"]', delimiter: ",", parallel: true }
  - routing-slip:   { starlang: 'headers["pipeline"]', delimiter: "->" }
  - dynamic-router: { starlang: 'pickNext(state)' }

  - load-balancer:
      strategy: weighted    # round-robin | random | sticky | failover | weighted
      uris: ["http://a", "http://b"]
      weights: [3, 1]
      sticky-header: "session-id"

steps:
  - retry:
      max-attempts: 3
      delay: "1s"
      backoff-multiplier: 2.0
      steps:
        - target: http://flaky-service

  - timeout:
      duration: 10s
      steps:
        - target: http://slow-service
      fallback:
        - set-body: { constant: '{"error":"timeout"}' }

  - throttle:
      rate: 100
      period: "1s"
      steps:
        - target: http://rate-limited

  - delay:
      duration: 250ms
      steps:
        - target: log://debug

  - circuit-breaker:
      threshold: 5
      half-open-after: "30s"
      steps:
        - target: http://unstable
      fallback:
        - target: http://backup

steps:
  - do-try:
      steps:
        - target: http://primary?timeout=5s
      do-catch:
        - exceptions: ["HTTP:TOO_MANY_REQUESTS"]
          when: 'properties["HelixExceptionCode"] == "HTTP:429"'
          on-error-continue: false        # rethrow after running
          steps:
            - log: { message: "rate limited", level: warn }
        - exceptions: ["HTTP:*", "HTTP:CONNECTIVITY"]
          steps:
            - transform:
                starlang: '{ error: properties["HelixExceptionMessage"] }'
      do-finally:
        - log: { message: "tried", level: debug }

steps:
  - loop:
      count: 5
      max_iterations: 100
      steps:
        - target: direct://once

  - loop:
      starlang: 'len(payload.items)'
      steps:
        - target: direct://each

  - saga:
      steps:
        - target: direct://create-order
        - target: direct://charge-card
      compensate:
        - target: direct://refund
        - target: direct://cancel

  - idempotent-consumer:
      starlang: 'headers["X-Message-Id"]'
      repository: redis
      redis-url: "redis://localhost:6379/0"
      redis-prefix: "myapp:idem:"
      ttl: "24h"

  - claim-check: { operation: push, key: 'headers["X-Job-Id"]' }
  - resequence:  { starlang: 'headers.seq', batch-size: 50, timeout: "10s" }
  - sort:        { starlang: 'a.priority < b.priority' }
  - sample:      { rate: 10, period: "1s" }

steps:
  - throw-exception: { message: "bad input", type: "ValidationError" }
  - rollback:        { message: "abort tx" }
  - stop:            {}                # halts the route gracefully
  - set-exchange-pattern: "InOnly"     # or InOut

steps:
  - validate: 'sizeOf(payload) > 0 && headers["Content-Type"] != null'

  - bean:
      ref: "myService"
      method: "process"

  - step:
      id: "validate-and-enrich"        # for tracing/metrics
      steps:
        - validate: 'payload.id != null'
        - enrich: { uri: postgres://customers/findById }

  - normalize:
      conditions:
        - starlang: 'headers["format"] == "xml"'
          steps: [{ transform: { starlang: 'xmlToJson(payload)' } }]
        - starlang: 'headers["format"] == "csv"'
          steps: [{ transform: { starlang: 'csvToJson(payload)' } }]

  - threads:
      pool-size: 10
      max-queue: 100
      steps:
        - target: direct://heavy

steps:
  - on-completion:
      on-when: 'headers["notify"] == "true"'
      after-consumer: true              # fires after consumer reply, not just route
      steps:
        - target: postgres://orders/insert
      on-complete:
        - target: kafka://order-events
      on-failure:
        - target: kafka://order-dlq

steps:
  - http-request:
      url: https://api.example.com/users
      method: POST
      headers:
        Content-Type: application/json
      body: '{"name":"alice"}'
      timeout: 10s
      auth: { type: bearer, token: "${API_TOKEN}" }

  - db-query:
      dsn: "postgres://localhost/db"
      operation: query
      sql: "SELECT * FROM users WHERE active = $1"
      params: [true]

workflows:
  - id: kafka-to-pg
    trigger: { uri: kafka://orders }
    steps:
      - target: postgres://orders/insert
    error-handler:
      dead-letter-channel: kafka://order-dlq
      max-redeliveries: 3
      redelivery-delay: "5s"
      backoff-multiplier: 2.0

error-handlers:
  - id: api-defaults
    on-error:
      - exceptions: ["HTTP:TOO_MANY_REQUESTS"]
        on-error-continue: false
        steps:
          - log: { message: "rate limited", level: warn }
      - exceptions: ["HTTP:*"]
        steps:
          - transform: { starlang: '{ error: properties["HelixExceptionMessage"] }' }
    do-finally:
      - log: { message: "request done", level: debug }

workflows:
  - id: search
    error-handler-ref: api-defaults
    trigger: { uri: http-server://0.0.0.0:8080/search }
    steps:
      - target: http://upstream/search

workflows:
  - id: secure-api
    trigger: { uri: http-server://0.0.0.0:8080/api }
    auth:
      type: jwt
      jwt:
        secret: "${JWT_SECRET}"
        issuer: "https://auth.example.com"
        audience: "myapp"
    steps:
      - target: direct://handler

helix:
  logging:
    level: "${LOG_LEVEL:-info}"

workflows:
  - id: kafka-in
    trigger:
      uri: "kafka://orders?brokers=${KAFKA_BROKERS}"
    steps:
      - target: "postgres://orders/insert?dsn=${PG_DSN}"

helix:
  logging: { level: info, format: json }

error-handlers:
  - id: outbound-defaults
    on-error:
      - exceptions: ["HTTP:*", "HTTP:CONNECTIVITY"]
        steps:
          - log: { message: "upstream error", level: error }
          - target: kafka://order-dlq

workflows:
  - id: order-fanout
    description: "Fan out an order to three downstream systems, dead-letter on any failure."
    error-handler-ref: outbound-defaults
    trigger:
      uri: kafka://orders?brokers=${KAFKA_BROKERS}&group=fanout
    steps:
      - validate: 'payload.id != null && payload.total > 0'

      - on-completion:
          on-complete:
            - target: kafka://order-events?topic=order.fanned-out
          on-failure:
            - target: kafka://order-events?topic=order.fanout-failed
          steps:
            - do-try:
                steps:
                  - multicast:
                      uris:
                        - "http://shipping/api/orders"
                        - "http://billing/api/charges"
                        - "http://crm/api/orders"
                      parallel: true
                      strategy: "first-wins"
                do-catch:
                  - exceptions: ["HTTP:5*"]
                    steps:
                      - retry:
                          max-attempts: 3
                          delay: "2s"
                          backoff-multiplier: 2.0
                          steps:
                            - target: kafka://order-retry-queue
                do-finally:
                  - log: { message: "fanout cycle done", level: debug }