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

2026-09-11 23:54:23作者：何将鹤

AI 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 的触发器只有两种主流形态，它们对应完全不同的数据获取方式：

特性	Polling（轮询）	Webhook（推送）
数据获取方式	定期主动调用 API 查询	由第三方 API 主动回调
实时性	取决于轮询间隔（默认约 5 分钟）	近乎即时
资源消耗	持续占用请求配额	仅在事件发生时产生流量
依赖条件	只需 API 提供读取端点	要求 API 支持 webhook 注册

核心原则：只要 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）：

onEnable —— 用 context.webhookUrl 向第三方 API 注册 webhook（Activepieces 为每次流程运行生成唯一回调地址）；
run —— 处理收到的 webhook 载荷，返回数组，每个元素会成为一次独立的流程运行；
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。

四、触发策略速查表

策略	适用场景	关键要点
`TriggerStrategy.POLLING`	API 没有 webhook	用 `pollingHelper` + TIMEBASED 或 LAST_ITEM 去重
`TriggerStrategy.WEBHOOK`	API 支持 webhook 注册	`onEnable` 注册、`onDisable` 删除
`TriggerStrategy.APP_WEBHOOK`	OAuth2 应用级平台 webhook（如 Slack）	使用 `context.app.createListeners()`

（速查表见 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）：

aiMetadata.description 必填 —— 用一两句话说明「事件何时触发」以及「一个载荷代表什么」（每条记录一次？每批一次？更新时也触发吗？）；
触发器不接受 audience —— audience 是 action 专属字段，因为「触发器是一个事件，不是可被 Agent 调用的操作」；
不需要 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 触发器两份真实实现，它们分别代表了两种策略的仓库级最佳实践。