Building Flows

A flow moves data from one or more source systems to one or more destination systems. It runs on a schedule, in response to events (webhooks, listeners), or when triggered by another flow. Flows are the primary way integrations get work done in Celigo.

A flow has page generators (exports that fetch data) and page processors (imports and lookups that process each record). Processors run sequentially in a flat list, or conditionally through routers that branch records to different paths. These processing pipeline mechanics -- routers, branches, page processors, response mapping -- are shared with APIs and tools (see

building-apis

and

building-tools

What Starts a Flow

Flows start themselves -- this is the biggest thing that separates them from APIs (invoked by an HTTP caller) and tools (invoked by a consumer). Every flow begins with one or more page generators, of two kinds:

Scheduled exports -- the flow runs on a cron cadence and each run pulls from the source: everything (full sync), only what changed since the last run (delta sync), records matching a query, or files landed in an FTP/SFTP/S3 folder. The right primitive for batch work: nightly reconciliations, hourly delta syncs, backfills, off-peak windows.
Listeners -- the source pushes to the flow. A webhook fires (or a NetSuite/Salesforce native real-time event triggers) and the payload immediately starts flowing. No schedule; the flow runs as events arrive. The right primitive for event-driven work ("when X happens, do Y"), especially when latency matters.

A flow can mix both, and multi-generator designs are common:

Real-time plus batch safety net -- a listener catches events as they fire; a scheduled export reconciles at off-peak hours, catching up after webhook outages
Consolidating sources -- customers from Salesforce AND HubSpot, each with its own generator, feeding the same downstream pipeline
Different slices of the same source -- one export pulls new records, another pulls updated records, when the API exposes them separately

If the requirement is "every night at 2 AM, do X" or "when a webhook arrives, do Y" -- that lives on a flow. APIs and tools have no schedule and no listener; they only run when invoked.

Fetched Data Needs a Downstream Consumer

A common design mistake: ending a flow on an import that fetches data back from a remote system (a preview call, a query, a lookup-shaped POST) and relying on response mapping to capture the result. Response mapping makes fields visible to the NEXT step -- if no next step exists, the captured data is discarded when the run ends and nobody sees it.

When the requirement says "preview / estimate / retrieve / fetch / check / look up", the design needs at least one of:

A write-back import to the source system (most common) -- e.g. source export -> preview import -> update import that writes the captured fields onto the source record
A persistent destination the user named (file to S3/SFTP, email, database)
A router or AI agent step that consumes the captured data within the same run

A two-step export -> fetch-shaped-import flow with nothing after it is a smell -- re-read the intent for where the fetched data should end up. The same applies in reverse: capturing a created record's ID via response mapping is only useful if a later step writes it somewhere.

Flow Topologies

Linear Flows

A flat

pageProcessors[]

list with no routers. One or more page generators feed records through a sequential chain of page processors. Each processor is either an import (

type: "import"

) or a lookup export (

type: "export"

). Records pass through every step in order. Unique to flows -- APIs and tools always use routers.

Branching Flows

Page generators feed records into

routers[]

instead of

pageProcessors[]

. Each router evaluates records against branch conditions and routes them to matching branches. Branches contain their own

pageProcessors[]

and can chain to other routers via

nextRouterId

Two routing modes (shared with APIs and tools):

Input filters (
```
routeRecordsUsing: "input_filters"
```
) -- S-expression rules on each branch; last branch can omit filter as a catch-all
Script-based (
```
routeRecordsUsing: "script"
```
) -- a JavaScript function returns the branch name

Flows support both

first_matching_branch

and

all_matching_branches

routing. APIs only support

first_matching_branch

. Tools support

first_matching_branch

only.

A flow uses EITHER

pageProcessors

(linear) OR

routers

(branching) at the top level -- not both.

When a branching flow needs linear steps before the branch point (e.g., a lookup enrichment or AI classification that all branches depend on), use a pass-through router: a single-branch router with

nextRouterId

pointing to the branching router. Omit

routeRecordsTo

and

routeRecordsUsing

on the pass-through router -- including them makes it appear as a filter-based branch in the UI. The API defaults are sufficient.

Abstract / Instance Flows

A template/inheritance model. An abstract flow (

isAbstract: true

) defines the complete graph but cannot execute. Instance flows (

_abstractFlowId

) inherit the graph and customize via an

overrides

object (connections, schedules, mappings, filters).

Use when the same flow structure is deployed across multiple regions, tenants, or environments with different connections or parameters.

Quick Reference

Flow Type Decision Matrix

Pattern Structure Key fields Read schema

Linear

Flat processor list

Pattern	Structure	Key fields	Read schema
Linear	Flat processor list	`pageGenerators[]` , `pageProcessors[]`	`request.yml` , `page-generator.yml` , `page-processor.yml`
Branching (routers)	Routers with conditional branches	`pageGenerators[]` , `routers[]`	+ `router.yml` , `branch.yml`
Abstract / Instance	Template + per-instance overrides	`isAbstract: true` / `_abstractFlowId` , `overrides`	+ `overrides-helper.yml` , `overrides.yml`

pageGenerators[]

pageProcessors[]

request.yml

page-generator.yml

page-processor.yml

Branching (routers)

Routers with conditional branches

pageGenerators[]

routers[]

router.yml

branch.yml

Abstract / Instance

Template + per-instance overrides

isAbstract: true

_abstractFlowId

overrides

overrides-helper.yml

overrides.yml

Minimum Required Fields

Every flow needs at minimum:

```
name
```
-- display name
```
_integrationId
```
-- parent integration
```
disabled: true
```
-- always create disabled
```
pageGenerators[]
```
-- at least one entry with
```
_exportId
```
Either
```
pageProcessors[]
```
(linear) or
```
routers[]
```
(branching) -- never both

Which Schemas to Read

Always read:

request.yml -- base flow fields
page-generator.yml -- export sources, per-generator schedules, delta coordination
page-processor.yml -- import/export steps with responseMapping and hooks

Add for branching flows:

router.yml -- routing strategy, record distribution mode
branch.yml -- input filters, per-branch processors, chaining

Add if response mapping is needed:

response-mapping.yml -- extract/generate pairs for carrying data between steps

All available schemas (in references/schemas/):

Base fields (all flows): request.yml
Response shape: response.yml
Page generators: page-generator.yml
Page processors: page-processor.yml
Response mapping: response-mapping.yml
Routers: router.yml
Branches: branch.yml
Abstract flow helpers: overrides-helper.yml
Instance overrides: overrides.yml
Cloning: clone-request.yml, clone-response.yml

Related Skills

configuring-connections > Quick Reference -- connection types and auth methods for page generators and processors
configuring-exports > Quick Reference -- building exports used as page generators and lookup processors
configuring-imports > Quick Reference -- building imports used as page processors
writing-mappings > Mapper 2.0 Workflow -- field mappings on imports and response mapping between steps
writing-scripts > Data Pipeline Hooks -- preSavePage, preMap, postMap, postSubmit, postResponseMap hooks
writing-handlebars > Quick Reference -- dynamic values in URIs, filters, delta tokens, SQL queries
troubleshooting-flows > Diagnostic Workflow -- diagnosing flow failures, errors, and performance issues

How to Build a Flow

1. Plan the flow

Before creating anything, decide what kind of operation this is:

Decision tree:

Modifying an existing flow's step config (export settings, import mappings, scripts) -- work on the step directly, not the flow. Use
```
celigo exports set
```
,
```
celigo imports set
```
, or the relevant skill (configuring-exports, configuring-imports, writing-scripts, writing-mappings)
Modifying an existing flow's structure (add/remove steps, change schedule, rename) -- GET the flow, modify the structure, PUT it back. Don't rebuild from scratch
Building a new flow where every step is known -- build directly, bottom-up (skip to step 2)
Any ambiguity about what steps are needed -- design first. List every system, every data direction, every step before writing any JSON

Design checklist (when ambiguity exists):

What source systems? What destination systems?
What data moves between them, in which direction?
How often? (cron schedule, webhook trigger, on-demand)
What happens when a step fails? (
```
proceedOnFailure
```
, error notifications)
Do downstream steps need data from upstream responses? (response mapping)
Is this a one-off or a reusable template? (abstract/instance flow)
Sandbox or production? (never mix --
```
sandbox: true
```
flows only use
```
sandbox: true
```
connections)

2. Identify the integration

Every flow belongs to an integration (the container). Find or create the integration first.

bash

celigo integrations list

3. Check for existing patterns

Before building from scratch, check what already exists in the account and marketplace.

bash

# Search for similar resources in the account index
celigo account search "<keyword>"

# Show what an existing resource uses and what uses it
celigo account dependencies flow <id>

# Find orphaned resources, offline connections, untriggered flows
celigo account lint

# Search marketplace for pre-built integration templates
celigo templates marketplace

# Preview a template before installing
celigo templates preview <id> --summary

The account index auto-refreshes when stale (>4 hours). Force a fresh snapshot with

celigo account snapshot

4. Build the connections, exports, and imports

Flows reference existing resources. Build bottom-up: connections first, then exports and imports that use those connections, then the flow that wires them together.

bash

celigo connections list
celigo exports list
celigo imports list

For every step, match the adaptor to the target application -- raw HTTP is the fallback, not the default. Use the native adaptor when one exists (NetSuite, Salesforce, databases, FTP/S3); otherwise check for a pre-built HTTP connector (550+ apps:

celigo http-connectors list

) and build the connection from it; hand-write HTTP config from public API docs only when no connector exists or it doesn't cover the endpoint. See configuring-exports > Check for a pre-built connector and configuring-imports > Check for a pre-built connector.

See

configuring-exports

and

configuring-imports

for how to build each resource.

5. Choose the topology

Scenario	Topology
All records follow the same path	Linear ( `pageProcessors` )
Records need conditional routing by field values	Branching with input filters
Routing logic requires custom JavaScript	Branching with script router
Records should fan out to all matching paths	Branching with `all_matching_branches`
Same structure across multiple tenants/regions	Abstract + instance flows

Abstract/instance flows: Abstract flows are reusable templates that cannot run directly. Instance flows inherit the abstract's structure and override specific fields (connections, filters, schedules). Use when the same integration pattern repeats across tenants or regions. Create with

isAbstract: true

. Top-level

pageProcessors

are automatically wrapped into a single-branch router. Instance flows reference the abstract via

_abstractFlowId

and specify overrides -- they do NOT use the normal scaffolding process.

_integrationId

is NOT inherited and must be set explicitly on the instance.

6. Design the step sequence

For each step, decide:

Type --
```
import
```
(write to destination) or
```
export
```
(lookup for enrichment)
Response mapping -- what data from this step's response do downstream steps need? Only add response mapping when downstream steps need fields that aren't already in the record or when field names need to change. If the lookup returns fields with the same names the downstream step expects, skip the response mapping -- it adds complexity without value.
proceedOnFailure -- should the pipeline continue if this step fails?
Hooks -- does this step need a
```
postResponseMap
```
script?

7. Build the flow JSON

Reference the schemas listed in the Quick Reference above for exact field schemas.

8. Configure scheduling

Pair

schedule

(6-field cron) with

timezone

(IANA). Omit both for listener/webhook/realtime flows.

timezone

defaults to UTC, which is usually wrong for human-facing schedules ("9 AM every weekday" should survive daylight saving).

Individual page generators can override the flow schedule via their own

schedule

field (e.g. one source syncs hourly, another nightly, in the same flow).

9. Set the runtime controls

Flow-level and per-step switches that change production behavior. APIs and tools have none of these -- they are flow-only.

Control	Where	Default	Flip it when
`proceedOnFailure`	per processor	`false` (a failed record stops there)	The step is non-critical and downstream steps still do meaningful work without it (a Slack notification late in the flow shouldn't block the sync). Keep `false` when downstream depends on this step's output
`skipRetries`	flow, and per generator	`false` (failed jobs retry)	Work is time-sensitive (retrying a stale webhook is meaningless) or non-idempotent (retries risk duplicates). Per-generator override: set it only on the real-time generator
`runPageGeneratorsInParallel`	flow	`false` (generators run sequentially)	Sources are independent and can take the load. Careful: parallel generators hitting the same API can blow rate limits that sequential runs respect
`autoResolveMatchingTraceKeys`	flow	duplicate trace keys raise an error	The source genuinely emits duplicates in normal operation, or the flow is intentionally idempotent. Don't enable it to paper over upstream duplication

10. Configure chaining (if needed)

```
_runNextFlowIds
```
-- trigger other flows when this one completes. The classic use is multi-stage pipelines: "after the customer-master sync finishes, run the orders sync." When a requirement says "X has to happen, then Y", chain two focused flows rather than building one big one
```
_runNextExportIds
```
-- more granular: trigger specific exports inside other flows instead of the whole flow

11. Create disabled, verify, enable

Always create with

disabled: true

. Verify the structure with

celigo flows get

. Enable only after verification.

CLI Commands

bash

# CRUD
celigo flows list
celigo flows get <id>
celigo flows create < flow.json
celigo flows update <id> < flow.json
celigo flows set <id> key=value [key2=value2 ...]
celigo flows delete <id>

# Run
celigo flows run <id> [--start-date <ISO8601>] [--end-date <ISO8601>] [--export-ids <ids>] -y

# Test run (stage-by-stage)
celigo flows test-run <id> --export <exportId>
celigo flows test-run-step-results <id> <runId> <exportOrImportId>

# Clone
echo '{"connectionMap":{"oldId":"newId"}}' | celigo flows clone <id> <integrationId> <environmentId> [--flow-group <id>]

# Structure manipulation
celigo flows add-generator <id> <exportId> [--schedule '<cron>'] [--index <pos>]
celigo flows remove-generator <id> <exportId>
celigo flows add-processor <id> <exportOrImportId> [--router <routerId>] [--branch <branchName>]
celigo flows remove-processor <id> <exportOrImportId> [--router <routerId>] [--branch <branchName>]
celigo flows replace-connection <id> <oldConnectionId> <newConnectionId>

# Error management
celigo flows errors <id> <exportOrImportId>
celigo flows resolved-errors <id> <exportOrImportId>
celigo flows resolve-errors <id> <exportOrImportId> [errorIds] [-y]
celigo flows retry-errors <id> <exportOrImportId> [retryDataKeys] [-y]
celigo flows assign-errors <id> <exportOrImportId> <email> [errorIds] [-y]
celigo flows delete-resolved-errors <id> <exportOrImportId> [errorIds] [-y]
celigo flows error <id> <exportOrImportId> <errorId> [--retry-data] [--request-detail]
celigo flows update-error-data <id> <exportOrImportId> <errorId>
celigo flows tag-errors <id> <exportOrImportId>
celigo flows error-summary <id>
celigo flows error-analysis <id> <exportOrImportId> [--limit <n>]

# Debug
celigo flows debug-requests <id> <exportOrImportId> [--since <minutes>]
celigo flows debug-request-detail <id> <exportOrImportId> <key>
celigo flows enable-execution-logs <id> [--duration <minutes>]
celigo flows disable-execution-logs <id>
celigo flows execution-logs <id> <jobId>
celigo flows query-execution-logs <id> <jobId> --export-or-import-id <id> --group-id <gid> --record-id <rid>
celigo flows execution-log-detail <id> <jobId> --export-or-import-id <id> --stage <stage> --group-id <gid> --record-id <rid>

# Metadata
celigo flows last-export-date <id>

# Integration-level flow management
celigo integrations flow-groups <integrationId>
celigo integrations create-flow-group <integrationId> <name>
celigo flows set-group <flowGroupingId> <flowIds...>

Pre-Submit Checklist

Before creating or updating a flow, verify:

```
_integrationId
```
references a real integration (confirm with
```
celigo integrations get <id>
```
)
```
disabled: true
```
is set for initial creation -- an enabled flow with a schedule runs immediately
```
schedule
```
is 6-field cron with seconds:
```
"? */5 * * * *"
```
(first field is always
```
?
```
)
```
pageProcessors[]
```
and
```
routers[]
```
are mutually exclusive -- a flow uses one or the other, never both
Router IDs are unique within the flow (random alphanumeric, e.g.,
```
"N8Q9NX24Sj5"
```
)
Every branch
```
nextRouterId
```
references an existing router
```
id
```
in the same flow

Gotchas

PUT erases omitted fields. Always GET first, modify, then PUT. The
```
set
```
command handles this automatically.
pageProcessors
and
routers
are mutually exclusive. A flow uses one or the other at the top level. Setting both causes validation errors.
Router IDs must be unique within a flow. Use random alphanumeric strings (e.g.,
```
"N8Q9NX24Sj5"
```
).
```
nextRouterId
```
must reference an existing router
```
id
```
in the same flow.
Create flows with
disabled: true
. An enabled flow with a schedule will run immediately. Enable only after verification.
Build order matters. Connection -> Export -> Import -> Flow. The API rejects references to non-existent resources.
Schedule is 6-field cron with seconds. Format:
```
"? minute hour dayOfMonth month dayOfWeek"
```
. The first field is always
```
?
```
. Common mistake: using 5-field cron without the seconds position.
Instance flows cannot define structure. Do not set
```
pageGenerators
```
,
```
pageProcessors
```
, or
```
routers
```
on instance flows -- these are inherited from the abstract flow. All customizations go through
```
overrides
```
.
Instance flow
overrides
is full-replace on PUT. Omitting an override entry removes it. Always GET, merge changes, then PUT.
Empty
pageProcessors: []
in a branch is the discard pattern. Records matching that branch are dropped. A branch with no
```
inputFilter
```
serves as a catch-all.
responseMapping
uses Transformation 1.0 syntax (extract/generate pairs), not expression-based transforms. Lookup export responses use
```
data[0].fieldName
```
; import responses use
```
_json.fieldName
```
.
Don't add unnecessary transforms on lookups. If the lookup returns fields with the same names the downstream import expects, skip the transform -- the data flows through as-is. Only add a transform when you need to rename fields, reshape nested structures, or drop fields. An identity transform (e.g.,
```
errorId
```
->
```
$.errorId
```
) adds complexity for no benefit.

Common Errors

Error	Cause	Fix
`"pageProcessors" is not allowed when "routers" is present`	Both `pageProcessors[]` and `routers[]` set on the same flow	Remove one -- use `pageProcessors` for linear, `routers` for branching
`Invalid reference: _integrationId`	Integration ID does not exist or is misspelled	Verify with `celigo integrations get <id>`
`Invalid reference: _exportId` / `_importId`	Export or import referenced in a page generator/processor does not exist	Create the export/import first, then reference it
`Invalid reference: _connectionId`	Connection ID on an export or import does not exist	Verify with `celigo connections get <id>`
`Duplicate router id`	Two routers in the same flow share the same `id`	Assign unique alphanumeric IDs to each router
`Invalid nextRouterId`	A branch references a router `id` that does not exist in the flow	Ensure `nextRouterId` matches an actual router `id` in the same flow
`Invalid cron expression`	Schedule uses 5-field cron or wrong format	Use 6-field format: `"? /5 * * *"` (seconds field first, always `?` )
Flow runs immediately after creation	Created with `disabled: false` or `disabled` omitted (defaults to enabled)	Always set `disabled: true` on create; enable after verification

building-flows

NPX Install

Tags

SKILL.md Content

Building Flows

What Starts a Flow

Fetched Data Needs a Downstream Consumer

Flow Topologies

Linear Flows

Branching Flows

Abstract / Instance Flows

Quick Reference

Flow Type Decision Matrix

Minimum Required Fields

Which Schemas to Read

Related Skills

How to Build a Flow

1. Plan the flow

2. Identify the integration

3. Check for existing patterns

4. Build the connections, exports, and imports

5. Choose the topology

6. Design the step sequence

7. Build the flow JSON

8. Configure scheduling

9. Set the runtime controls

10. Configure chaining (if needed)

11. Create disabled, verify, enable

CLI Commands

Pre-Submit Checklist

Gotchas

Common Errors