Run an Action for Each CSV Row
The most common CSV workflow: take a file of records and do one thing per row. Parse the CSV into rows with data.csv-to-items, loop them with core.for-each, and put the per-row action inside the loop. The example logs each row so it runs the moment you paste it — swap the log for the real action and the shape doesn’t change, whether you’re calling an API, writing to a data store, or sending an email.
Workflow JSON
Section titled “Workflow JSON”The recipe takes the CSV inline so it runs as-is. To feed it from an uploaded file instead, Setup shows how to read the CSV from a form file upload.
{ "name": "Run an Action for Each CSV Row", "initial": { "csv": "email,firstName,plan\njane@acme.com,Jane,pro\nbob@acme.com,Bob,starter\ncarol@acme.com,Carol,pro" }, "steps": [ { "stepId": "parse-csv", "stepType": "data.csv-to-items", "input": { "source": "content", "csv": "{{ initial.csv }}", "skipHeader": true } }, { "stepId": "process-rows", "stepType": "core.for-each", "input": { "items": "{{ parse-csv.items }}", "concurrency": 8, "continueOnError": true, "steps": [ { "stepId": "process-row", "stepType": "core.log", "input": { "message": "Processing {{ $item.email }} ({{ $item.firstName }}) on the {{ $item.plan }} plan. Replace this step with the real action: an HTTP call, a data store write, an email." } } ] } }, { "stepId": "respond", "stepType": "core.return", "input": { "formSubmission": { "type": "success", "message": "Processed {{ process-rows.count }} rows", "data": { "Total": "{{ process-rows.count }}", "Succeeded": "{{ process-rows.succeeded }}", "Failed": "{{ process-rows.failed }}" } } } } ]}Trigger: add a Form trigger with a file upload field so a partner can drop a CSV and the workflow runs on submit. The uploaded file arrives at {{ initial.<field>.url }} — point the parse step at it by switching source to file and setting fileUrl:
{ "stepId": "parse-csv", "stepType": "data.csv-to-items", "input": { "source": "file", "fileUrl": "{{ initial.leads.url }}", "skipHeader": true }}Reading from a file URL streams the rows instead of holding them in memory, so one run can iterate a file far larger than it could load at once. For files over ~30 MB, raise the field’s Max file size in the form builder.
How it works
Section titled “How it works”CSV (inline, or a form file upload) → parse into rows → for each row, concurrency 8: → run the action (the example logs the row) → return how many succeeded and failedconcurrency controls how many rows run at once — 8 here. Raise it to move faster, lower it to stay under an API’s rate limit. continueOnError keeps the batch going when a single row fails; the failure is counted, not fatal. core.for-each reports count, succeeded, and failed, which the Return step hands back as the submission summary.
What to customize first
Section titled “What to customize first”- Swap the action. Replace the
process-rowlog step with whatever you need per row — a data store write, an email, an LLM call, or a sub-workflow. Everything insidestepsruns once per row with{{ $item }}bound to that row. - Tune concurrency to the downstream limit. If the API you’re calling allows 10 requests/second, keep concurrency low and add a Rate Limit check inside the loop rather than cranking it up.
- Collect the failures. Each row’s result is in
{{ process-rows.items }}. Add a step after the loop that filters to the failed ones and writes them to a data store or emails them back, so a partner knows which rows didn’t go through. - Validate before you act. Add a
core.ifstep at the top of the loop that skips rows missing a required field, so a malformed line doesn’t burn an API call.
Related recipes
Section titled “Related recipes”- Lead Dedupe + Enrich Pipeline — when you need to clean and transform the rows as a set before acting on them
- Async Queue Dispatcher — same per-item loop, but draining a queue at a controlled rate instead of a file