企业官网建设流程全解析

Activepieces Piece 触发器开发完整指南：Polling 与 Webhook 模式实战

【免费下载链接】activepiecesAI Agents & MCPs & AI Workflow Automation • (~400 MCP servers for AI agents) • AI Automation / AI Agent with MCPs • AI Workflows & AI Agents • MCPs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/ac/activepieces

导读

本指南以 Activepieces 官方 Piece 开发规范（.agents/skills/piece-builder/trigger-patterns.md）为骨架，系统讲解如何为第三方 API 编写两种核心触发器——Polling（轮询）与Webhook（推送），并覆盖去重策略、webhook 握手、webhook 续期、context完整传递、AI 元数据等进阶主题。读完本文，你将能够：为任意 REST API 选择正确的触发策略、写出类型安全且符合仓库规范的触发器代码、对照仓库真实样例（Airtable、Stripe）验证实现，并理解框架底层（pollingHelper、TriggerStrategy、WebhookRenewStrategy）的完整工作原理。

一、先选型：Polling 还是 Webhook？

Activepieces 的触发器只有两种主流形态，它们对应完全不同的数据获取方式：

核心原则：只要 API 支持 webhook，优先使用 webhook—— 它即时生效且资源开销远低于轮询（见 trigger-patterns.md）。

从框架源码看，TriggerStrategy一共定义了四种策略，后两者用于特殊场景（见 packages/core/piece-types/src/lib/trigger.ts）：

export enum TriggerStrategy { POLLING = 'POLLING', WEBHOOK = 'WEBHOOK', APP_WEBHOOK = 'APP_WEBHOOK', MANUAL = 'MANUAL', }

其中MANUAL用于人工触发的测试场景，APP_WEBHOOK用于 Slack 这类「OAuth2 应用级 webhook」平台，本文聚焦POLLING与WEBHOOK两种主流形态。

二、Polling 触发器：两种去重策略全解

2.1 工作原理与两种去重策略

当 API 不提供 webhook 时使用轮询：Activepieces 每隔约 5 分钟调用一次你的items函数拉取数据。为了防止重复触发，框架提供了两种去重（deduplication）策略（见 trigger-patterns.md）：

• TIMEBASED（最常见）—— 每条数据带时间戳，只返回比上次轮询时间更新的条目；
• LAST_ITEM—— 每条数据带唯一 ID，只返回「上次已知 ID 之后」的条目。

两者的语义定义在packages/pieces/common/src/lib/polling/index.ts中（源码）：

export enum DedupeStrategy { TIMEBASED, LAST_ITEM, } // TIMEBASED：items 返回带 epochMilliSeconds 的条目 interface TimebasedPolling<AuthValue, PropsValue> { strategy: DedupeStrategy.TIMEBASED; items: (params: { auth: AuthValue; store: Store; propsValue: PropsValue; lastFetchEpochMS: number; // 上次轮询的时间戳 server?: ServerContext; }) => Promise<{ epochMilliSeconds: number; data: unknown }[]>; } // LAST_ITEM：items 返回带唯一 id 的条目 interface LastItemPolling<AuthValue, PropsValue> { strategy: DedupeStrategy.LAST_ITEM; items: (params: { auth: AuthValue; store: Store; propsValue: PropsValue; lastItemId: unknown; // 上次已知的最后一条 ID server?: ServerContext; }) => Promise<{ id: unknown; data: unknown }[]>; }

去重状态如何持久化？两种策略都把断点写入store（Key-Value 存储）：

• TIMEBASED 使用store.get('lastPoll')/store.put('lastPoll', ...)保存「上次轮询时间」；
• LAST_ITEM 使用store.get('lastItem')/store.put('lastItem', ...)保存「上次最后一条 ID」。

poll时框架会取出断点值、调用你的items、再计算新的断点写回（源码）。TIMEBASED 的具体逻辑是：过滤掉epochMilliSeconds <= lastFetchEpochMS的条目，并用本次条目中的最大时间戳推进lastPoll：

