Author a production task

An Envoy task is one streaming pipeline. It declares the connector types it needs, accepts typed runtime arguments, and passes records through named steps.

Keep task YAML portable: declare connector slots by name and type, but keep URLs, credentials, paths, and account-specific configuration in stored connectors or connectors.yaml.

Design before writing YAML

Define:

1. the source and its pagination behavior;
2. the destination operation;
3. a stable source identity;
4. the canonical payload used for change detection;
5. failure and retry behavior;
6. the expected result and reconciliation checks;
7. the smallest useful test dataset.

A task has one source stream. Split independent source pulls into separate tasks and compose them with a job. Keep dependent child pulls in the same task when each child query uses its parent input.

Start from a minimal task

name: contacts/stage
description: Stage contacts from CSV into SQLite
output: none
 
connectors:
  source:
    type: csv
  staging:
    type: sqlite
 
arguments:
  dry_run:
    type: boolean
    default: true
 
steps:
  - name: read_contacts
    type: call_method
    config:
      connector: source
      method: read
      args:
        resource: contacts.csv
 
  - name: normalize
    type: transform
    config: |
      {
        "id": $string(input.id),
        "name": $trim(input.name),
        "email": $lowercase(input.email)
      }
 
  - name: write_contact
    type: call_method
    run_if: "arguments.dry_run = false"
    config: |
      {
        "connector": "staging",
        "method": "upsert",
        "args": {
          "table": "contacts",
          "data": input,
          "conflict_columns": ["id"]
        }
      }

output: none is appropriate because the final step writes to a connector. Use ndjson when stdout is the intended data output. Avoid json for large pipelines because it collects all results.

Use static and dynamic configuration correctly

A YAML object is static. Use it when the step does not need the current record. Use a JSONata string when referencing input, arguments, variables, job_config, or earlier step output.

call_method replaces the current input with the connector result. If a later step needs the original identity after a write, carry it in a preceding transform and read steps.<name>.output.

Filter explicitly

run_if and skip_if decide whether a step executes; skipped records pass through unchanged. Use filter to remove records:

- name: usable_email
  type: filter
  config: "$exists(input.email) and input.email != ''"

A transform that evaluates to a missing value also passes its original input through before execution. Do not use that behavior as a filter.

Preserve streaming

Use a call_method step for list, find, read, and other collection methods. It can consume async results with backpressure.

Use $callMethod() inside JSONata only for bounded lookups such as find_one, count, or a grouped aggregate. It buffers the complete method result.

Add concurrency to a slow per-record step only after measuring correctness and provider limits:

- name: write_contact
  type: call_method
  concurrency: 5
  error_handling: ignore
  config: |
    { ... }

Use batch to emit fixed-size arrays. Use spool only when the next step truly requires the whole dataset; it is intentionally unbounded.

Choose error handling

• fail stops the task and is the default.
• ignore records the error, drops that step result, and continues.
• ignore_rest skips remaining steps for the affected item.

If a write uses ignore, guard the success-mark step against a missing result and record a separate error state. Otherwise a failed write can be marked successful.

Validate and test

In the UI, validate before saving. From the CLI:

envoy validate tasks/contacts/stage.yaml

Static validation catches structure, known step types, connector references, argument schemas, and expression compilation. It does not validate live credentials, upstream fields, or evaluation against real data.

Use this progression:

1. inspect one representative source record;
2. run with writes disabled;
3. inspect transformed output and logs;
4. enable writes for a small bounded set;
5. verify destination records and mappings;
6. rerun the same set and confirm no duplicates;
7. test one invalid record;
8. scale page size or concurrency gradually.

See Task structure, Step types, and Task patterns.