Electric Agents 编程式运行时客户端createRuntimeServerClient 完整使用指南【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric本文导读createRuntimeServerClient()是 Electric Agents 面向应用服务的底层 HTTP 客户端用于在 handler 之外管理实体生命周期、发送消息、注册唤醒wake、管理调度与标签以及建立共享状态与外部数据源连接。读完本文你将掌握如何从应用服务、CLI、测试与集成代码中驱动 agent 运行时理解每个 API 背后的 REST 端点与实现细节并能够在ctx.spawn()/ctx.send()/ctx.observe()之外按需选择合适的客户端。本文基于electric-ax/agents-runtime包的源码runtime-server-client.ts与官方文档编写。什么时候需要使用编程式客户端在 Electric Agents 的架构中实体 handler 内部通常使用运行时上下文ctx提供的高层 APIctx.spawn()生成子实体、ctx.send()发送消息、ctx.observe()观察数据源、ctx.mkdb()创建本地数据库。这些 API 封装了权限、作用域与事务细节是 handler 内推荐的用法。但以下场景必须跳出 handler 的范围从外部驱动运行时应用服务application servicesWeb 后端需要根据业务事件生成实体、向其投递消息CLI 工具运维脚本需要列出实体、触发运行、设置标签或删除实体测试与集成代码测试需要独立于 handler 环境准备实体、注册数据源并断言结果编排层需要注册唤醒、调度、Webhook 订阅等横切能力。createRuntimeServerClient()就是为这些场景提供的低级 HTTP 客户端。文档明确建议handler 代码应优先使用ctx.spawn()、ctx.send()、ctx.observe()与ctx.mkdb()该客户端面向需要在 handler 之外管理实体的应用服务、测试、CLI 与集成代码。最小初始化客户端是纯函数工厂直接传入配置即可使用不依赖任何运行时上下文import { createRuntimeServerClient } from electric-ax/agents-runtime const client createRuntimeServerClient({ baseUrl: http://localhost:4437, })baseUrl指向 Electric Agents 服务器地址。所有请求通过fetch发出默认使用globalThis.fetch见 runtime-server-client.ts。客户端配置详解完整的配置接口定义如下源码位置runtime-server-client.tsinterface RuntimeServerClientConfig { baseUrl: string fetch?: typeof globalThis.fetch headers?: HeadersProvider writeTokenHeader?: ClaimTokenHeader track?: T(promise: PromiseT) PromiseT principalKey?: string }字段类型说明baseUrlstringElectric Agents 服务器的 Base URL必填。fetchtypeof globalThis.fetch可选的 fetch 实现。在测试或非标准运行时Node 18、Deno 之外的运行时、自定义传输中注入 mock 或替代实现。headersHeadersProvider静态或异步函数形式的头附加到每个请求。典型用途认证如authorization、租户tenant作用域隔离。writeTokenHeaderClaimTokenHeaderclaim 作用域写令牌write token的传输方式authorization、electric-claim-token或both默认authorization。trackT(promise: PromiseT) PromiseT所有请求的包装器用于遥测telemetry或维护 pending 状态。principalKeystring以Electric-Principal头发送的 principal key用于权限标识。配置项的实现语义从源码可以确认以下几点实现细节headers支持同步对象与异步函数resolveHeaders会判断config.headers是否为函数若是则await config.headers()获取基础头再将每次请求的initHeaders合并覆盖见 runtime-server-client.ts。这意味着你可以在请求时动态计算租户 ID 或刷新令牌。principalKey恒定为每个请求注入Electric-Principal头头部常量定义在源码中const ELECTRIC_PRINCIPAL_HEADER electric-principalruntime-server-client.ts。这用于声明请求方身份与权限系统配合。writeTokenHeader决定 claim 写令牌的放置位置applyTokenHeader在authorization时写Bearer token在electric-claim-token时写同名头both时两个头都写若authorization已存在则不覆盖见 runtime-server-client.ts。track包裹所有请求 Promise通过track(fetchImpl(...))统一拦截可在请求开始/结束时记录指标或维护 in-flight 集合。实体生命周期管理实体Entity是 Electric Agents 的基本执行单元通过/{type}/{id}形式的 URL 标识。客户端提供完整的生命周期 API。spawnEntity生成实体const info await client.spawnEntity({ type: horton, id: onboarding, args: { timezone: Europe/London }, initialMessage: Help me get started., tags: { project: docs }, sandbox: { profile: local, scope: entity }, }) console.log(info.entityUrl) // /horton/onboardingspawnEntity()底层调用PUT /_electric/entities/type/id见 runtime-server-client.ts返回RuntimeEntityInfointerface RuntimeEntityInfo { entityUrl: string entityType?: string streamPath: string }entityUrl实体的规范 URL形如/horton/onboardingentityType实体类型如hortonstreamPath实体的主 stream 路径可用于订阅其状态流。幂等性spawnEntity()对已存在的/{type}/{id}URL 是幂等的。源码中若服务器返回409冲突客户端会尝试从冲突响应体中的error.details.entity解析实体信息若解析失败则回退到GET /_electric/entities/type/id重新获取实体runtime-server-client.ts。也就是说重复 spawn 同一个实体不会报错而是返回已有实体的信息——这是典型的创建或复用语义。完整选项interface SpawnEntityOptions { type: string id: string args?: Recordstring, unknown parentUrl?: string initialMessage?: unknown initialMessageType?: string tags?: Recordstring, string sandbox?: { profile?: string key?: string scope?: entity | wake persistent?: boolean owner?: boolean inherit?: boolean } dispatch_policy?: DispatchPolicy wake?: { subscriberUrl: string condition: | runFinished | { on: change collections?: string[] ops?: Arrayinsert | update | delete } debounceMs?: number timeoutMs?: number includeResponse?: boolean manifestKey?: string } }关键字段语义type/id必填确定实体 URL。id可自由命名如onboarding、research-123args实体构造参数作为entity.args暴露给 handlerparentUrl将新实体挂为某实体的子实体对应 body 字段parentinitialMessageinitialMessageTypespawn 时立即投递的首条消息及其类型tags键值标签用于实体检索与 membership 观察entities({ tags })sandbox沙箱选择。profile指定沙箱模板scope可为entity实体级或wake唤醒级persistent控制是否持久key指定共享沙箱键inherit表示继承父实体解析出的沙箱owner标记实体是否为沙箱属主dispatch_policy分发策略支持 runner 目标或 webhook 目标两种形态type DispatchPolicy | { targets: [{ type: runner; runnerId: string; subscription_id?: string }] } | { targets: [{ type: webhook; url: string; subscription_id?: string }] }wakespawn 时一并注册唤醒订阅subscriber 指向subscriberUrlcondition可为runFinished运行完成时或{ on: change, collections, ops }集合变更时debounceMs/timeoutMs控制防抖与超时includeResponse是否把运行响应包含进唤醒消息manifestKey指定 manifest 锚点。forkEntity从最近一次运行分叉forkEntity()包装POST /_electric/entities/type/id/fork从源实体最近一次已完成的运行latest_completed_run锚点创建新实体见 runtime-server-client.tsconst fork await client.forkEntity({ sourceEntityUrl: /horton/onboarding, instanceId: onboarding-variant, initialMessage: { text: Try a different approach. }, tags: { branch: variant }, }) console.log(fork.entityUrl)interface ForkEntityOptions { sourceEntityUrl: string instanceId?: string parent?: string wake?: { subscriberUrl: string condition: RegisterWakeOptions[condition] debounceMs?: number timeoutMs?: number includeResponse?: boolean manifestKey?: string } initialMessage?: unknown tags?: Recordstring, string }sourceEntityUrl源实体 URLinstanceId映射到服务端 body 字段instance_id用于区分同一源的不同分叉实例parent使新分叉成为该 URL 的子实体wake在 fork 时注册订阅配合父实体 manifest 上的锚定唤醒使用与spawn模型一致initialMessage将 fork 与 send 合并为一次往返但并非原子操作——它是在 fork 创建成功、dispatch 订阅链接之后才发送的因此部分失败可能留下一个空闲的idle分叉实体。源码注释明确建议handler 内应优先使用ctx.fork()或ctx.forkSelf()完成分叉它们具备完整的权限与生命周期语义forkEntity()主要面向 handler 之外的应用/编排代码。getEntity获取实体信息const info await client.getEntity(/horton/onboarding) // { entityUrl, entityType, streamPath }底层调用GET /_electric/entities/entityUrlruntime-server-client.ts返回与spawnEntity相同的RuntimeEntityInfo。若服务器响应缺少url或主 stream 路径客户端会抛出missing entity url or main stream path错误requireEntityInfo校验runtime-server-client.ts。deleteEntity删除实体await client.deleteEntity(/horton/onboarding)deleteEntity()的底层实现是向实体发送SIGKILL信号reason 为Runtime child cleanup并且删除一个已经不存在的实体会被视为成功——源码中捕获(404)错误后直接返回见 runtime-server-client.ts因此可以安全地对同一实体多次调用 delete。消息投递sendEntityMessage发送消息await client.sendEntityMessage({ targetUrl: /horton/onboarding, payload: What changed since last time?, type: user_message, mode: queued, })底层调用POST /_electric/entities/targetUrl/sendruntime-server-client.ts。interface SendEntityMessageOptions { targetUrl: string payload: unknown type?: string afterMs?: number mode?: immediate | queued | paused | steer position?: string fromPrincipal?: string fromAgent?: string writeToken?: string }各字段语义字段说明targetUrl目标实体 URL必填。payload消息体任意 JSON 值字符串、对象等。type消息类型如user_message不传时使用默认类型。afterMs延迟投递请求服务器在指定毫秒后再投递该消息。mode服务器队列与施加消息的方式immediate立即、queued入队、paused暂停态投递、steer引导。position消息在队列中的位置控制。fromPrincipal高级字段声明消息来源 principal映射到请求体from_principal用于 claim 作用域的运行时写入。fromAgent高级字段声明消息来源 agent映射到from_agent。writeTokenclaim 作用域写令牌通过writeTokenHeader指定的头传输。源码细节fromPrincipal/fromAgent/afterMs/mode/position仅在定义时写入请求体if (x ! undefined) body.x xwriteToken通过applyTokenHeader写入认证头。响应非 2xx 时抛出格式化的错误含状态码与响应文本。生命周期信号Signalsawait client.signalEntity({ entityUrl: /horton/onboarding, signal: SIGINT, reason: User stopped the current run, })底层调用POST /_electric/entities/entityUrl/signalruntime-server-client.ts。signal的类型为EntitySignal从entity-schema.ts导出常见值如SIGINT中断当前运行、SIGKILL强制终止。可选字段reason原因说明会写入实体运行记录与payload附加数据。成功时返回{ txid: number }——信号以事务形式写入实体流。正如上文所述deleteEntity()正是通过向实体发送SIGKILL信号实现的。附件Attachments附件通过实体路由上传存储在私有的附件流中并由 manifest 条目引用。客户端提供创建与读取两个方法实现见 runtime-server-client.tsconst { attachment } await client.createAttachment({ entityUrl: /horton/onboarding, attachment: { bytes: imageBytes, mimeType: image/png, filename: diagram.png, subject: { type: inbox, key: message-1 }, role: input, }, }) const bytes await client.readAttachment({ entityUrl: /horton/onboarding, id: attachment.id, })createAttachment()调用POST /_electric/entities/entityUrl/attachments使用multipart/form-data上传。源码中bytes支持Blob、Uint8Array或原始字节统一封装为Blobsubject附件归属如 inbox 中的某条消息与role如input以 JSON 序列化字段提交。返回{ txid, attachment: ManifestAttachmentEntry }其中attachment.id供后续读取。readAttachment()调用GET /_electric/entities/entityUrl/attachments/id返回Uint8Array原始字节由调用方决定如何解码如new TextDecoder().decode(bytes)。附件与 manifest 条目、消息队列的关系可参考 attachments.md。共享状态流Shared State共享状态Shared State是实体间协作的数据层。客户端提供两个方法实现见 runtime-server-client.tsconst streamPath await client.ensureSharedStateStream(research-123) // /_electric/shared-state/research-123 const samePath client.getSharedStateStreamPath(research-123)getSharedStateStreamPath(id)同步纯函数将共享状态 ID 规范化为 stream 路径/_electric/shared-state/id源码 runtime-server-client.ts不发起网络请求。ensureSharedStateStream(id, ownerEntityUrl?)调用PUT创建或确认存在该 streamContent-Type 为application/json若传了ownerEntityUrl会附加electric-owner-entity头声明属主实体。当应用代码需要在实体连接共享状态之前先创建好该 stream 时应使用ensureSharedStateStream()——这是文档明确给出的使用时机。服务端返回409已存在时视为成功。唤醒与数据源Wakes and Sources唤醒wake是 Electric Agents 中数据源变化 → 实体被唤醒执行的核心机制。客户端提供从外部注册唤醒与数据源的低级操作。registerWake注册唤醒订阅registerWake()创建从某个源 stream 到订阅者实体的唤醒订阅实现见 runtime-server-client.ts调用POST /_electric/wake。运行完成唤醒source 实体每次 runFinished 唤醒 subscriberawait client.registerWake({ subscriberUrl: /coordinator/research, sourceUrl: /worker/analyst/main, condition: runFinished, includeResponse: true, })集合变更唤醒观察指定集合的插入/更新await client.registerWake({ subscriberUrl: /monitor/main, sourceUrl: /horton/onboarding/main, condition: { on: change, collections: [runs, texts], ops: [insert, update], }, debounceMs: 250, })RegisterWakeOptions与spawnEntity中的wake字段同构interface RegisterWakeOptions { subscriberUrl: string sourceUrl: string condition: | runFinished | { on: change collections?: string[] ops?: Arrayinsert | update | delete } debounceMs?: number timeoutMs?: number includeResponse?: boolean manifestKey?: string }subscriberUrl被唤醒的实体subscriberURLsourceUrl唤醒源的 stream URLsourceconditionrunFinished或集合变更条件。debounceMs对变更型唤醒做防抖合并如上面示例合并 250ms 内的变更includeResponse是否将 source 的运行响应包含进唤醒消息。对应的注销操作为unregisterWake({ subscriberUrl, sourceUrl?, manifestKey? })调用POST /_electric/wake/unregisterruntime-server-client.ts。ensureCronStream定时触发流const streamUrl await client.ensureCronStream( 0 9 * * *, Europe/London )调用POST /_electric/observations/cron/ensure-stream请求体{ expression, timezone }返回 cron stream 的 URLruntime-server-client.ts。之后实体可通过observe(streamUrl)在每次 cron 触发时被唤醒。cron 表达式的解析与 stream 路径规范见 cron-utils.ts。ensureEntitiesMembershipStream按标签观察实体const source await client.ensureEntitiesMembershipStream({ project: docs }) // { streamUrl, sourceRef }调用POST /_electric/observations/entities/ensure-stream请求体为{ tags }返回{ streamUrl, sourceRef }runtime-server-client.ts。这是observe(entities({ tags }))背后的低级实现见 observation-sources.ts 中entities()的定义标签归一化后由sourceRefForTags生成 source refstream 路径为getEntitiesStreamPath(sourceRef)。当实体的标签集合匹配时该 stream 会产出 membership 变更事件。registerPgSyncSource接入 Postgres 同步数据源const source await client.registerPgSyncSource({ url: http://localhost:3000/v1/shape, table: todos, where: project_id $1, params: [docs], }) // { streamUrl, sourceRef }调用POST /_electric/pg-sync/registerruntime-server-client.ts。这是observe(pgSync({ url, table, where, params }))的低级实现——服务器会把 Postgres 的 shape经 Electric 的 shape 同步协议转换为 Electric Agents 的观察流实体的changes集合会收到pg_sync_change类型的行集合定义见 observation-sources.ts。PgSyncOptions完整字段observation-sources.ts字段类型说明urlstringshape 端点 URL如http://localhost:3000/v1/shape。tablestring要同步的 Postgres 表名必填。columnsstring[]可选的列过滤。wherestringSQL 条件如project_id $1。paramsstring[] \| Recordstring, string参数绑定与where中的占位符对应。replicadefault \| full副本模式默认default。metadataPgSyncRequestMetadata请求元数据tenantId、principal 信息、entityUrl 等。底层注意点sourceRefForPgSync()会基于规范化后的 options 计算哈希作为 source 身份身份相关的 metadatatenant、principal、观察实体会改变 source 身份而wakeId、runtimeConsumerId、streamPath等每次唤醒/运行都会变化的临时字段被排除在身份之外observation-sources.ts——这样不同的 principal 不会共享同一个绑定到首个注册者的 bridge同时同一观察可在多次唤醒间复用 bridge。移除观察通过 source ref 删除某实体的 pg-sync 观察await client.removePgSyncObservation({ entityUrl: /horton/onboarding, sourceRef: source.sourceRef, })底层调用DELETE /_electric/entities/entityUrl/pg-sync-observations/sourceRefruntime-server-client.ts返回{ txid }。仓库测试 runtime-server-client-pg-sync.test.ts 验证了该流程客户端会把{ options }POST 到/_electric/pg-sync/register并解析{ sourceRef, streamUrl }响应非 2xx 时抛出包含状态码与响应文本的错误如registerPgSyncSource failed (400): bad table。Webhook 数据源Webhook 数据源暴露 webhook 支撑的 feedagent 可以订阅。客户端提供三个方法实现见 runtime-server-client.tsconst sources await client.listWebhookSources() await client.subscribeToWebhookSource({ entityUrl: /horton/onboarding, id: github-main, webhookKey: github, bucketKey: repo, params: { repo: electric-sql/electric }, lifetime: { kind: until_entity_stopped }, }) await client.unsubscribeFromWebhookSource({ entityUrl: /horton/onboarding, id: github-main, })listWebhookSources()GET /_electric/webhook-sources返回WebhookSourceContract[]含webhookKey、endpointKey、status、buckets、revision等契约信息。subscribeToWebhookSource()PUT /_electric/entities/entityUrl/webhook-source-subscriptions/id。id可不传由客户端基于webhookKey/bucketKey/params/filterKey自动构造buildWebhookSourceSubscriptionId见 webhook-sources.ts。lifetime支持三种形态webhook-sources.ts{ kind: until_entity_stopped }跟随实体生命周期实体停止即失效默认{ kind: expires_at, at: string }指定过期时间{ kind: manual }手动管理。unsubscribeFromWebhookSource()DELETE同一路由按订阅 id 移除。Webhook 契约的 bucket 定义paramsSchema用于参数校验仓库中用 Ajv 做运行时校验见 webhook-sources.ts。更完整的 webhook 源使用可参考 webhook-sources.md。调度Schedules调度存储于实体的 manifest 上相关 API 返回写入事务 IDwrite transaction id。三个方法分别对应 cron 调度、一次性延迟调度与删除实现见 runtime-server-client.tsawait client.upsertCronSchedule({ entityUrl: /horton/onboarding, id: daily-checkin, expression: 0 9 * * *, timezone: Europe/London, payload: Run the daily check-in., }) await client.upsertFutureSendSchedule({ entityUrl: /horton/onboarding, id: follow-up, fireAt: new Date(Date.now() 60_000).toISOString(), payload: Follow up now., }) await client.deleteSchedule({ entityUrl: /horton/onboarding, id: follow-up, })upsertCronSchedule()PUT /_electric/entities/entityUrl/schedules/id请求体{ scheduleType: cron, expression, timezone, payload, debounceMs?, timeoutMs? }。id为调度标识upsert 语义存在则更新expression为 cron 表达式payload为触发时投递给实体的消息内容。upsertFutureSendSchedule()同一路由但scheduleType: future_send请求体含fireAtISO 时间戳触发时刻与可选targetUrl目标实体默认当前实体、messageType。fireAt到达时服务器自动向实体投递一条延迟消息。deleteSchedule()DELETE /_electric/entities/entityUrl/schedules/id。标签管理Tagsawait client.setTag(/horton/onboarding, title, Onboarding, writeToken) await client.deleteTag(/horton/onboarding, title, writeToken)setTag()POST /_electric/entities/entityUrl/tags/keybody 为{ value }deleteTag()DELETE /_electric/entities/entityUrl/tags/key。两者的请求都会通过authedRequest将writeToken写入认证头runtime-server-client.ts返回{ txid? }。重要提示setTag()与deleteTag()主要面向已持有当前 claim 作用域写令牌的 handler/运行时自有流程。文档明确建议外部客户端优先使用sendEntityMessage()且只写入实体的 inbox而不要直接写实体的状态——直接操作实体状态会绕过 handler 的权限与一致性语义容易造成状态损坏或权限逃逸。如何选择合适的客户端文档给出了四类客户端的选用决策表API使用时机ctx.spawn/send/observe你正处于实体 handler 内部。createAgentsClient()你需要观察 stream 并驱动 UI 状态如 React 前端订阅实体/共享状态流。createRuntimeServerClient()你需要在 handler 之外管理实体、消息、唤醒、调度或标签。electric-ax/entity-stream-db你需要 CLI 风格的实体流加载器并带有close()方法。对应源码位置createAgentsClient导出自 agents-client.tscreateEntityStreamDB导出自 entity-stream-db.ts三者与createRuntimeServerClient一并从electric-ax/agents-runtime的入口 index.ts 导出。简单判断方法代码运行在实体运行流内部有ctx→ 用ctx.*高层 API代码在浏览器/前端需要把 agent 状态渲染成 UI → 用createAgentsClient()配合 clients-and-react.md代码在服务端应用/CLI/测试需要管理实体、消息、唤醒、调度、标签或建立共享状态/外部数据源 → 用createRuntimeServerClient()需要按 CLI 习惯加载实体流并显式close()→ 用electric-ax/entity-stream-db。实战示例从应用服务编排一次完整流程将以上 API 串成一个真实可运行的编排流程应用服务收到用户请求后创建实体、发送消息、注册调度并最终清理import { createRuntimeServerClient } from electric-ax/agents-runtime const client createRuntimeServerClient({ baseUrl: http://localhost:4437, headers: async () ({ authorization: Bearer ${await getServiceToken()}, }), principalKey: svc/onboarding-orchestrator, writeTokenHeader: electric-claim-token, track: (p) telemetry.track(p), }) // 1. 创建或复用实体 const { entityUrl } await client.spawnEntity({ type: horton, id: user-42-onboarding, args: { userId: 42 }, tags: { project: docs, env: prod }, }) // 2. 投递首条消息 await client.sendEntityMessage({ targetUrl: entityUrl, payload: { step: welcome, text: 开始引导流程 }, type: user_message, mode: queued, }) // 3. 注册每日 cron 调度 await client.upsertCronSchedule({ entityUrl, id: daily-checkin, expression: 0 9 * * *, timezone: Europe/London, payload: { kind: checkin }, }) // 4. 共享状态实体连接前先建好 stream const sharedPath await client.ensureSharedStateStream(onboarding-42) console.log(sharedPath) // /_electric/shared-state/onboarding-42 // 5. 清理幂等 await client.deleteEntity(entityUrl) await client.deleteSchedule({ entityUrl, id: daily-checkin })这个流程演示了文档强调的两个设计要点幂等性重复 spawn 复用实体、删除已不存在的实体视为成功与分工外部编排只负责创建、投递、调度、清理业务逻辑仍在实体 handler 内通过ctx.*完成。常见问题与注意事项Q1spawnEntity返回的streamPath有什么用它是实体的主 stream 路径可配合createAgentsClient()或electric-ax/entity-stream-db订阅实体的状态变化用于 UI 展示或日志追踪。Q2sendEntityMessage的mode应该怎么选immediate立即投递不排队queued进入实体消息队列按序处理paused在实体暂停期间暂存、恢复后处理steer用于运行时内部引导。默认行为取决于服务器对消息类型的配置通常业务消息用queued保证顺序。Q3为什么文档不建议外部客户端直接setTag标签写操作需要当前 claim 作用域的写令牌writeToken这是运行时/ handler 持有的凭据外部代码直接写实体状态会绕过实体自身的权限校验与一致性流程。外部代码应通过sendEntityMessage告知实体去做什么由实体自己决定是否改标签。Q4ensureSharedStateStream与getSharedStateStreamPath的区别前者是网络操作PUT 创建/确认 stream 存在后者是本地纯函数仅拼接路径字符串不会发起请求。需要在实体连接前确保 stream 存在时用前者。Q5registerPgSyncSource返回的sourceRef是什么它是 Postgres 观察源的规范化哈希身份标识由表、列、where、params 及身份性 metadata 计算得出用于removePgSyncObservation时精准移除对应观察也是 manifest 中 source 条目的 key 组成部分source:pgSync:sourceRef形式。延伸阅读spawning-and-coordinating.mdctx.spawn()/ctx.fork()等 handler 内的高层实体编排 APIwaking-entities.md唤醒机制的概念与使用shared-state.md共享状态的读写与观察signals.md信号SIGINT/SIGKILL 等对实体运行的控制语义webhook-sources.mdWebhook 数据源订阅的完整流程attachments.md附件上传/读取与 manifest 条目clients-and-react.mdcreateAgentsClient()与 React 集成permissions-and-principals.mdprincipal、claim 写令牌与权限模型源码runtime-server-client.ts、observation-sources.ts、webhook-sources.ts、测试 runtime-server-client-pg-sync.test.ts。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Electric Agents 编程式运行时客户端：createRuntimeServerClient 完整使用指南
Electric Agents 编程式运行时客户端createRuntimeServerClient 完整使用指南【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric本文导读createRuntimeServerClient()是 Electric Agents 面向应用服务的底层 HTTP 客户端用于在 handler 之外管理实体生命周期、发送消息、注册唤醒wake、管理调度与标签以及建立共享状态与外部数据源连接。读完本文你将掌握如何从应用服务、CLI、测试与集成代码中驱动 agent 运行时理解每个 API 背后的 REST 端点与实现细节并能够在ctx.spawn()/ctx.send()/ctx.observe()之外按需选择合适的客户端。本文基于electric-ax/agents-runtime包的源码runtime-server-client.ts与官方文档编写。什么时候需要使用编程式客户端在 Electric Agents 的架构中实体 handler 内部通常使用运行时上下文ctx提供的高层 APIctx.spawn()生成子实体、ctx.send()发送消息、ctx.observe()观察数据源、ctx.mkdb()创建本地数据库。这些 API 封装了权限、作用域与事务细节是 handler 内推荐的用法。但以下场景必须跳出 handler 的范围从外部驱动运行时应用服务application servicesWeb 后端需要根据业务事件生成实体、向其投递消息CLI 工具运维脚本需要列出实体、触发运行、设置标签或删除实体测试与集成代码测试需要独立于 handler 环境准备实体、注册数据源并断言结果编排层需要注册唤醒、调度、Webhook 订阅等横切能力。createRuntimeServerClient()就是为这些场景提供的低级 HTTP 客户端。文档明确建议handler 代码应优先使用ctx.spawn()、ctx.send()、ctx.observe()与ctx.mkdb()该客户端面向需要在 handler 之外管理实体的应用服务、测试、CLI 与集成代码。最小初始化客户端是纯函数工厂直接传入配置即可使用不依赖任何运行时上下文import { createRuntimeServerClient } from electric-ax/agents-runtime const client createRuntimeServerClient({ baseUrl: http://localhost:4437, })baseUrl指向 Electric Agents 服务器地址。所有请求通过fetch发出默认使用globalThis.fetch见 runtime-server-client.ts。客户端配置详解完整的配置接口定义如下源码位置runtime-server-client.tsinterface RuntimeServerClientConfig { baseUrl: string fetch?: typeof globalThis.fetch headers?: HeadersProvider writeTokenHeader?: ClaimTokenHeader track?: T(promise: PromiseT) PromiseT principalKey?: string }字段类型说明baseUrlstringElectric Agents 服务器的 Base URL必填。fetchtypeof globalThis.fetch可选的 fetch 实现。在测试或非标准运行时Node 18、Deno 之外的运行时、自定义传输中注入 mock 或替代实现。headersHeadersProvider静态或异步函数形式的头附加到每个请求。典型用途认证如authorization、租户tenant作用域隔离。writeTokenHeaderClaimTokenHeaderclaim 作用域写令牌write token的传输方式authorization、electric-claim-token或both默认authorization。trackT(promise: PromiseT) PromiseT所有请求的包装器用于遥测telemetry或维护 pending 状态。principalKeystring以Electric-Principal头发送的 principal key用于权限标识。配置项的实现语义从源码可以确认以下几点实现细节headers支持同步对象与异步函数resolveHeaders会判断config.headers是否为函数若是则await config.headers()获取基础头再将每次请求的initHeaders合并覆盖见 runtime-server-client.ts。这意味着你可以在请求时动态计算租户 ID 或刷新令牌。principalKey恒定为每个请求注入Electric-Principal头头部常量定义在源码中const ELECTRIC_PRINCIPAL_HEADER electric-principalruntime-server-client.ts。这用于声明请求方身份与权限系统配合。writeTokenHeader决定 claim 写令牌的放置位置applyTokenHeader在authorization时写Bearer token在electric-claim-token时写同名头both时两个头都写若authorization已存在则不覆盖见 runtime-server-client.ts。track包裹所有请求 Promise通过track(fetchImpl(...))统一拦截可在请求开始/结束时记录指标或维护 in-flight 集合。实体生命周期管理实体Entity是 Electric Agents 的基本执行单元通过/{type}/{id}形式的 URL 标识。客户端提供完整的生命周期 API。spawnEntity生成实体const info await client.spawnEntity({ type: horton, id: onboarding, args: { timezone: Europe/London }, initialMessage: Help me get started., tags: { project: docs }, sandbox: { profile: local, scope: entity }, }) console.log(info.entityUrl) // /horton/onboardingspawnEntity()底层调用PUT /_electric/entities/type/id见 runtime-server-client.ts返回RuntimeEntityInfointerface RuntimeEntityInfo { entityUrl: string entityType?: string streamPath: string }entityUrl实体的规范 URL形如/horton/onboardingentityType实体类型如hortonstreamPath实体的主 stream 路径可用于订阅其状态流。幂等性spawnEntity()对已存在的/{type}/{id}URL 是幂等的。源码中若服务器返回409冲突客户端会尝试从冲突响应体中的error.details.entity解析实体信息若解析失败则回退到GET /_electric/entities/type/id重新获取实体runtime-server-client.ts。也就是说重复 spawn 同一个实体不会报错而是返回已有实体的信息——这是典型的创建或复用语义。完整选项interface SpawnEntityOptions { type: string id: string args?: Recordstring, unknown parentUrl?: string initialMessage?: unknown initialMessageType?: string tags?: Recordstring, string sandbox?: { profile?: string key?: string scope?: entity | wake persistent?: boolean owner?: boolean inherit?: boolean } dispatch_policy?: DispatchPolicy wake?: { subscriberUrl: string condition: | runFinished | { on: change collections?: string[] ops?: Arrayinsert | update | delete } debounceMs?: number timeoutMs?: number includeResponse?: boolean manifestKey?: string } }关键字段语义type/id必填确定实体 URL。id可自由命名如onboarding、research-123args实体构造参数作为entity.args暴露给 handlerparentUrl将新实体挂为某实体的子实体对应 body 字段parentinitialMessageinitialMessageTypespawn 时立即投递的首条消息及其类型tags键值标签用于实体检索与 membership 观察entities({ tags })sandbox沙箱选择。profile指定沙箱模板scope可为entity实体级或wake唤醒级persistent控制是否持久key指定共享沙箱键inherit表示继承父实体解析出的沙箱owner标记实体是否为沙箱属主dispatch_policy分发策略支持 runner 目标或 webhook 目标两种形态type DispatchPolicy | { targets: [{ type: runner; runnerId: string; subscription_id?: string }] } | { targets: [{ type: webhook; url: string; subscription_id?: string }] }wakespawn 时一并注册唤醒订阅subscriber 指向subscriberUrlcondition可为runFinished运行完成时或{ on: change, collections, ops }集合变更时debounceMs/timeoutMs控制防抖与超时includeResponse是否把运行响应包含进唤醒消息manifestKey指定 manifest 锚点。forkEntity从最近一次运行分叉forkEntity()包装POST /_electric/entities/type/id/fork从源实体最近一次已完成的运行latest_completed_run锚点创建新实体见 runtime-server-client.tsconst fork await client.forkEntity({ sourceEntityUrl: /horton/onboarding, instanceId: onboarding-variant, initialMessage: { text: Try a different approach. }, tags: { branch: variant }, }) console.log(fork.entityUrl)interface ForkEntityOptions { sourceEntityUrl: string instanceId?: string parent?: string wake?: { subscriberUrl: string condition: RegisterWakeOptions[condition] debounceMs?: number timeoutMs?: number includeResponse?: boolean manifestKey?: string } initialMessage?: unknown tags?: Recordstring, string }sourceEntityUrl源实体 URLinstanceId映射到服务端 body 字段instance_id用于区分同一源的不同分叉实例parent使新分叉成为该 URL 的子实体wake在 fork 时注册订阅配合父实体 manifest 上的锚定唤醒使用与spawn模型一致initialMessage将 fork 与 send 合并为一次往返但并非原子操作——它是在 fork 创建成功、dispatch 订阅链接之后才发送的因此部分失败可能留下一个空闲的idle分叉实体。源码注释明确建议handler 内应优先使用ctx.fork()或ctx.forkSelf()完成分叉它们具备完整的权限与生命周期语义forkEntity()主要面向 handler 之外的应用/编排代码。getEntity获取实体信息const info await client.getEntity(/horton/onboarding) // { entityUrl, entityType, streamPath }底层调用GET /_electric/entities/entityUrlruntime-server-client.ts返回与spawnEntity相同的RuntimeEntityInfo。若服务器响应缺少url或主 stream 路径客户端会抛出missing entity url or main stream path错误requireEntityInfo校验runtime-server-client.ts。deleteEntity删除实体await client.deleteEntity(/horton/onboarding)deleteEntity()的底层实现是向实体发送SIGKILL信号reason 为Runtime child cleanup并且删除一个已经不存在的实体会被视为成功——源码中捕获(404)错误后直接返回见 runtime-server-client.ts因此可以安全地对同一实体多次调用 delete。消息投递sendEntityMessage发送消息await client.sendEntityMessage({ targetUrl: /horton/onboarding, payload: What changed since last time?, type: user_message, mode: queued, })底层调用POST /_electric/entities/targetUrl/sendruntime-server-client.ts。interface SendEntityMessageOptions { targetUrl: string payload: unknown type?: string afterMs?: number mode?: immediate | queued | paused | steer position?: string fromPrincipal?: string fromAgent?: string writeToken?: string }各字段语义字段说明targetUrl目标实体 URL必填。payload消息体任意 JSON 值字符串、对象等。type消息类型如user_message不传时使用默认类型。afterMs延迟投递请求服务器在指定毫秒后再投递该消息。mode服务器队列与施加消息的方式immediate立即、queued入队、paused暂停态投递、steer引导。position消息在队列中的位置控制。fromPrincipal高级字段声明消息来源 principal映射到请求体from_principal用于 claim 作用域的运行时写入。fromAgent高级字段声明消息来源 agent映射到from_agent。writeTokenclaim 作用域写令牌通过writeTokenHeader指定的头传输。源码细节fromPrincipal/fromAgent/afterMs/mode/position仅在定义时写入请求体if (x ! undefined) body.x xwriteToken通过applyTokenHeader写入认证头。响应非 2xx 时抛出格式化的错误含状态码与响应文本。生命周期信号Signalsawait client.signalEntity({ entityUrl: /horton/onboarding, signal: SIGINT, reason: User stopped the current run, })底层调用POST /_electric/entities/entityUrl/signalruntime-server-client.ts。signal的类型为EntitySignal从entity-schema.ts导出常见值如SIGINT中断当前运行、SIGKILL强制终止。可选字段reason原因说明会写入实体运行记录与payload附加数据。成功时返回{ txid: number }——信号以事务形式写入实体流。正如上文所述deleteEntity()正是通过向实体发送SIGKILL信号实现的。附件Attachments附件通过实体路由上传存储在私有的附件流中并由 manifest 条目引用。客户端提供创建与读取两个方法实现见 runtime-server-client.tsconst { attachment } await client.createAttachment({ entityUrl: /horton/onboarding, attachment: { bytes: imageBytes, mimeType: image/png, filename: diagram.png, subject: { type: inbox, key: message-1 }, role: input, }, }) const bytes await client.readAttachment({ entityUrl: /horton/onboarding, id: attachment.id, })createAttachment()调用POST /_electric/entities/entityUrl/attachments使用multipart/form-data上传。源码中bytes支持Blob、Uint8Array或原始字节统一封装为Blobsubject附件归属如 inbox 中的某条消息与role如input以 JSON 序列化字段提交。返回{ txid, attachment: ManifestAttachmentEntry }其中attachment.id供后续读取。readAttachment()调用GET /_electric/entities/entityUrl/attachments/id返回Uint8Array原始字节由调用方决定如何解码如new TextDecoder().decode(bytes)。附件与 manifest 条目、消息队列的关系可参考 attachments.md。共享状态流Shared State共享状态Shared State是实体间协作的数据层。客户端提供两个方法实现见 runtime-server-client.tsconst streamPath await client.ensureSharedStateStream(research-123) // /_electric/shared-state/research-123 const samePath client.getSharedStateStreamPath(research-123)getSharedStateStreamPath(id)同步纯函数将共享状态 ID 规范化为 stream 路径/_electric/shared-state/id源码 runtime-server-client.ts不发起网络请求。ensureSharedStateStream(id, ownerEntityUrl?)调用PUT创建或确认存在该 streamContent-Type 为application/json若传了ownerEntityUrl会附加electric-owner-entity头声明属主实体。当应用代码需要在实体连接共享状态之前先创建好该 stream 时应使用ensureSharedStateStream()——这是文档明确给出的使用时机。服务端返回409已存在时视为成功。唤醒与数据源Wakes and Sources唤醒wake是 Electric Agents 中数据源变化 → 实体被唤醒执行的核心机制。客户端提供从外部注册唤醒与数据源的低级操作。registerWake注册唤醒订阅registerWake()创建从某个源 stream 到订阅者实体的唤醒订阅实现见 runtime-server-client.ts调用POST /_electric/wake。运行完成唤醒source 实体每次 runFinished 唤醒 subscriberawait client.registerWake({ subscriberUrl: /coordinator/research, sourceUrl: /worker/analyst/main, condition: runFinished, includeResponse: true, })集合变更唤醒观察指定集合的插入/更新await client.registerWake({ subscriberUrl: /monitor/main, sourceUrl: /horton/onboarding/main, condition: { on: change, collections: [runs, texts], ops: [insert, update], }, debounceMs: 250, })RegisterWakeOptions与spawnEntity中的wake字段同构interface RegisterWakeOptions { subscriberUrl: string sourceUrl: string condition: | runFinished | { on: change collections?: string[] ops?: Arrayinsert | update | delete } debounceMs?: number timeoutMs?: number includeResponse?: boolean manifestKey?: string }subscriberUrl被唤醒的实体subscriberURLsourceUrl唤醒源的 stream URLsourceconditionrunFinished或集合变更条件。debounceMs对变更型唤醒做防抖合并如上面示例合并 250ms 内的变更includeResponse是否将 source 的运行响应包含进唤醒消息。对应的注销操作为unregisterWake({ subscriberUrl, sourceUrl?, manifestKey? })调用POST /_electric/wake/unregisterruntime-server-client.ts。ensureCronStream定时触发流const streamUrl await client.ensureCronStream( 0 9 * * *, Europe/London )调用POST /_electric/observations/cron/ensure-stream请求体{ expression, timezone }返回 cron stream 的 URLruntime-server-client.ts。之后实体可通过observe(streamUrl)在每次 cron 触发时被唤醒。cron 表达式的解析与 stream 路径规范见 cron-utils.ts。ensureEntitiesMembershipStream按标签观察实体const source await client.ensureEntitiesMembershipStream({ project: docs }) // { streamUrl, sourceRef }调用POST /_electric/observations/entities/ensure-stream请求体为{ tags }返回{ streamUrl, sourceRef }runtime-server-client.ts。这是observe(entities({ tags }))背后的低级实现见 observation-sources.ts 中entities()的定义标签归一化后由sourceRefForTags生成 source refstream 路径为getEntitiesStreamPath(sourceRef)。当实体的标签集合匹配时该 stream 会产出 membership 变更事件。registerPgSyncSource接入 Postgres 同步数据源const source await client.registerPgSyncSource({ url: http://localhost:3000/v1/shape, table: todos, where: project_id $1, params: [docs], }) // { streamUrl, sourceRef }调用POST /_electric/pg-sync/registerruntime-server-client.ts。这是observe(pgSync({ url, table, where, params }))的低级实现——服务器会把 Postgres 的 shape经 Electric 的 shape 同步协议转换为 Electric Agents 的观察流实体的changes集合会收到pg_sync_change类型的行集合定义见 observation-sources.ts。PgSyncOptions完整字段observation-sources.ts字段类型说明urlstringshape 端点 URL如http://localhost:3000/v1/shape。tablestring要同步的 Postgres 表名必填。columnsstring[]可选的列过滤。wherestringSQL 条件如project_id $1。paramsstring[] \| Recordstring, string参数绑定与where中的占位符对应。replicadefault \| full副本模式默认default。metadataPgSyncRequestMetadata请求元数据tenantId、principal 信息、entityUrl 等。底层注意点sourceRefForPgSync()会基于规范化后的 options 计算哈希作为 source 身份身份相关的 metadatatenant、principal、观察实体会改变 source 身份而wakeId、runtimeConsumerId、streamPath等每次唤醒/运行都会变化的临时字段被排除在身份之外observation-sources.ts——这样不同的 principal 不会共享同一个绑定到首个注册者的 bridge同时同一观察可在多次唤醒间复用 bridge。移除观察通过 source ref 删除某实体的 pg-sync 观察await client.removePgSyncObservation({ entityUrl: /horton/onboarding, sourceRef: source.sourceRef, })底层调用DELETE /_electric/entities/entityUrl/pg-sync-observations/sourceRefruntime-server-client.ts返回{ txid }。仓库测试 runtime-server-client-pg-sync.test.ts 验证了该流程客户端会把{ options }POST 到/_electric/pg-sync/register并解析{ sourceRef, streamUrl }响应非 2xx 时抛出包含状态码与响应文本的错误如registerPgSyncSource failed (400): bad table。Webhook 数据源Webhook 数据源暴露 webhook 支撑的 feedagent 可以订阅。客户端提供三个方法实现见 runtime-server-client.tsconst sources await client.listWebhookSources() await client.subscribeToWebhookSource({ entityUrl: /horton/onboarding, id: github-main, webhookKey: github, bucketKey: repo, params: { repo: electric-sql/electric }, lifetime: { kind: until_entity_stopped }, }) await client.unsubscribeFromWebhookSource({ entityUrl: /horton/onboarding, id: github-main, })listWebhookSources()GET /_electric/webhook-sources返回WebhookSourceContract[]含webhookKey、endpointKey、status、buckets、revision等契约信息。subscribeToWebhookSource()PUT /_electric/entities/entityUrl/webhook-source-subscriptions/id。id可不传由客户端基于webhookKey/bucketKey/params/filterKey自动构造buildWebhookSourceSubscriptionId见 webhook-sources.ts。lifetime支持三种形态webhook-sources.ts{ kind: until_entity_stopped }跟随实体生命周期实体停止即失效默认{ kind: expires_at, at: string }指定过期时间{ kind: manual }手动管理。unsubscribeFromWebhookSource()DELETE同一路由按订阅 id 移除。Webhook 契约的 bucket 定义paramsSchema用于参数校验仓库中用 Ajv 做运行时校验见 webhook-sources.ts。更完整的 webhook 源使用可参考 webhook-sources.md。调度Schedules调度存储于实体的 manifest 上相关 API 返回写入事务 IDwrite transaction id。三个方法分别对应 cron 调度、一次性延迟调度与删除实现见 runtime-server-client.tsawait client.upsertCronSchedule({ entityUrl: /horton/onboarding, id: daily-checkin, expression: 0 9 * * *, timezone: Europe/London, payload: Run the daily check-in., }) await client.upsertFutureSendSchedule({ entityUrl: /horton/onboarding, id: follow-up, fireAt: new Date(Date.now() 60_000).toISOString(), payload: Follow up now., }) await client.deleteSchedule({ entityUrl: /horton/onboarding, id: follow-up, })upsertCronSchedule()PUT /_electric/entities/entityUrl/schedules/id请求体{ scheduleType: cron, expression, timezone, payload, debounceMs?, timeoutMs? }。id为调度标识upsert 语义存在则更新expression为 cron 表达式payload为触发时投递给实体的消息内容。upsertFutureSendSchedule()同一路由但scheduleType: future_send请求体含fireAtISO 时间戳触发时刻与可选targetUrl目标实体默认当前实体、messageType。fireAt到达时服务器自动向实体投递一条延迟消息。deleteSchedule()DELETE /_electric/entities/entityUrl/schedules/id。标签管理Tagsawait client.setTag(/horton/onboarding, title, Onboarding, writeToken) await client.deleteTag(/horton/onboarding, title, writeToken)setTag()POST /_electric/entities/entityUrl/tags/keybody 为{ value }deleteTag()DELETE /_electric/entities/entityUrl/tags/key。两者的请求都会通过authedRequest将writeToken写入认证头runtime-server-client.ts返回{ txid? }。重要提示setTag()与deleteTag()主要面向已持有当前 claim 作用域写令牌的 handler/运行时自有流程。文档明确建议外部客户端优先使用sendEntityMessage()且只写入实体的 inbox而不要直接写实体的状态——直接操作实体状态会绕过 handler 的权限与一致性语义容易造成状态损坏或权限逃逸。如何选择合适的客户端文档给出了四类客户端的选用决策表API使用时机ctx.spawn/send/observe你正处于实体 handler 内部。createAgentsClient()你需要观察 stream 并驱动 UI 状态如 React 前端订阅实体/共享状态流。createRuntimeServerClient()你需要在 handler 之外管理实体、消息、唤醒、调度或标签。electric-ax/entity-stream-db你需要 CLI 风格的实体流加载器并带有close()方法。对应源码位置createAgentsClient导出自 agents-client.tscreateEntityStreamDB导出自 entity-stream-db.ts三者与createRuntimeServerClient一并从electric-ax/agents-runtime的入口 index.ts 导出。简单判断方法代码运行在实体运行流内部有ctx→ 用ctx.*高层 API代码在浏览器/前端需要把 agent 状态渲染成 UI → 用createAgentsClient()配合 clients-and-react.md代码在服务端应用/CLI/测试需要管理实体、消息、唤醒、调度、标签或建立共享状态/外部数据源 → 用createRuntimeServerClient()需要按 CLI 习惯加载实体流并显式close()→ 用electric-ax/entity-stream-db。实战示例从应用服务编排一次完整流程将以上 API 串成一个真实可运行的编排流程应用服务收到用户请求后创建实体、发送消息、注册调度并最终清理import { createRuntimeServerClient } from electric-ax/agents-runtime const client createRuntimeServerClient({ baseUrl: http://localhost:4437, headers: async () ({ authorization: Bearer ${await getServiceToken()}, }), principalKey: svc/onboarding-orchestrator, writeTokenHeader: electric-claim-token, track: (p) telemetry.track(p), }) // 1. 创建或复用实体 const { entityUrl } await client.spawnEntity({ type: horton, id: user-42-onboarding, args: { userId: 42 }, tags: { project: docs, env: prod }, }) // 2. 投递首条消息 await client.sendEntityMessage({ targetUrl: entityUrl, payload: { step: welcome, text: 开始引导流程 }, type: user_message, mode: queued, }) // 3. 注册每日 cron 调度 await client.upsertCronSchedule({ entityUrl, id: daily-checkin, expression: 0 9 * * *, timezone: Europe/London, payload: { kind: checkin }, }) // 4. 共享状态实体连接前先建好 stream const sharedPath await client.ensureSharedStateStream(onboarding-42) console.log(sharedPath) // /_electric/shared-state/onboarding-42 // 5. 清理幂等 await client.deleteEntity(entityUrl) await client.deleteSchedule({ entityUrl, id: daily-checkin })这个流程演示了文档强调的两个设计要点幂等性重复 spawn 复用实体、删除已不存在的实体视为成功与分工外部编排只负责创建、投递、调度、清理业务逻辑仍在实体 handler 内通过ctx.*完成。常见问题与注意事项Q1spawnEntity返回的streamPath有什么用它是实体的主 stream 路径可配合createAgentsClient()或electric-ax/entity-stream-db订阅实体的状态变化用于 UI 展示或日志追踪。Q2sendEntityMessage的mode应该怎么选immediate立即投递不排队queued进入实体消息队列按序处理paused在实体暂停期间暂存、恢复后处理steer用于运行时内部引导。默认行为取决于服务器对消息类型的配置通常业务消息用queued保证顺序。Q3为什么文档不建议外部客户端直接setTag标签写操作需要当前 claim 作用域的写令牌writeToken这是运行时/ handler 持有的凭据外部代码直接写实体状态会绕过实体自身的权限校验与一致性流程。外部代码应通过sendEntityMessage告知实体去做什么由实体自己决定是否改标签。Q4ensureSharedStateStream与getSharedStateStreamPath的区别前者是网络操作PUT 创建/确认 stream 存在后者是本地纯函数仅拼接路径字符串不会发起请求。需要在实体连接前确保 stream 存在时用前者。Q5registerPgSyncSource返回的sourceRef是什么它是 Postgres 观察源的规范化哈希身份标识由表、列、where、params 及身份性 metadata 计算得出用于removePgSyncObservation时精准移除对应观察也是 manifest 中 source 条目的 key 组成部分source:pgSync:sourceRef形式。延伸阅读spawning-and-coordinating.mdctx.spawn()/ctx.fork()等 handler 内的高层实体编排 APIwaking-entities.md唤醒机制的概念与使用shared-state.md共享状态的读写与观察signals.md信号SIGINT/SIGKILL 等对实体运行的控制语义webhook-sources.mdWebhook 数据源订阅的完整流程attachments.md附件上传/读取与 manifest 条目clients-and-react.mdcreateAgentsClient()与 React 集成permissions-and-principals.mdprincipal、claim 写令牌与权限模型源码runtime-server-client.ts、observation-sources.ts、webhook-sources.ts、测试 runtime-server-client-pg-sync.test.ts。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考