case DedupeStrategy.TIMEBASED: { const lastEpochMilliSeconds = await store.get<number>('lastPoll'); if (isNil(lastEpochMilliSeconds)) { throw new Error("lastPoll doesn't exist in the store."); } const items = await polling.items({ store, auth, propsValue, lastFetchEpochMS: lastEpochMilliSeconds, server }); const newLastEpochMilliSeconds = items.reduce( (acc, item) => Math.max(acc, item.epochMilliSeconds), lastEpochMilliSeconds ); await store.put('lastPoll', newLastEpochMilliSeconds); return items .filter((f) => f.epochMilliSeconds > lastEpochMilliSeconds) .map((item) => item.data); }

LAST_ITEM 则要求items返回按 ID 从新到旧排序的数组，框架在items结果中定位lastItemId的位置，取其之前的条目作为新数据（源码）：

const lastItemIndex = items.findIndex((f) => f.id === lastItemId); let newItems = []; if (isNil(lastItemId) || lastItemIndex == -1) { newItems = items ?? []; } else { newItems = items?.slice(0, lastItemIndex) ?? []; // 新的在最前 } // Sorted from newest to oldest if (!isNil(maxItemsToPoll)) { newItems = newItems.slice(-maxItemsToPoll); // 可选：截断数量 } const newLastItem = newItems?.[0]?.id; if (!isNil(newLastItem)) { await store.put('lastItem', newLastItem); } return newItems.map((item) => item.data);

关键约束（源码可证）：TIMEBASED 的items返回顺序无要求，但epochMilliSeconds必须是严格递增的时间戳；LAST_ITEM 的items必须按 ID 降序（新→旧）返回，否则去重会漏数据。

2.2 必须把整个context传给 pollingHelper

这是轮询触发器最容易踩的坑（见 trigger-patterns.md）：

永远把完整的context对象传给pollingHelper.onEnable/onDisable/poll/test，绝不能只传子集（如{ store, auth, propsValue }）。

原因从源码可以精确定位：pollingHelper的每个方法参数类型中每个字段都是可选的（如poll的参数包含store、auth、propsValue、files、maxItemsToPoll、server，见 packages/pieces/common/src/lib/polling/index.ts），所以传子集在 TypeScript 类型检查上能通过，但被省略的字段会被静默丢弃。更隐蔽的是：onEnable现在会读取context.isRepublish（源码）：

async onEnable(polling, { store, auth, propsValue, server, isRepublish }) { case DedupeStrategy.TIMEBASED: { if (isRepublish && !isNil(await store.get<number>('lastPoll'))) { break; // 重新发布已运行的流程时：保留现有 lastPoll } await store.put('lastPoll', Date.now()); break; } // ... }

重新发布（republish）语义：当用户重新发布一个正在运行的流程时，框架传isRepublish: true，此时onEnable会保留已存在的lastPoll/lastItem断点，避免重复消费上一次轮询以来的所有事件。如果传的是子集（缺isRepublish），断点会被重置，上一次轮询后产生的所有事件都会重复触发 —— 数据丢失与风暴同时发生。

因此仓库规范明确要求：

编辑一个已存在的轮询触发器时，顺手把该 piece 里所有pollingHelper调用都改成传完整context。仓库中大多数 piece 仍传子集，SKILL.md 给出了「触及即修复（fix-on-touch）」而非全仓 codemod 的理由——一次改动 300 个 piece 的 PR 会波及大量没人用的死代码，而触及即修复能覆盖真正被使用的 piece。改完记得在 PR 描述中说明，避免被当作无关改动（见 trigger-patterns.md 与 SKILL.md）。

2.3 TIMEBASED 轮询完整模板（最常见）

以下代码来自 trigger-patterns.md，是 TIMEBASED 轮询触发器的标准骨架：

import { createTrigger, TriggerStrategy, AppConnectionValueForAuthProperty } from '@activepieces/pieces-framework'; import { DedupeStrategy, Polling, pollingHelper, httpClient, HttpMethod, AuthenticationType } from '@activepieces/pieces-common'; import { myAppAuth } from '../auth'; const polling: Polling<AppConnectionValueForAuthProperty<typeof myAppAuth>, Record<string, never>> = { strategy: DedupeStrategy.TIMEBASED, items: async ({ auth, propsValue, lastFetchEpochMS }) => { const response = await httpClient.sendRequest<{ data: any[] }>({ method: HttpMethod.GET, url: 'https://api.example.com/v1/records', authentication: { type: AuthenticationType.BEARER_TOKEN, token: auth.secret_text, }, queryParams: { sort: 'created_at', order: 'desc', limit: '100', }, }); return response.body.data.map((item) => ({ epochMilliSeconds: new Date(item.created_at).getTime(), data: item, })); }, }; export const newRecordTrigger = createTrigger({ auth: myAppAuth, name: 'new_record', displayName: 'New Record', description: 'Triggers when a new record is created', props: {}, sampleData: {}, type: TriggerStrategy.POLLING, async test(context) { return await pollingHelper.test(polling, context); }, async onEnable(context) { await pollingHelper.onEnable(polling, context); }, async onDisable(context) { await pollingHelper.onDisable(polling, context); }, async run(context) { return await pollingHelper.poll(polling, context); }, });

模板要点拆解：

• 泛型参数：Polling<AuthValue, PropsValue>的第一个泛型是连接（connection）的值类型，AppConnectionValueForAuthProperty<typeof myAppAuth>会根据你在auth.ts中定义的认证方式自动推导（SecretText 对应secret_text，OAuth2 对应access_token）；第二个泛型是 props 值类型，无 props 时用Record<string, never>；
• items的职责：只负责「带鉴权拉数据 + 转成框架要求的结构」。TIMEBASED 必须返回{ epochMilliSeconds, data }[]，data是真正会进入流程的载荷；
• 四个生命周期方法：test（UI 中测试，返回最近若干条）、onEnable（启用流程时初始化断点）、onDisable（停用时清理，轮询场景为空操作）、run（每次轮询执行）；
• sampleData必填：缺失会导致构建报错（见 SKILL.md）。

真实案例：Airtable 的「New Record」触发器完整实现了上述模式（new-record.trigger.ts），它额外展示了带 props 的写法——base、tableId、viewId三个下拉框通过StaticPropsValue<typeof props>类型注入，items内直接使用propsValue.base、propsValue.tableId!拼装查询，同时用classification: 'READ'和aiMetadata标注了 AI 元数据：

const props = { base: airtableCommon.base, tableId: airtableCommon.tableId, viewId: airtableCommon.views, }; const polling: Polling<AppConnectionValueForAuthProperty<typeof airtableAuth>, StaticPropsValue<typeof props>> = { strategy: DedupeStrategy.TIMEBASED, items: async ({ auth, propsValue }) => { const records = await airtableCommon.getTableSnapshot({ personalToken: auth.secret_text, baseId: propsValue.base, tableId: propsValue.tableId!, limitToView: propsValue.viewId, }); return records.map((record) => ({ epochMilliSeconds: Date.parse(record.createdTime), data: record, })); }, };

2.4 LAST_ITEM 轮询模板

当数据有唯一 ID 但没有可靠时间戳时使用 LAST_ITEM（见 trigger-patterns.md）：

const polling: Polling<undefined, Record<string, never>> = { strategy: DedupeStrategy.LAST_ITEM, items: async ({ auth, propsValue, lastItemId }) => { const response = await httpClient.sendRequest<any[]>({ method: HttpMethod.GET, url: 'https://api.example.com/v1/records', queryParams: { sort: 'id', order: 'desc', limit: '50' }, }); return response.body.map((item) => ({ id: item.id, // 唯一标识 data: item, })); }, };

使用要点（结合源码）：

• 无认证时第一个泛型写undefined，auth字段在items中也不可用；
• 返回结构必须是{ id, data }[]，其中id用于定位去重断点；
• queryParams必须按id降序排序（sort: 'id', order: 'desc'），因为框架用items.findIndex((f) => f.id === lastItemId)定位断点后取slice(0, lastItemIndex)——顺序反了会全部重放或全部丢弃；
• 可选的maxItemsToPoll参数会从最新端截断返回数量（源码）。

2.5 带 Props 的 Polling（类型收紧）

当触发器有用户可配置的 props（如项目过滤器）时，必须把 props 类型填入Polling的第二个泛型，否则propsValue.projectId无法获得类型提示（见 trigger-patterns.md）：

const props = { projectId: Property.Dropdown({ /* ... */ }) }; const polling: Polling< AppConnectionValueForAuthProperty<typeof myAppAuth>, StaticPropsValue<typeof props> // ← 把 props 类型填进来 > = { strategy: DedupeStrategy.TIMEBASED, items: async ({ auth, propsValue }) => { // propsValue.projectId 现在可用且有类型 const response = await httpClient.sendRequest<{ data: any[] }>({ method: HttpMethod.GET, url: `https://api.example.com/v1/projects/${propsValue.projectId}/records`, // ... }); return response.body.data.map((item) => ({ epochMilliSeconds: new Date(item.created_at).getTime(), data: item, })); }, };

记得把props传给createTrigger({ ..., props })，其余生命周期方法与基础 TIMEBASED 模板完全一致。Airtable 示例就是「带 props 的 TIMEBASED 轮询」的仓库级范本。

三、Webhook 触发器：注册、接收、清理

3.1 标准生命周期

当 API 支持 webhook 注册时使用TriggerStrategy.WEBHOOK。三个钩子的职责（见 trigger-patterns.md）：

1. onEnable—— 用context.webhookUrl向第三方 API 注册 webhook（Activepieces 为每次流程运行生成唯一回调地址）；
2. run—— 处理收到的 webhook 载荷，返回数组，每个元素会成为一次独立的流程运行；
3. onDisable—— 流程关闭时删除已注册的 webhook。

完整模板（见 trigger-patterns.md）：

import { createTrigger, TriggerStrategy } from '@activepieces/pieces-framework'; import { httpClient, HttpMethod, AuthenticationType } from '@activepieces/pieces-common'; import { myAppAuth } from '../auth'; export const newRecordWebhookTrigger = createTrigger({ auth: myAppAuth, name: 'new_record_webhook', displayName: 'New Record', description: 'Triggers when a new record is created', props: {}, sampleData: { id: '123', name: 'Example record', created_at: '2024-01-01T00:00:00Z', }, type: TriggerStrategy.WEBHOOK, async onEnable(context) { // 向外部服务注册 webhook const response = await httpClient.sendRequest<{ id: string }>({ method: HttpMethod.POST, url: 'https://api.example.com/v1/webhooks', authentication: { type: AuthenticationType.BEARER_TOKEN, token: context.auth.secret_text, }, body: { url: context.webhookUrl, // Activepieces 提供回调地址 events: ['record.created'], }, }); // 保存 webhook ID 供清理时使用 await context.store.put('webhookId', response.body.id); }, async onDisable(context) { const webhookId = await context.store.get<string>('webhookId'); if (webhookId) { await httpClient.sendRequest({ method: HttpMethod.DELETE, url: `https://api.example.com/v1/webhooks/${webhookId}`, authentication: { type: AuthenticationType.BEARER_TOKEN, token: context.auth.secret_text, }, }); } }, async run(context) { // 返回 webhook 载荷数组（每个元素 = 一次独立流程运行） return [context.payload.body]; }, async test(context) { // 可选：拉取最近数据用于 UI 测试 const response = await httpClient.sendRequest<{ data: any[] }>({ method: HttpMethod.GET, url: 'https://api.example.com/v1/records', authentication: { type: AuthenticationType.BEARER_TOKEN, token: context.auth.secret_text, }, queryParams: { limit: '5' }, }); return response.body.data || []; }, });

模板要点拆解：

• onEnable/onDisable是一对：注册时把第三方返回的 webhook ID 存进context.store，清理时取出来 DELETE，保证流程启停不留孤儿 webhook；
• context.webhookUrl由框架注入，是外部 API 回调的目标地址；
• run必须返回数组：[context.payload.body]意味着整包载荷作为一次运行；拆开数组则一次事件可产生多次运行；
• test是可选的：从源码看（trigger.ts），不提供test时框架回退为SIMULATION测试策略（直接返回sampleData），提供后使用TEST_FUNCTION策略。

真实案例：Stripe 的「New Customer」触发器是仓库中最经典的 webhook 实现（new-customer.ts）：

async onEnable(context) { const webhook = await stripeCommon.subscribeWebhook( 'customer.created', // Stripe 事件名 context.webhookUrl, context.auth.secret_text ); await context.store.put<WebhookInformation>('_new_customer_trigger', { webhookId: webhook.id, }); }, async onDisable(context) { const response = await context.store?.get<WebhookInformation>('_new_customer_trigger'); if (response !== null && response !== undefined) { await stripeCommon.unsubscribeWebhook(response.webhookId, context.auth.secret_text); } }, async run(context) { const payloadBody = context.payload.body as PayloadBody; return [payloadBody.data.object]; // Stripe 模式：取 data.object },

3.2 Webhook 嵌套事件数据（Stripe 模式）

许多 API 会把事件数据包一层壳，例如 Stripe 的事件体是{ data: { object: {...} } }，真正有用的部分是data.object。此时在run中解包（见 trigger-patterns.md）：

async run(context) { const payload = context.payload.body as { data: { object: unknown } }; return [payload.data.object]; // Stripe pattern }

Stripe 真实实现（new-customer.ts）与之完全一致，并用type PayloadBody显式声明了载荷形状。

3.3 Webhook 握手（Challenge-Response）

部分 API（如 Slack、Okta）在注册 webhook 时会先发送一个验证 challenge，你的端点必须在响应中原样返回它，注册才算完成。Activepieces 提供handshakeConfiguration+onHandshake钩子（见 trigger-patterns.md）：

import { WebhookHandshakeStrategy } from '@activepieces/pieces-framework'; export const myTrigger = createTrigger({ // ... type: TriggerStrategy.WEBHOOK, handshakeConfiguration: { strategy: WebhookHandshakeStrategy.HEADER_PRESENT, paramName: 'x-verification-challenge', }, async onHandshake(context) { const challenge = context.payload.headers['x-verification-challenge']; return { status: 200, body: { challenge }, headers: { 'Content-Type': 'application/json' }, }; }, // ... 其余触发器配置 });

从框架源码看，握手策略共有五种（packages/core/piece-types/src/lib/trigger.ts），HEADER_PRESENT只是其中一种：

export enum WebhookHandshakeStrategy { NONE = 'NONE', HEADER_PRESENT = 'HEADER_PRESENT', QUERY_PRESENT = 'QUERY_PRESENT', BODY_PARAM_PRESENT = 'BODY_PARAM_PRESENT', HEAD_REQUEST = 'HEAD_REQUEST', }

选择哪种策略取决于第三方 API 用何种方式传递 challenge：在 header 中、在 URL query 中、在请求体参数中，还是通过 HEAD 请求。paramName用于指定携带 challenge 的字段名（header 名 / query 参数名 / body 参数名）。注意：未配置handshakeConfiguration时框架默认NONE，onHandshake默认返回{ status: 200 }（见 trigger.ts）。

3.4 Webhook 续期（Renewal）

部分 API 的 webhook 会过期（如 Google Sheets 的 channel 有 TTL）。Activepieces 提供renewConfiguration+onRenew钩子，用 cron 表达式周期性地重建 webhook（见 trigger-patterns.md）：

import { WebhookRenewStrategy } from '@activepieces/pieces-framework'; export const myTrigger = createTrigger({ // ... renewConfiguration: { strategy: WebhookRenewStrategy.CRON, cronExpression: '0 */12 * * *', // 每 12 小时 }, async onRenew(context) { // 删除旧 webhook，创建新 webhook const oldId = await context.store.get<string>('webhookId'); if (oldId) await deleteWebhook(oldId, context.auth); const newWebhook = await createWebhook(context.webhookUrl, context.auth); await context.store.put('webhookId', newWebhook.id); }, // ... });

从框架源码看（trigger.ts），续期策略只有两种，且未配置时默认为NONE：

export enum WebhookRenewStrategy { CRON = 'CRON', NONE = 'NONE', } export const WebhookRenewConfiguration = z.union([ z.object({ strategy: z.literal(WebhookRenewStrategy.CRON), cronExpression: z.string() }), z.object({ strategy: z.literal(WebhookRenewStrategy.NONE) }), ]);

续期与握手的组合语义：onRenew与onEnable的职责不同——onEnable只在流程启用时执行一次（含重新发布），onRenew按cronExpression周期执行。续期时的标准做法是「先删旧、后建新、再存新 ID」，确保任意时刻只有一个有效 webhook 指向 Activepieces。

四、触发策略速查表

（速查表见 trigger-patterns.md。）其中APP_WEBHOOK适用于 Slack 这类以 OAuth2 应用（App）为单位注册 webhook 的平台，与单实例 webhook 的生命周期管理方式不同。

五、AI-Ready 元数据：新触发器的强制要求

随着 Activepieces 同时服务人类流程构建者和 AI Agent（通过 MCP server 与 agent 工具链），每个新增触发器必须携带 AI 元数据（见 trigger-patterns.md）：

createTrigger({ // ... aiMetadata: { description: 'Fires when a new record is created in My App, once per record.', // 1~2 句 }, })

三条硬规则（详见 ai-metadata.md）：

1. aiMetadata.description必填—— 用一两句话说明「事件何时触发」以及「一个载荷代表什么」（每条记录一次？每批一次？更新时也触发吗？）；
2. 触发器不接受audience——audience是 action 专属字段，因为「触发器是一个事件，不是可被 Agent 调用的操作」；
3. 不需要idempotent—— 该字段同样仅用于 action 的「安全重试」推导。

此外，从 ai-metadata.md 和 SKILL.md 可知：所有触发器（无论轮询还是 webhook）的classification一律为'READ'——这个徽标回答的是「该步骤是否改变外部状态」，而轮询/webhook 都是被动接收事件，天然属于只读：

export const newRecordTrigger = createTrigger({ name: 'new_record', classification: 'READ', // 所有触发器恒为 READ displayName: 'New Record', description: 'Triggers when a new record is created', aiMetadata: { description: 'Fires when a new record is created in My App, once per record.', }, // ... type, props, sampleData, run, onEnable, onDisable });

在仓库的真实实现中，Airtable 的 new-record.trigger.ts 与 Stripe 的 new-customer.ts 都完整携带了classification: 'READ'与aiMetadata.description，可作对照范本。

六、触发器接线与验证清单

编写完触发器后，还需要完成接线（wiring）与验证（来自 SKILL.md）：

接线检查清单：

• 在src/index.ts中import触发器并加入triggers: [...]；
• 触发器携带aiMetadata与classification: 'READ'；
• 在仓库根目录tsconfig.base.json中注册路径映射（按字母序插入，缺失会导致构建失败）：

  "@activepieces/piece-<name>": ["packages/pieces/community/<name>/src/index.ts"]

• 已有 piece 的改动需在package.json中升级版本号（新增触发器 →PATCH；移除/破坏性变更 →MAJOR，见 SKILL.md）。

构建与本地测试：

bun install # 仅新建 piece 需要——创建工作区符号链接 npx turbo run build --filter=@activepieces/piece-<name> npx turbo run lint --filter=@activepieces/piece-<name>

构建与 lint 必须全部通过；lint 失败（未使用的 import、any类型、未使用变量）即使构建通过也会阻塞 CI。常见 TS 错误包括：src/index.ts缺少导入、tsconfig.base.json缺少条目、触发器缺少sampleData。本地测试时，在packages/server/api/.env中添加AP_DEV_PIECES=<name>，npm start后访问localhost:4200。

结语

掌握本文内容后，你已经可以独立完成三类最常见的 piece 触发器开发：TIMEBASED 轮询（有可靠时间戳的增量数据）、LAST_ITEM 轮询（有 ID 无时间戳的数据）、标准 Webhook（注册-接收-清理三件套），并能正确处理嵌套事件解包、握手验证、到期续期等进阶场景。编写时请始终牢记两条仓库级铁律：pollingHelper 调用必须传完整context（防止isRepublish断点丢失导致重复触发），以及每个新触发器都必须携带aiMetadata+classification: 'READ'。动手前可先对照 Airtable 轮询触发器 与 Stripe webhook 触发器 两份真实实现，它们分别代表了两种策略的仓库级最佳实践。

【免费下载链接】activepiecesAI Agents & MCPs & AI Workflow Automation • (~400 MCP servers for AI agents) • AI Automation / AI Agent with MCPs • AI Workflows & AI Agents • MCPs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/ac/activepieces