iii 自定义 Trigger Type 开发指南：从绑定既有事件源到发布自己的触发器
【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii
本文是一份面向 iii 工作流（worker）开发者的触发器（Trigger）实战指南。你将理解"触发器"在 iii 中消费与发布两种角色，学会用 Node/TypeScript、Python、Rust 三种 SDK 将函数绑定到http、cron、queue、state等既有触发器类型，并进一步以发布者身份声明自定义触发器类型（Trigger Type）、维护绑定路由表、附加 JSON Schema 契约，最终通过worker.trigger把事件分发到已绑定的函数。读完本文，你能够从零实现一个类似iii-http的迷你 HTTP 触发器发布者，并让其他 worker 的函数在你的事件源上运行。
理解"编写触发器"的两种角色
一个 worker 使用触发器有两种方式：
- 作为消费者（Consumer，最常见）：把 worker 自己的函数绑定到系统中已存在的触发器类型上，例如
http、cron（定时执行）、queue 消息（每条消息触发一次）、state状态变化，以及其他任何事件源。对应 API 是worker.registerTrigger({ type, function_id, config })。 - 作为发布者（Publisher，较少见）：从自己的 worker 注册一个全新的触发器类型，让其他 worker 可以把函数绑定到你 worker 发出的事件上，例如"HTTP 请求到达""webhook 命中""文件变化""数据库更新"。
本文的主体是第二种：如何创造你自己的触发器。如果只是想在新 worker 中使用既有触发器，参考 Using iii / Triggers；调用侧的机制（worker.trigger/iii trigger直接调用、TriggerAction变体、用条件门控、同一函数多个绑定）同样见该页。
把函数绑定到既有触发器类型
大多数 worker 消费其他 worker 已发布的触发器类型：http把函数暴露为 HTTP 端点，cron让函数按计划执行，queue 触发器让函数随每条消息触发，state让函数响应数据变化。绑定操作通过worker.registerTrigger({ type, function_id, config })完成：
Node / TypeScript
worker.registerTrigger({ type: "http", function_id: "math::add", config: { api_path: "/math/add", http_method: "POST" }, });Python
worker.register_trigger({ "type": "http", "function_id": "math::add", "config": {"api_path": "/math/add", "http_method": "POST"}, })Rust
use iii_sdk::RegisterTriggerInput; use serde_json::json; worker.register_trigger(RegisterTriggerInput { trigger_type: "http".into(), function_id: "math::add".into(), config: json!({ "api_path": "/math/add", "http_method": "POST" }), metadata: None, })?;config的具体结构由各触发器类型自行定义，并记录在对应发布 worker 的文档中。例如http类型的 config 就是{ api_path, http_method }（见后文 schema 一节）。一个值得注意的细节是：你可以在发布者 worker 尚未连入时就发起注册。引擎会乐观地存储该绑定，并在发布者加入网络后自动激活它——这种"乐观注册"的行为细节同样参见 Using iii / Triggers。
其他绑定机制——注销句柄（trigger.unregister()）、同一函数绑定多个触发器、用condition_function_id做门控、TriggerAction变体（Void、Enqueue等）——均属于调用侧主题，见 Using iii / Triggers。
为触发器绑定附加元数据
每个触发器绑定都可以携带一个可选的metadataJSON 对象，由消费者在注册时设置。引擎原样存储它，并在两个地方对外暴露：
- 发布者可见：发布者的
TriggerHandler.registerTrigger(config)回调会以config.metadata收到它，发布者可以根据消费者打上的标签做处理——优先级提示、审计标签、路由键等内部记账信息。 - 可被发现：
engine::triggers::list会在每个TriggerInfo上返回它，控制台（console）以及任何做服务发现的 worker 都能读到。
Node / TypeScript
worker.registerTrigger({ type: "http", function_id: "math::add", config: { api_path: "/math/add", http_method: "POST" }, metadata: { team: "platform", env: "staging" }, });Python
worker.register_trigger({ "type": "http", "function_id": "math::add", "config": {"api_path": "/math/add", "http_method": "POST"}, "metadata": {"team": "platform", "env": "staging"}, })Rust
use iii_sdk::RegisterTriggerInput; use serde_json::json; worker.register_trigger(RegisterTriggerInput { trigger_type: "http".into(), function_id: "math::add".into(), config: json!({ "api_path": "/math/add", "http_method": "POST" }), metadata: Some(json!({ "team": "platform", "env": "staging" })), })?;别混淆：metadata 与 schema
触发器类型本身没有 metadata 字段——metadata 是按绑定附加的，而非按类型。更关键的是不要把它与触发器类型的schema（trigger_request_format与call_request_format）混为一谈，二者的设定方和用途完全不同：
- Metadata：由消费者在每次绑定时（
worker.registerTrigger()调用）设置。它是引擎原样存储的自由标签袋，服务于发布者的记账与发现。例如绑定到http的消费者可能附带metadata: { team: "platform", env: "staging", on_call: "alice" }，发布者据此记录团队信息，engine::triggers::list也能在请求时返回这些信息。 - Schemas：由发布者在声明触发器类型时设置。它们描述消费者交互的 JSON 形状。例如
iii-http发布的http类型会声明：config（消费者绑定时传入的内容）：{ api_path, http_method }；- 调用载荷（绑定函数在每次请求时收到的东西）：
{ method, headers, query_params, body }。
声明一个触发器类型（成为发布者）
前面的内容都是"消费者"视角：你的 worker 函数被绑定到其他 worker 发布的类型上。现在角色反转：你的 worker 是发布者，你希望其他 worker 注册的函数，能在你 worker 观察到的事件（HTTP 请求、webhook 命中、文件变化、数据库更新）上被触发。
触发器类型的组成
一个触发器类型由两部分捆绑而成：
- 一个字符串
id：消费者绑定时引用它，例如type: "mini-http"。 - 一个按绑定维护的路由表：这个表由你的 worker 在进程内自行维护。引擎的注册表（registry）会以规范形式记录绑定（这正是
engine::triggers::list返回的内容），但引擎并不会基于它做分发。引擎只负责把网络上任何消费者 worker 的绑定/解绑事件作为回调转发给你的发布者 worker，具体如何处置每个绑定由你的 worker 决定。
在启动时用worker.registerTriggerType({ id, description }, handler)声明一次触发器类型。
你需要实现的TriggerHandler接口暴露两个回调，每当有消费者绑定或解绑时引擎会在你的发布者 worker 上调用它们：
registerTrigger(config)：任何消费者 worker 把函数绑定到你的类型时触发。config携带触发器实例的id、消费者的function_id，以及符合你类型所接受形状的消费者config。把它存起来。unregisterTrigger(config)：解绑时触发，从你的表里删除它。
触发器类型可以在运行期任意时刻被拆除，调用worker.unregisterTriggerType(...)（Python 与 Rust 中为worker.unregister_trigger_type(...)），签名见下文"注销触发器类型"一节。
示例：从零实现一个迷你iii-http
下面这个例子勾勒出真实http触发器类型发布者的精简版。发布者 worker 需要做三件事：
- 声明一个名为
mini-http、形状为 HTTP 的触发器类型； - 维护一张
bindings映射表（{ trigger id → function_id, method+path }），随着消费者绑定/解绑而增删； - 之后在收到 HTTP 请求时查询正确的绑定并触发。触发绑定函数的细节见下文"向已绑定函数分发事件"。
Node / TypeScript
import { registerWorker } from "iii-sdk"; import type { TriggerConfig, TriggerHandler } from "iii-sdk"; const url = process.env.III_URL; if (!url) throw new Error("III_URL must be set"); const worker = registerWorker(url); type MiniHttpConfig = { api_path: string; // leading slash, e.g. "/orders" http_method?: "GET" | "POST" | "PUT" | "DELETE"; }; const bindings = new Map<string, TriggerConfig<MiniHttpConfig>>(); const httpHandler: TriggerHandler<MiniHttpConfig> = { async registerTrigger(config) { bindings.set(config.id, config); }, async unregisterTrigger(config) { bindings.delete(config.id); }, }; worker.registerTriggerType( { id: "mini-http", description: "Routes HTTP requests to bound functions" }, httpHandler, );Python
import os from iii import ( InitOptions, RegisterTriggerTypeInput, TriggerConfig, TriggerHandler, register_worker, ) worker = register_worker( os.environ.get("III_URL"), InitOptions(worker_name="mini-http-worker"), ) bindings: dict[str, TriggerConfig] = {} class HttpHandler(TriggerHandler): async def register_trigger(self, config: TriggerConfig) -> None: bindings[config.id] = config async def unregister_trigger(self, config: TriggerConfig) -> None: bindings.pop(config.id, None) worker.register_trigger_type( RegisterTriggerTypeInput( id="mini-http", description="Routes HTTP requests to bound functions", ), HttpHandler(), )Rust
use std::collections::HashMap; use std::sync::{Arc, Mutex}; use iii_sdk::{ InitOptions, RegisterTriggerType, TriggerConfig, TriggerHandler, register_worker, }; let url = std::env::var("III_URL").expect("III_URL must be set"); let worker = register_worker(&url, InitOptions::default()); #[derive(Default)] struct HttpHandler { bindings: Arc<Mutex<HashMap<String, TriggerConfig>>>, } #[async_trait::async_trait] impl TriggerHandler for HttpHandler { async fn register_trigger(&self, config: TriggerConfig) -> Result<(), iii_sdk::IIIError> { self.bindings.lock().unwrap().insert(config.id.clone(), config); Ok(()) } async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), iii_sdk::IIIError> { self.bindings.lock().unwrap().remove(&config.id); Ok(()) } } worker.register_trigger_type( RegisterTriggerType::new( "mini-http", "Routes HTTP requests to bound functions", HttpHandler::default(), ), );从 SDK 源码可以看到TriggerConfig的真实字段：除了文档中提到的id、function_id、config、metadata之外，还包含一个namespace字段——当注册时省略命名空间，SDK 会用注册 worker 自身的命名空间补全；发布者若存储了该 config 并在之后调用trigger()，必须把这个已解析的命名空间透传下去（见 sdk/packages/node/iii/src/triggers.ts、sdk/packages/python/iii/src/iii/triggers.py、sdk/packages/rust/iii/src/triggers.rs）。
为触发器类型附加 Schema
一个触发器类型可以携带两个可选的 JSON Schema，用于描述它的载荷：
trigger_request_format：消费者在worker.registerTrigger(...)绑定函数时传入的按绑定config的 schema。call_request_format：触发器触发时你的 worker 交付给绑定函数的调用载荷的 schema。
两者都会输入 iii 控制台、Agent 可读的 skills，以及engine::trigger-types::list的输出，让消费者知道该传什么、会收到什么。
注意：运行时校验目前尚未支持。附加的 schema 仅是信息性的——引擎不会拒绝不符合它们的
config值或调用载荷。请把 schema 当作面向消费者、Agent 和控制台的契约文档，这与函数请求/响应 schema 的注意事项一致。
各 SDK 以自己惯用的方式接收这两个 schema：
| SDK | 传入方式 |
|---|---|
| Node / Browser | 在trigger_request_format/call_request_format上传原始 JSON Schema 对象；Zod 4+ schema 可用z.toJSONSchema(...)转换。 |
| Python | 在RegisterTriggerTypeInput的相同字段上传 Pydantic 模型类（自动转换）或原始 dict。 |
| Rust | RegisterTriggerType上的构造器方法：.trigger_request_format::<T>()与.call_request_format::<T>()，其中T: schemars::JsonSchema。 |
以引擎内置的http类型为例，其真实 schema 定义在 engine/src/trigger_formats.rs：HttpTriggerConfig包含api_path（如/users/:id）、可选的http_method（缺省为 GET）以及可选的condition_function_id；HttpCallRequest则包含query_params、路径参数等字段。Rust SDK 侧，RegisterTriggerType构造器会把这些 schema 序列化进注册消息（见 sdk/packages/rust/iii/src/iii.rs）。
注销一个触发器类型
当触发器类型所路由的工作不再需要时，可以在运行期拆除它。当发布者 worker 断线时，它宣告的所有触发器类型都会被自动移除，引擎也会停止路由依赖它们的事件——所以显式注销只在"worker 保持连接但想丢弃某个类型"时才必要。
可以在registerTriggerType之后的任意时刻调用，前提是 worker 保持连接。典型场景包括：底层资源进入维护模式、功能开关关闭了该对外面、或想在不重启的情况下把类型轮换到新 schema。沿用mini-http的例子，这里 worker 因其 HTTP 监听器被配置关闭而丢弃mini-http：
Node / TypeScript
// e.g. config reload disabled the HTTP listener; stop accepting new bindings // while the worker keeps serving other trigger types. worker.unregisterTriggerType({ id: "mini-http", description: "Routes HTTP requests to bound functions", });Python
# e.g. config reload disabled the HTTP listener; stop accepting new bindings # while the worker keeps serving other trigger types. worker.unregister_trigger_type( {"id": "mini-http", "description": "Routes HTTP requests to bound functions"} )Rust
// e.g. config reload disabled the HTTP listener; stop accepting new bindings // while the worker keeps serving other trigger types. worker.unregister_trigger_type("mini-http");三种 SDK 在此处存在签名差异（见 sdk/packages/node/iii/src/types.ts、sdk/packages/python/iii/src/iii/iii.py）：
- Node 的
registerTriggerType会返回一个TriggerTypeRef，带有.unregister()快捷方式，内部委托给worker.unregisterTriggerType(...)； - Python 的
TriggerTypeRef只暴露register_trigger和register_function，拆除类型本身要走worker.unregister_trigger_type(...)； - Rust 只接收
id字符串；Node 与 Python 接收完整输入对象，但实际只使用其中的id字段来定位被拆除的类型。
向已绑定函数分发事件
iii没有专门的"触发"API。当底层事件源送来内容（一个 HTTP 请求、一次 cron 滴答、一次 webhook 命中）时，你的发布者 worker 在registerTrigger回调构建的bindings表中查找对应条目，然后通过worker.trigger(...)调用每个匹配的函数。沿用上面的mini-http例子：
Node / TypeScript
// Inside the worker's HTTP listener, after matching method+path to an // entry in the `bindings` map from the declare-trigger-type example: const binding = bindings.get(matchedTriggerId); await worker.trigger({ function_id: binding.function_id, payload: { method, headers, body }, });Python
# Inside the worker's HTTP listener, after matching method+path to an # entry in the `bindings` dict from the declare-trigger-type example: binding = bindings[matched_trigger_id] worker.trigger({ "function_id": binding.function_id, "payload": {"method": method, "headers": headers, "body": body}, })Rust
use iii_sdk::TriggerRequest; use serde_json::json; // Inside the worker's HTTP listener, after matching method+path to an // entry in the handler's `bindings` map from the declare-trigger-type example: let binding = handler.bindings.lock().unwrap().get(&matched_trigger_id).cloned(); if let Some(binding) = binding { worker .trigger(TriggerRequest { function_id: binding.function_id.clone(), payload: json!({ "method": method, "headers": headers, "body": body }), action: None, timeout_ms: None, }) .await?; }每一次分发事件时，引擎都会评估消费者的config与可选的condition_function_id，然后把匹配的调用路由到绑定函数，并把结果返回给调用方。
源码佐证：引擎侧的注册与分发机制
文档描述的"引擎转发绑定/解绑回调、但不负责分发"这一设计，在引擎源码中有清晰的印证。TriggerRegistry（engine/src/trigger.rs）维护三类数据结构：
trigger_types：按(namespace, type id)键控的提供者（provider）表；triggers：当前存活的绑定；pending_triggers：挂起意图——当触发器类型尚未注册、提供者 worker 断线、绑定未能送达或提供者异步拒绝激活时，绑定会被"停车"（park）在这里；它们被禁用（不会触发），直到触发器类型（重新）注册时被激活并移入triggers。
这正是"发布者未连接时注册也能成功"的底层实现：register_trigger（engine/src/trigger.rs）在找不到提供者时不会失败，而是打印[PENDING]警告并把意图插入pending_triggers，同时在类型注册的并发窗口内做一次 re-check 以关闭"停车/排空"竞争。register_trigger_type（engine/src/trigger.rs）则负责先发布类型、再重放（replay）匹配的存活绑定、排空挂起意图，并处理"回迁"（re-homing）——把曾回退到default命名空间的绑定迁移回新注册的自家提供者。unregister_worker（engine/src/trigger.rs）则展示了断线清理：提供者离开后其绑定变成孤儿，会重新解析提供者、尽力回退，实在无处解析的绑定被[DISABLED]并停车，等待类型回归时自动恢复。
引擎对已知触发器类型的提供者做了内置映射（KNOWN_TRIGGER_TYPE_PROVIDERS，见 engine/src/trigger.rs），http、cron、subscribe、state、durable:subscriber、stream、log、trace、configuration等均在其中；当挂起警告持久存在时，日志会提示缺失的 worker 并给出iii trigger -n <namespace> compose::add worker=<worker>的安装建议。
小结
iii 的触发器体系把"事件路由"的职责完全交给了发布者 worker：引擎只负责记录绑定并转发回调，如何查表、如何分发由你决定。实际开发中，你通常只需用worker.registerTrigger消费既有类型；当需要把自有的 HTTP、webhook、文件或数据库事件暴露给其他 worker 时，则用registerTriggerType+TriggerHandler声明类型、用worker.trigger分发事件，并用trigger_request_format/call_request_format把契约写清楚。记住几个关键边界：metadata 是消费者按绑定打的标签，schema 是发布者按类型声明的契约且暂不参与运行时校验，而触发器类型的生命周期（断线自动清理、挂起重放、回迁）由引擎的TriggerRegistry托管，无需你手动兜底。
【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
