Composio TypeScript SDK Triggers 实战指南订阅与推送实时事件、管理触发器实例【免费下载链接】composioComposio powers 1000 toolkits, tool search, context management, authentication, and a sandboxed workbench to help you build AI agents that turn intent into action.项目地址: https://gitcode.com/GitHub_Trending/co/composio本文围绕 Composio TypeScript SDK 的 Triggers API 展开系统讲解如何在 AI Agent 应用中基于已连接账户创建、查询、启停触发器实例并通过 Pusher 实时通道订阅触发事件同时结合仓库源码Triggers 实现 与 triggers 类型定义深入剖析参数校验、连接账户解析、工具包版本控制、事件负载结构与 Webhook 验签原理。读完本文你将能够独立完成「创建触发器 → 订阅实时事件 → 处理事件负载 → 启停与清理实例」的完整链路。概述Triggers 在 Composio 中的定位Triggers触发器是发生在已连接账户Connected Account上的实时事件。例如 Gmail 收到新邮件、GitHub 仓库出现新 commit、Slack 收到新消息等。Composio SDK 的 Triggers API 让 Agent 不再是轮询式地被动检查状态而是主动接收服务端推送的事件从而构建事件驱动的自动化工作流。SDK 提供的核心能力包括列出所有活动触发器实例支持多种过滤条件与分页为指定用户与触发器类型创建新的触发器实例更新既有触发器实例启用 / 禁用触发器订阅实时触发器事件含复合过滤管理触发器类型列表、详情、枚举从源码结构看Triggers 类 在构造时会同时初始化ComposioClient负责 REST 调用走/api/v3/trigger_instances接口与PusherService负责实时事件订阅并读取全局toolkitVersions配置默认值为latest见 ConfigDefaults.node.ts。这意味着管理与订阅两条通道在 SDK 内部是分离的管理类操作走 HTTP API实时事件走 Pusher 长连接。列出活动触发器listActive用于拉取当前项目的所有活动触发器实例支持按认证配置、连接账户、触发器 ID / 名称过滤并支持游标分页const triggers await composio.triggers.listActive({ authConfigIds: [auth-config-id], connectedAccountIds: [connected-account-id], limit: 10, cursor: cursor-string, // 使用 cursor 进行分页 showDisabled: false, triggerIds: [trigger-id], triggerNames: [trigger-name], });参数说明与源码映射对照 TriggerInstanceListActiveParamsSchema各参数含义如下参数类型说明authConfigIdsstring[]可选按认证配置Auth ConfigID 过滤connectedAccountIdsstring[]可选按连接账户 ID 过滤limitnumber可选单页返回条数cursorstring可选分页游标取上一页响应的nextCursorshowDisabledboolean可选是否包含已禁用的触发器triggerIdsstring[]可选按触发器实例 ID 过滤triggerNamesstring[]可选按触发器名称过滤在实现层面listActive 方法 会先用 zod schema 做safeParse校验失败即抛出ValidationError随后将 camelCase 参数转换为后端接口所需的 snake_case 查询参数auth_config_ids、connected_account_ids、show_disabled、trigger_ids、trigger_names等后调用client.triggerInstances.listActive。整个调用被withCancellation包裹支持通过requestOptions.signal实现请求取消。响应体经 transformTriggerInstanceListActiveResponse 转换后每个条目包含id、connectedAccountId、disabledAt、state、triggerConfig、triggerName、updatedAt以及可选的triggerData、uuid分页信息为nextCursor与totalPages。创建触发器实例create为指定用户与触发器类型创建触发器实例签名如下composio.triggers.create(userId, slug, body?)指定连接账户 IDconst trigger await composio.triggers.create(default, GMAIL_NEW_GMAIL_MESSAGE, { connectedAccountId: ca_jjYIG9L40LDIS, // 指定使用哪个连接账户 triggerConfig: { labelIds: INBOX, userId: me, interval: 60, }, });不指定连接账户 ID使用第一个可用账户const trigger await composio.triggers.create(default, GMAIL_NEW_GMAIL_MESSAGE, { triggerConfig: { labelIds: INBOX, userId: me, interval: 60, }, }); // 后端会解析该用户与工具包下第一个活动连接注意当同一工具包下有多个连接账户时建议显式提供connectedAccountId确保触发器创建到目标账户上。若未提供后端会解析该用户与工具包下的第一个活动连接按最近创建时间排序。参数与返回值userIdstring必填为其创建触发器实例的用户 IDslugstring必填触发器类型的 slug如GMAIL_NEW_GMAIL_MESSAGEbodyTriggerInstanceUpsertParams可选触发器实例配置connectedAccountIdstring可选使用的连接账户 ID不传则由后端解析triggerConfigobject可选触发器特有配置参数如 Gmail 的labelIds、轮询interval返回PromiseTriggerInstanceUpsertResponse结构为{ triggerId: string; // 创建的触发器实例 ID }关键行为与实现细节工具包版本create使用 Composio 客户端初始化时配置的全局工具包版本默认latest。如需指定版本可在初始化时配置toolkitVersions详见下文工具包版本控制。源码中create 方法 会将toolkit_versions、user_id、connected_account_id、trigger_config一并封装进 upsert 请求体。slug 前置校验创建前 SDK 会先调用getType(slug)校验触发器类型是否存在。当后端对未知 slug 返回 400/404 时SDK 会抛出客户端侧的ComposioTriggerTypeNotFoundError见 create 中的校验逻辑并附带排查建议。userId 校验userId为空或全空白会直接抛出ValidationError源码。连接解析后移连接解析现在发生在后端。若用户与工具包下不存在活动连接或指定的connectedAccountId无效upsert 调用会以后端错误的形式 reject而不再抛出客户端侧的ComposioConnectedAccountNotFoundError。指定工具包版本创建// 初始化时配置工具包版本 const composio new Composio({ apiKey: your-api-key, toolkitVersions: { gmail: 12082025_00, github: 10082025_01 } }); // create 将使用为 Gmail 配置的版本 const trigger await composio.triggers.create(default, GMAIL_NEW_GMAIL_MESSAGE, { connectedAccountId: ca_jjYIG9L40LDIS, triggerConfig: { labelIds: INBOX, userId: me, interval: 60, }, }); // 该触发器实例将使用 Gmail 的 12082025_00 版本创建带错误处理的完整示例try { const trigger await composio.triggers.create(default, GMAIL_NEW_GMAIL_MESSAGE, { // connectedAccountId 可选——不传则使用第一个可用账户 connectedAccountId: ca_jjYIG9L40LDIS, triggerConfig: { labelIds: INBOX, userId: me, interval: 60, }, }); console.log(Trigger created:, trigger.triggerId); } catch (error) { if (error instanceof ComposioTriggerTypeNotFoundError) { console.error(Trigger type not found:, error.message); // 处理无效的触发器类型 } else if (error instanceof ValidationError) { console.error(Invalid parameters:, error.message); // 处理参数校验错误 } else { // 连接问题现在以后端错误形式在此处暴露例如 // 用户与工具包下无活动连接或指定的 connectedAccountId 无效 console.error(Unexpected error:, error); // 处理其他错误 } }更新触发器实例对已存在的触发器实例进行更新const updatedTrigger await composio.triggers.update(trigger-id, { // 更新后的配置 });从源码看update 方法 直接委托给client.triggerInstances.manage.update并支持requestOptions与请求取消TriggerInstanceManageUpdateParamsSchema 表明更新参数为status: enable | disable响应为{ status: success }。此外 SDK 还提供delete(triggerId)删除触发器实例返回{ triggerId }。启用 / 禁用触发器通过状态切换控制触发器是否激活// 禁用触发器 await composio.triggers.disable(trigger-id); // 启用触发器 await composio.triggers.enable(trigger-id);源码中 enable / disable 均基于manage.update实现enable发送{ status: enable }disable发送{ status: disable }与 TriggerStatuses 常量 完全对应。被禁用的触发器可通过listActive的showDisabled: true过滤条件查询到对应响应的disabledAt字段。订阅实时触发器事件subscribe通过 Pusher 长连接接收实时事件并可传入过滤条件composio.triggers.subscribe( triggerData { console.log(Received trigger:, triggerData); }, { toolkits: [toolkit-name], triggerId: specific-trigger-id, connectedAccountId: connected-account-id, triggerSlug: [trigger-type], triggerData: custom-data, userId: user-id, } );底层实现Pusher 通道与复合过滤订阅并非简单的 WebSocket 直连而是经过 PusherService 的完整链路获取实时凭证首次使用时通过内部接口/internal/sdk/realtime/credentials拉取pusherKey、pusherCluster、projectId等凭证getPusherClient 方法失败抛出ComposioFailedToGetSDKRealtimeCredentialsError。创建 Pusher 客户端以项目 ID 构造私有频道private-${clientId}_triggers并通过/api/v3/internal/sdk/realtime/auth做频道鉴权鉴权请求头携带x-api-key。绑定事件订阅频道上的trigger_to_client事件同时绑定chunked-trigger_to_client分块事件支持超长负载的分片重组bindWithChunking。复合过滤收到的原始数据经版本解析后由 shouldSendTriggerAfterFilters 执行过滤全部命中才回调你的处理函数。过滤条件在 TriggerSubscribeParamSchema 中定义除文档列出的toolkits、triggerId、connectedAccountId、triggerSlug、triggerData、userId外还支持authConfigId按认证配置过滤仅 V3 负载携带该信息。各过滤器均按不区分大小写匹配 toolkits / triggerSlugtriggerId与负载顶层id精确匹配connectedAccountId与metadata.connectedAccount.id精确匹配。事件负载格式回调收到的triggerData为统一规范化的IncomingTriggerPayloadinterface IncomingTriggerPayload { id: string; // 触发器实例唯一 ID triggerSlug: string; // 触发器类型 toolkitSlug: string; // 关联的工具包 userId: string; // 与触发器关联的用户 ID payload: unknown; // 处理后的触发器负载 originalPayload: unknown; // 原始触发器负载 metadata: { id: string; triggerConfig: unknown; // 触发器配置 triggerSlug: string; toolkitSlug: string; triggerData: string; connectedAccount: { id: string; // 连接账户 nano ID uuid: string; // 连接账户 UUID authConfigId: string; // 认证配置 nano ID authConfigUUID: string; // 认证配置 UUID userId: string; // 用户 ID status: string; // 连接状态ACTIVE / INACTIVE }; }; }多版本负载兼容Pusher 推送的原始数据可能是 V1、V2、V3 或遗留TriggerData格式。SDK 在 parsePusherPayload 中依次尝试WebhookPayloadV3Schema要求type以composio.开头、V2、V1 schema 进行解析并统一归一化为上述结构对无法识别的格式会输出告警日志并返回最小负载结构避免订阅中断。V1/V2 负载信息有限如 V1 无userId、toolkitSlug由trigger_name首段推断而 V3 触发器事件携带完整的trigger_id、connected_account_id、auth_config_id、user_id元数据见 WebhookTriggerPayloadV3Schema这也是订阅过滤器区分负载版本的原因。取消订阅停止接收触发器事件await composio.triggers.unsubscribe();unsubscribe 内部调用pusherService.unsubscribe()从私有频道private-${clientId}_triggers退订若此前已建立长连接也会复用并关闭该连接。管理触发器类型触发器类型Trigger Type描述某工具包的某类事件的元信息slug、名称、描述、payload schema、config schema 等是创建触发器实例的前提。列出触发器类型const triggerTypes await composio.triggers.listTypes({ toolkits: [github], cursor: cursor-string, limit: 10 });参数toolkitsstring[]可选按工具包 slug 过滤cursorstring可选分页游标limitnumber可选单页最大条数返回PromiseTriggersTypeListResponse分页结构包含items、nextCursor、totalPages。每个类型条目TriggerTypeSchema含slug、name、description、instructions、toolkitlogo / slug / name、payload、config、version。实现上listTypes 会将toolkits映射为toolkit_slugs查询参数并附带全局toolkit_versions。获取触发器类型详情按 slug 获取单个触发器类型。该操作同样受初始化时配置的全局toolkitVersions影响默认latest// 使用全局配置的工具包版本获取触发器类型 const triggerType await composio.triggers.getType(GMAIL_NEW_GMAIL_MESSAGE);参数slugstring必填要获取的触发器类型 slug返回PromiseTriggersTypeRetrieveResponse结构示例{ slug: string; name: string; description: string; toolkit: { slug: string; name: string; }; // ... 其他触发器类型属性 }指定工具包版本获取// 初始化时配置工具包版本 const composio new Composio({ apiKey: your-api-key, toolkitVersions: { gmail: 12082025_00, github: 10082025_01 } }); // getType 将使用为 Gmail 配置的版本 const triggerType await composio.triggers.getType(GMAIL_NEW_GMAIL_MESSAGE); // 该请求将使用 Gmail 的 12082025_00 版本获取触发器类型源码中 getType 将全局toolkitVersions作为toolkit_versions参数传给triggersTypes.retrieve返回结果经 transformTriggerTypeRetrieveResponse 规范化为 camelCase 结构。获取触发器枚举获取全部可用触发器类型的枚举列表const triggerEnum await composio.triggers.listEnum();该方法返回所有可用触发器类型的枚举主要供 CLI 使用不需要过滤条件实现见 listEnum。工具包版本控制全局配置的影响文档反复强调的toolkitVersions是创建触发器实例、获取触发器类型时的版本锚点。初始化方式如下详见 Toolkit Versions 配置const composio new Composio({ apiKey: your-api-key, toolkitVersions: { gmail: 12082025_00, github: 10082025_01 } });该配置的默认值为latestConfigDefaults.node.ts、ConfigDefaults.workerd.ts。从 composio.ts 看初始化时还会通过getToolkitVersionsFromEnv支持从环境变量读取。它决定了create创建的触发器实例对应哪个工具包版本getType/listTypes查询到的是哪个版本的触发器类型定义。因此在排查找不到某触发器类型时除了检查 slug 拼写还应确认目标工具包版本下是否真的存在该触发器这也是ComposioTriggerTypeNotFoundError的possibleFixes提示之一。进阶Webhook 方式接收事件与签名校验除 Pusher 实时订阅外SDK 还支持以 Webhook 方式接收触发器事件适合服务端被动接收场景。核心方法setWebhookSubscription({ webhookUrl, enabledEvents?, version? })创建或更新项目级 Webhook 订阅默认订阅composio.trigger.message事件、默认 V3 版本源码接口路径/api/v3.1/webhook_subscriptionsparse(request, options?)将入站 Webhook HTTP 请求解析为规范化的IncomingTriggerPayload。request既可以是 Fetch API 的RequestNext.js App Router、Hono、Remix也可以是{ body, headers }普通对象Express express.raw、Next.js Pages Router头信息不区分大小写verifyWebhook({ payload, signature, id, timestamp, secret, tolerance? })校验签名与时间戳返回{ version, payload, rawPayload }。签名校验机制verifyWebhookSignature签名格式v1,base64EncodedSignature支持空格分隔的多个签名签名输入HMAC-SHA256(${webhookId}.${webhookTimestamp}.${payload}, secret)即webhook-id头、webhook-timestamp头与原始请求体拼接后做 HMAC-SHA256 并 Base64校验使用恒定时间比较timingSafeEqual防时序攻击tolerance默认 300 秒5 分钟设0可关闭时间戳校验校验失败抛出ComposioWebhookSignatureVerificationErrorHTTP 401负载解析失败抛出ComposioWebhookPayloadErrorHTTP 400相关错误类定义见 TriggerErrors.ts。Express 场景示例app.post(/webhooks/composio, express.raw({ type: application/json }), async (req, res) { try { const result await composio.triggers.parse(req, { verifySecret: process.env.COMPOSIO_WEBHOOK_SECRET, }); console.log(Trigger:, result.payload.triggerSlug); console.log(Event data:, result.payload.payload); res.sendStatus(200); } catch (error) { res.sendStatus(401); } });注意parse的verifySecret若显式传入但解析为空字符串如环境变量未设置SDK 会主动抛错而非静默跳过校验以避免伪造事件被接受源码如确实不需要校验请整体省略verifySecret选项。小结与最佳实践多账户场景务必显式传connectedAccountId连接解析已后移至后端不指定时按最近创建选择第一个活动连接容易创建到非预期账户。slug 前置校验由 SDK 完成create会先校验触发器类型ComposioTriggerTypeNotFoundError的possibleFixes会提示核对 slug、工具包版本与工具包页面。区分两类事件通道实时低延迟场景用subscribePusher 长连接含复合过滤与分块重组服务端被动接收用 Webhook parse/verifyWebhookHMAC-SHA256 验签默认 5 分钟容差。善用toolkitVersions锚定版本创建与查询都受全局工具包版本影响排查类型缺失时先确认版本一致性。生命周期管理创建后应记录triggerId不再需要时调用disable暂停接收或delete彻底删除避免无效实例持续产生事件。相关实现与文档触发器核心实现 Triggers.ts、类型与 schema 定义 triggers.types.ts、实时通道 Pusher.ts、响应转换 transformers/triggers.ts、错误定义 TriggerErrors.ts。【免费下载链接】composioComposio powers 1000 toolkits, tool search, context management, authentication, and a sandboxed workbench to help you build AI agents that turn intent into action.项目地址: https://gitcode.com/GitHub_Trending/co/composio创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
