Electric Agents Webhook Sources 实战让 Agent 订阅外部事件流并被精准唤醒【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electricWebhook sources 是 Electric Agents 平台中连接外部世界与 Agent 实体的订阅机制它允许 Agent如 Horton 运行时发现并订阅 GitHub、Stripe、邮件、CI 等外部 Webhook 集成产生的事件流订阅关系持久化在实体 manifest 中当匹配的外部事件到达时平台会携带已水合hydrated的事件数据唤醒对应实体。读完本文你将掌握 Webhook source 的契约结构contract/bucket/filter、四个内置工具list_webhook_sources等的用法、编程式订阅的客户端 API以及唤醒载荷wake payload的完整数据形态并能将其接入自定义运行时。整体机制订阅—持久化—唤醒Webhook sources 的核心意图是让 Agent 订阅外部事件源如 GitHub、Stripe、email、CI 或其他 Webhook 集成。整个流程可以概括为三步发现Agent 调用list_webhook_sources获取当前可见的 webhook source 契约contract其中声明了可订阅的webhookKey、bucket路径模板分桶、参数 schema 与可选的命名过滤器订阅Agent或宿主代码调用subscribe_webhook_source订阅关系写入实体 manifestkind: source、sourceType: webhook的条目随实体流持久化唤醒当被订阅的源产生匹配事件时实体被唤醒唤醒载荷wake payload中直接携带水合后的 webhook 事件Horton 会把这些数据放进 trigger message让模型无需二次查询即可做出反应。内置的 Horton 运行时默认通过ctx.electricTools暴露 webhook-source 工具无需额外接线。契约WebhookSourceContract 与 Bucket一个 webhook source 契约描述了 Agent 可以订阅什么。完整类型定义见 webhook-sources.tstype WebhookSourceContract { serviceId?: string webhookKey: string sourceType: webhook endpointKey: string status: active | disabled | revoked label: string description?: string agentVisible: boolean buckets: WebhookSourceBucket[] updatedAt?: string revision: number }各字段的作用webhookKeyAgent 发起订阅时使用的源标识如githubendpointKey/sourceType底层端点标识与源类型固定为webhookstatus仅active状态的源可被订阅——从源码看resolveWebhookSourceSubscription 会对非agentVisible或非active的契约直接抛出is not active错误agentVisible控制该契约是否对 Agent 的list_webhook_sources可见revision契约版本号订阅时会以contractRevision记录便于后续追踪契约变更buckets路径模板分桶见下文。Bucket 描述路径模板与参数类型定义见 webhook-sources.tstype WebhookSourceBucket { key: string label: string description?: string pathTemplate: string paramsSchema: Recordstring, unknown eventTypes?: string[] filters?: WebhookSourceFilter[] }Bucket 的关键机制pathTemplate使用:name形式的模板占位符。运行时通过 renderWebhookSourceBucketPath 渲染正则/:([A-Za-z_][A-Za-z0-9_]*)/g逐个替换占位符占位符值缺失时抛出Missing bucket parameter错误值会被encodeURIComponent编码渲染结果不允许出现//、不允许以/开头也不能为空字符串paramsSchema是一份 JSON Schema订阅时用 Ajv 编译并对提交的params做严格校验见 validateBucketParams校验器按 schema 对象缓存在WeakMap中失败时错误信息会带具体路径如/repo is requiredeventTypes可选地描述该桶关注的事件类型filters是该桶下可用的命名过滤器。Agent 的标准工作方式是先调用list_webhook_sources然后使用其中声明的webhookKey、bucketKey、paramsSchema以及可选的filterKey来发起订阅——所有可订阅的表面都以契约声明为准。内置工具createWebhookSourceTools运行时工具工厂可添加四个工具实现位于 tools/webhook-sources.ts工具用途list_webhook_sources列出实体可订阅的外部 webhook 源。list_webhook_source_subscriptions列出本实体当前生效的订阅。subscribe_webhook_source让实体订阅某个源或桶。unsubscribe_webhook_source按 id 移除一个订阅。几个实现层面的细节值得注意列表来源list_webhook_sources实际调用运行时的listWebhookSources()其底层是 HTTPGET /_electric/webhook-sources见 runtime-server-client.ts返回服务端维护的契约数组订阅列表来源list_webhook_source_subscriptions不走网络而是直接扫描实体本地manifests集合通过 getWebhookSourceSubscriptions 过滤出kind source且sourceType webhook的 manifest 条目并按 id 排序。也就是说订阅清单是随实体流同步的跨唤醒across wakes依然可用幂等等待subscribe_webhook_source与unsubscribe_webhook_source执行后都会await db.utils.awaitTxId(txid, 10_000)等待本实体流确认该事务再返回订阅/删除结果保证工具返回值与本地状态一致日志每个工具调用都经withWebhookSourceToolLogging包装记录 start / success / failed 三个阶段及参数便于排查 Agent 行为。Horton 从内置运行时直接获得这些工具。自定义运行时可以用createWebhookSourceTools()提供它们或者通过createRuntimeHandler()传入createElectricToolsimport { createWebhookSourceTools } from electric-ax/agents-runtime/tools const runtime createRuntimeHandler({ baseUrl: http://localhost:4437, registry, createElectricTools: (context) createWebhookSourceTools(context), })注意内置运行时默认还会添加 schedule定时任务工具。如果你替换了createElectricTools想让 Horton 同时保留两种能力时需要把两套工具都包含进去。createElectricTools的接线点在 process-wake.ts运行时在处理唤醒时按需构建工具集。从工具发起订阅参数、确定性 ID 与生命周期subscribe_webhook_source接受如下输入类型见 webhook-sources.tstype WebhookSourceSubscriptionInput { id?: string webhookKey: string bucketKey?: string params?: Recordstring, unknown filterKey?: string lifetime?: SubscriptionLifetime reason?: string }id省略时的确定性派生运行时调用 buildWebhookSourceSubscriptionId用webhookKey、bucketKey缺省用root、filterKey拼接出规范化前缀小写化、非法字符替换为-、截断到 80 字符再对{webhookKey, bucketKey, params, filterKey}的稳定 JSON键排序序列化做 FNV-1a 哈希生成后缀。同一组参数重复订阅会派生出同一 id天然幂等省略bucketKey即订阅源根流root stream对应工具参数描述中Omit to subscribe to the source root streamfilterKey只能选择该源/桶声明过的命名过滤器resolveWebhookSourceSubscription 会校验 filter 是否存在于对应 bucket 的filters列表中reason是面向人的订阅理由会随 manifest 持久化并出现在后续唤醒载荷中。生命周期lifetime有三种取值type SubscriptionLifetime | { kind: until_entity_stopped } | { kind: expires_at; at: string } // at 为 ISO-8601 绝对时间 | { kind: manual }默认生命周期是until_entity_stopped——订阅随实体存活实体停止即失效expires_at允许设置明确到期时间manual表示需要显式取消。在工具侧lifetime 用 TypeBox schema 描述tools/webhook-sources.tsexpires_at.at要求 ISO-8601 字符串。编程式订阅createRuntimeServerClient宿主代码可以不经 Agent直接通过createRuntimeServerClient()返回的客户端订阅完整示例await client.subscribeToWebhookSource({ entityUrl: /horton/onboarding, webhookKey: github, bucketKey: repo, params: { repo: electric-sql/electric }, reason: Watch repo activity for this session, }) await client.unsubscribeFromWebhookSource({ entityUrl: /horton/onboarding, id: github-main, })用listWebhookSources()检查当前可用的契约const sources await client.listWebhookSources()从客户端实现看runtime-server-client.ts这三个方法对应的服务端路由为GET /_electric/webhook-sources—— 列出契约PUT entityRpcPath/webhook-source-subscriptions/id—— 创建/更新订阅请求体携带webhookKey、bucketKey、params、filterKey、lifetime、reason返回{ txid, subscription }DELETE entityRpcPath/webhook-source-subscriptions/id—— 按 id 删除订阅返回{ txid }。客户端在id缺省时同样调用buildWebhookSourceSubscriptionId派生确定性 id因此工具路径与编程式路径产生的 id 规则一致。服务端的订阅路由行为有对应测试覆盖webhook-source-subscriptions-route.test.ts。唤醒载荷HydratedWebhookSourceWake当被订阅的源触发时实体会被唤醒并携带水合后的 webhook-source 载荷type HydratedWebhookSourceWake { type: webhook_source_wake source: string sourceType: webhook endpointKey: string webhookKey: string subscription: { id: string bucketKey?: string params: Recordstring, unknown filterKey?: string reason?: string } bucket: string | null changes: Array{ collection: string kind: insert | update | delete key: string } events: WebhookEventRow[] missingEventKeys?: string[] }这个结构是如何被构建出来的从源码看process-wake.tswebhookSourceWakeInfoFromManifests检查当前唤醒事件它必须是一个wake事件且changes中包含webhook_event集合的变更然后遍历实体 manifests找到streamUrl与唤醒源一致的 webhook manifest还原出订阅信息sourceUrl、endpointKey、webhookKey、subscriptionId、params 等运行时通过wiringConfig.createSourceDb(sourceStreamUrl, ...)对该源流做一次预加载读取取出events行的实际数据buildHydratedWebhookSourceWake 把唤醒中声明的webhook_event变更 key 与实际读到的事件行做匹配产出events数组若某些声明的 key 在实际数据中读不到例如已被清理会收集到missingEventKeys中供处理方感知数据缺口。Handler 可以检查wake.payload或直接使用常规 agent context。Horton 会把水合后的 webhook-source 数据放入 trigger message 中——具体实现是 context-factory.ts 在构造触发消息时若存在hydratedWebhookSourceWake就将其序列化进消息文本模型可以立即基于事件内容做出反应而不需要再做第二次查询。相关行为在 process-wake.test.ts 与 context-factory.test.ts 中有测试覆盖。Manifest 条目订阅的持久化形态订阅以manifest行的形式存储kind: source并使用稳定的 manifest keywebhook-source:subscription-idbuildWebhookSourceManifestEntry 展示了完整条目结构key即webhook-source:idsourceRefendpointKey/bucketPath有桶时或endpointKey根流config.streamUrl订阅对应的实际流 URLconfig.webhookSource订阅全貌含id、webhookKey、bucketKey、params、filterKey、filterApplied、contractRevision、lifetime、reason、createdBy、createdAtwake唤醒规则{ on: change, collections: [webhook_event], ops: [insert] }——即当webhook_event集合出现 insert 时触发唤醒。因为订阅是 manifest 条目它随实体流同步与持久化这使得实体可以跨唤醒列出和管理自己的订阅list_webhook_source_subscriptions正是直接读 manifests 实现的。过滤器当前版本的 advisory 语义filterKey用于选择源声明的命名过滤器过滤器用于收窄外部 webhook 流。契约层面过滤器条件WebhookSourceFilterCondition已支持按collections、opsinsert/update/delete以及 CEL 表达式where描述。但需要注意当前的限制在本版本中过滤器是 advisory建议性的直到服务端 webhook 过滤器启用为止。订阅成功后条目中记录的filterApplied字段为false见 resolveWebhookSourceSubscription工具描述中也明确提示filters are advisory until server-side webhook filters are enabled。因此实践建议是Agent 在处理器中仍应防御性地处理不符合过滤预期的事件例如在 handler 里自行判断事件类型后再行动不能假设过滤已在链路上生效。小结Webhook sources 用契约 订阅 manifest 持久化 水合唤醒四个环节把外部事件源纳入了 Electric Agents 的同步体系契约WebhookSourceContract/WebhookSourceBucket声明可订阅面参数经 JSON Schema 校验、桶路径模板严格渲染工具与客户端两条订阅路径共用同一套确定性 id 派生与生命周期语义订阅落盘为webhook-source:idmanifest 条目唤醒时运行时自动预加载源流、匹配事件行并构建HydratedWebhookSourceWakeHorton 将其注入 trigger message实现事件到达即上下文就绪。如需进一步跟进实现细节可参考 webhook-sources.ts、tools/webhook-sources.ts、runtime-server-client.ts以及测试 webhook-sources.test.ts、webhook-source-tools.test.ts。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考