Workflow.foreach()
The .foreach() method creates a loop that executes a step for each item in an array. It always returns an array containing the output from each iteration, preserving the original order.
Usage exampleDirect link to Usage example
workflow.foreach(step1, { concurrency: 2 })
ParametersDirect link to Parameters
step:
opts?:
ReturnsDirect link to Returns
workflow:
BehaviorDirect link to Behavior
Execution and waitingDirect link to Execution and waiting
The .foreach() method processes every item before the next step executes, regardless of concurrency settings. The default concurrency: 1 processes items sequentially. Higher values use parallel batches, but the next step still waits for every batch to finish.
If you need to run multiple operations per item, use a nested workflow as the step. This keeps all operations for each item together and is cleaner than chaining multiple .foreach() calls. See Nested workflows inside foreach for examples.
Output structureDirect link to Output structure
.foreach() always outputs an array. Each element in the output array corresponds to the result of processing the element at the same index in the input array.
// Input: [{ value: 1 }, { value: 2 }, { value: 3 }]
// Step adds 10 to each value
// Output: [{ value: 11 }, { value: 12 }, { value: 13 }]
Using .then() after .foreach()Direct link to using-then-after-foreach
When you chain .then() after .foreach(), the next step receives the entire output array as its input. You can aggregate or process all results together.
workflow
.foreach(processItemStep) // Output: array of processed items
.then(aggregateStep) // Input: the entire array
.commit()
Using .map() after .foreach()Direct link to using-map-after-foreach
Use .map() to transform the array output before passing it to the next step:
workflow
.foreach(processItemStep)
.map(async ({ inputData }) => ({
total: inputData.reduce((sum, item) => sum + item.value, 0),
count: inputData.length,
}))
.then(nextStep)
.commit()
Chaining multiple .foreach() callsDirect link to chaining-multiple-foreach-calls
When you chain .foreach() calls, each operates on the array from the previous step:
workflow
.foreach(stepA) // If input is [a, b, c], output is [A, B, C]
.foreach(stepB) // Operates on [A, B, C], output is [A', B', C']
.commit()
If a step inside .foreach() returns an array, the output becomes an array of arrays. Use .map() with .flat() to flatten:
workflow
.foreach(chunkStep) // Output: [[chunk1, chunk2], [chunk3, chunk4]]
.map(async ({ inputData }) => inputData.flat()) // Output: [chunk1, chunk2, chunk3, chunk4]
.foreach(embedStep)
.commit()
Progress events during streamingDirect link to Progress events during streaming
When using run.stream(), foreach steps emit a workflow-step-progress event after each iteration completes. This lets you track real-time progress without waiting for the entire foreach to finish.
const run = await workflow.createRun()
const stream = run.stream({ inputData })
for await (const chunk of stream) {
if (chunk.type === 'workflow-step-progress') {
console.log(`${chunk.payload.completedCount}/${chunk.payload.totalCount}`)
// e.g. "1/3", "2/3", "3/3"
}
}
Each progress event payload contains:
id:
completedCount:
totalCount:
currentIndex:
iterationStatus:
iterationOutput?:
Resuming a single iterationDirect link to Resuming a single iteration
When a step inside .foreach() suspends, each iteration suspends independently. Pass forEachIndex to run.resume() to resume one iteration at a time with its own resumeData. Omitting forEachIndex resumes every suspended iteration with the same data.
await run.resume({
step: 'approve',
resumeData: { ok: true },
forEachIndex: 1,
})