docs/build-pieces/piece-reference/flow-control.mdx
Flow Controls let an action change the shape of the run — stop it early, send an intermediate HTTP response, or pause the flow and resume later when an external signal arrives. All of these are exposed on the ctx parameter of the action's run method.
Stop the flow and (optionally) return a response to the webhook trigger that started it.
With a response:
context.run.stop({
response: {
status: context.propsValue.status ?? StatusCodes.OK,
body: context.propsValue.body,
headers: (context.propsValue.headers as Record<string, string>) ?? {},
},
});
Without a response:
context.run.stop();
A waitpoint is a durable checkpoint: the run is marked PAUSED, its execution state is persisted, and the action will be invoked a second time once the waitpoint is resumed. Waitpoints survive worker restarts — see Durable Execution for the full model.
The same action runs twice — once to create the waitpoint, once to read the resume payload — so every pausing action branches on ctx.executionType:
import { ExecutionType } from '@activepieces/shared';
async run(ctx) {
if (ctx.executionType === ExecutionType.BEGIN) {
// First invocation: create the waitpoint and pause.
// A real WEBHOOK waitpoint must surface waitpoint.buildResumeUrl(...) to
// the outside world — see the "Wait for a webhook callback" section below.
const waitpoint = await ctx.run.createWaitpoint({ type: 'WEBHOOK' });
ctx.run.waitForWaitpoint(waitpoint.id);
return {};
}
// Second invocation: the waitpoint was resumed.
return {
body: ctx.resumePayload.body,
headers: ctx.resumePayload.headers,
queryParams: ctx.resumePayload.queryParams,
};
}
Two hooks do the work:
ctx.run.createWaitpoint({ type, ... }) — registers the waitpoint on the server and returns { id, resumeUrl, buildResumeUrl }.ctx.run.waitForWaitpoint(waitpointId) — tells the engine the step's verdict is paused; the run transitions to PAUSED after the action returns.There are two waitpoint types.
Create a WEBHOOK waitpoint and expose its resume URL — the flow will resume whenever that URL is called.
async run(ctx) {
if (ctx.executionType === ExecutionType.BEGIN) {
const waitpoint = await ctx.run.createWaitpoint({ type: 'WEBHOOK' });
const callbackUrl = waitpoint.buildResumeUrl({
queryParams: { runId: ctx.run.id },
});
// Send `callbackUrl` somewhere the outside world can reach it
// (email, Slack message, third-party API, etc).
ctx.run.waitForWaitpoint(waitpoint.id);
return {};
}
return {
approved: ctx.resumePayload.queryParams['action'] === 'approve',
};
}
buildResumeUrl takes an optional sync: true to return the caller a synchronous response produced by the remainder of the flow, and a queryParams object that is carried through to ctx.resumePayload.queryParams on resume.
Pause the flow and immediately reply to the webhook trigger — useful for "we got your submission, we'll call you back" patterns. Pass responseToSend and your HTTP response is sent right away; the flow then sits paused until the returned URL is called.
async run(ctx) {
if (ctx.executionType === ExecutionType.BEGIN) {
const waitpoint = await ctx.run.createWaitpoint({
type: 'WEBHOOK',
responseToSend: {
status: 200,
headers: {},
body: { accepted: true },
},
});
const nextWebhookUrl = waitpoint.buildResumeUrl({
queryParams: { created: new Date().toISOString() },
sync: true,
});
// nextWebhookUrl is what the counterpart should call to resume this run.
ctx.run.waitForWaitpoint(waitpoint.id);
return { nextWebhookUrl };
}
return {
body: ctx.resumePayload.body,
headers: ctx.resumePayload.headers,
queryParams: ctx.resumePayload.queryParams,
};
}
Create a DELAY waitpoint with the UTC timestamp you want to resume at. The server schedules a one-time job that fires at resumeDateTime and resumes the run automatically.
async run(ctx) {
if (ctx.executionType === ExecutionType.RESUME) {
return { success: true };
}
const futureTime = new Date(Date.now() + 60 * 60 * 1000); // 1 hour
const waitpoint = await ctx.run.createWaitpoint({
type: 'DELAY',
resumeDateTime: futureTime.toUTCString(),
});
ctx.run.waitForWaitpoint(waitpoint.id);
return {};
}
resumeDateTime is capped by the server's AP_PAUSED_FLOW_TIMEOUT_DAYS setting; the engine throws PausedFlowTimeoutError if you ask for a longer delay.
On the RESUME branch, ctx.resumePayload is whatever the resume call carried in:
ctx.resumePayload.body // request body from the webhook caller (empty for DELAY)
ctx.resumePayload.headers // HTTP headers from the webhook caller
ctx.resumePayload.queryParams // parsed ?foo=bar query string
For DELAY waitpoints there is no incoming HTTP request, so the payload is empty — use the RESUME branch simply to produce the step's final output.