iii 函数编写实战指南注册、JSON Schema 契约、HTTP 远端调用与错误处理【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii在 iii 系统中Worker 通过注册函数Functions对外贡献能力。函数以service::name形式的id唯一标识由接收 payload 并返回结果的 Handler 构成还可附带描述请求与响应结构的 JSON Schema。本指南以 iii 0.18 文档体系中的 创建 Worker / Functions 为核心结合仓库内 Node / Python / Rust 三套 SDK 的实现与测试完整讲解函数注册、Schema 契约声明、把外部 HTTP 端点注册为函数、返回值与错误传播、运行时注销等全流程。读完本文你将能够为 iii 系统编写可被worker.trigger、iii trigger以及各类事件触发器调用的高质量函数并理解底层消息协议与引擎行为。什么是编写一个函数在 iii 中Worker 是能力的载体而函数是能力的出口。一个函数由三部分组成id形如service::name例如math::add、notifications::send它是触发器的function_id也是引擎路由调用的依据Handler接收调用方传入的 payload返回一个结果值可选的 JSON Schema描述请求request_format与响应response_format的形态。函数由 Worker 通过 SDK 注册到引擎。调用方其他 Worker、CLI、事件触发器不需要关心函数在哪台机器、由哪个 Worker 提供——引擎负责把function_id路由到注册了该函数的 Worker。注意每种触发器对函数的入参结构有各自的约定。例如cron触发器调用函数时不传参数而http触发器会提供包含body、headers等属性的标准 HTTP 风格 payload。具体每个 Worker 的 payload 形态请参见对应 Worker 的文档页。调用方如何触发函数worker.trigger/iii trigger/ 事件绑定触发器属于使用 iii / Triggers的范畴详见 使用 iii / Triggers本文聚焦于函数的编写与注册。注册一个函数在 Worker 代码中通过 SDK 的注册 API 把函数注册给引擎。函数id就是后续触发器使用的function_id。Node / TypeScriptimport { registerWorker } from iii-sdk; const url process.env.III_URL; if (!url) throw new Error(III_URL must be set); const worker registerWorker(url); worker.registerFunction(math::add, async (payload: { a: number; b: number }) { return { c: payload.a payload.b }; });Pythonimport os from iii import register_worker, InitOptions worker register_worker( os.environ.get(III_URL), InitOptions(worker_namemath-worker), ) def add_handler(payload: dict) - dict: return {c: payload[a] payload[b]} worker.register_function(math::add, add_handler)Rustuse iii_sdk::{InitOptions, RegisterFunction, register_worker}; let url std::env::var(III_URL).expect(III_URL must be set); let worker register_worker(url, InitOptions::default()); worker.register_function(RegisterFunction::new(math::add, |input: AddInput| { Ok(serde_json::json!({ c: input.a input.b })) }));注册背后的实现细节从 SDK 源码可以看到注册并非简单的存个回调而是一次完整的消息协议交互Node SDKsdk/packages/node/iii/src/iii.ts的registerFunction会先校验functionId非空为空抛id is required并检查本地functions表中是否已存在同名 id重复注册抛function id already registered随后通过 WebSocket 发送RegisterFunction消息给引擎。Python SDKsdk/packages/python/iii/src/iii/iii.py的register_function同样做类型与重复校验非字符串抛TypeError、空串或重复抛ValueError。同步 Handler 会被自动包装到独立线程执行run_in_executor的等价实现避免阻塞事件循环异步 Handler 则直接 await。Rust SDK中函数注册消息RegisterFunctionMessagesdk/packages/rust/iii/src/protocol.rs包含id、description、request_format、response_format、metadata以及可选的invocationHTTP 远端调用配置字段通过with_id/with_description构建器构造后随协议消息发送。另外Node SDK 还会为每个本地 Handler 包一层跟踪tracing逻辑调用 Handler 前后记录iii.invocation.input/iii.invocation.outputspan 事件并对 payload 做脱敏与截断受III_DISABLE_TRACE_PAYLOADS环境变量与字节上限控制Handler 的执行 span 命名为execute {functionId}——这意味着注册函数后其调用链自动获得可观测性。为函数附加请求与响应 Schema在注册时附加 JSON Schema可以让请求/响应结构随函数一起被文档化。Schema 随函数存储在引擎中并会呈现在 iii Console 与 Agent 可读的 skill 文档中。重要当前尚不支持运行时校验。附加的 Schema 只是元数据contract documentation引擎不会强制执行某个 Schema不会拒绝不匹配的 payload也不会拒绝不符合 Schema 的 Handler 返回值。请将 Schema 视为面向函数调用方、Agent 与 Console 的契约文档而非运行时防线。Node / TypeScriptimport { registerWorker } from iii-sdk; const url process.env.III_URL; if (!url) throw new Error(III_URL must be set); const worker registerWorker(url); worker.registerFunction( math::add, async (payload) ({ c: payload.a payload.b }), { request_format: { type: object, properties: { a: { type: number }, b: { type: number } }, required: [a, b], }, response_format: { type: object, properties: { c: { type: number } }, required: [c], }, }, );Pythonimport os from iii import register_worker, InitOptions worker register_worker( os.environ.get(III_URL), InitOptions(worker_namemath-worker), ) worker.register_function( math::add, add_handler, request_format{ type: object, properties: {a: {type: number}, b: {type: number}}, required: [a, b], }, response_format{ type: object, properties: {c: {type: number}}, required: [c], }, )Rustuse iii_sdk::{InitOptions, RegisterFunction, register_worker}; use schemars::JsonSchema; use serde::Deserialize; #[derive(Deserialize, JsonSchema)] struct AddInput { a: f64, b: f64 } #[derive(serde::Serialize, JsonSchema)] struct AddOutput { c: f64 } let url std::env::var(III_URL).expect(III_URL must be set); let worker register_worker(url, InitOptions::default()); // Rust 通过 closure 的输入/输出类型结合 schemars::JsonSchema 自动派生 // request_format 与 response_format无需任何 builder 方法——为请求/响应 // 结构体标注派生宏SDK 即自动生成 Schema。 worker.register_function(RegisterFunction::new( math::add, |input: AddInput| - ResultAddOutput, String { Ok(AddOutput { c: input.a input.b }) }, ));这些 Schema 同时会喂给 iii Console 与 Agent 可读的 skill 文档方便人类与 Agent 发现函数的调用契约。各语言 SDK 的 Schema 生成策略差异Python SDK更自动register_function的request_format/response_format默认为None此时 SDK 会从 Handler 的类型注解如 PydanticBaseModel自动抽取 Schema传入显式 Schema 则可覆盖自动抽取。仓库示例 sdk/packages/python/iii-example/src/iii_function_example.py 演示了用 Pydantic 模型做入参/返回值并直接注册的写法。这是 Python 特有行为——Node SDK 依赖显式 Schema因为 TypeScript 类型在运行时已被擦除。Rust SDK借助schemars::JsonSchema派生宏从结构体自动生成 Schema。Node SDK需要显式传入request_format/response_format对象。HTTP 可调用函数HTTP-invokable functions除了本地 Handler你还可以把一个外部 HTTP 端点注册为函数函数被调用时由引擎发起 HTTP 请求Worker 只需声明端点即可。这一能力非常适合把已有基础设施包装成 iii 函数使用例如现有的 API Gateway 与 WebhookServerless 平台Lambda、Azure Functions、Google Cloud Functions上已有的处理逻辑任何希望以 iii 函数形式对外暴露的第三方 API。注册后该函数与普通函数完全等价worker.trigger、iii trigger以及任意绑定的触发器类型queue、cron、state、http都可以直接调用它无需额外改动。示例把外部 Webhook 注册为notifications::sendNode / TypeScriptimport { registerWorker } from iii-sdk; const url process.env.III_URL; if (!url) throw new Error(III_URL must be set); const worker registerWorker(url); worker.registerFunction( notifications::send, { url: https://hooks.provider.example.com/notify, method: POST, timeout_ms: 5000, headers: { X-Service: iii-worker }, auth: { type: bearer, token_key: PROVIDER_API_TOKEN }, }, { description: POST a notification to the provider webhook }, );Pythonimport os from iii import HttpInvocationConfig, InitOptions, register_worker from iii.iii_types import HttpAuthBearer worker register_worker( os.environ.get(III_URL), InitOptions(worker_namenotifications-worker), ) worker.register_function( notifications::send, HttpInvocationConfig( urlhttps://hooks.provider.example.com/notify, methodPOST, timeout_ms5000, headers{X-Service: iii-worker}, authHttpAuthBearer(token_keyPROVIDER_API_TOKEN), ), descriptionPOST a notification to the provider webhook, )Rustuse std::collections::HashMap; use iii_sdk::{ HttpAuthConfig, HttpInvocationConfig, HttpMethod, InitOptions, RegisterFunctionMessage, register_worker, }; let url std::env::var(III_URL).expect(III_URL must be set); let worker register_worker(url, InitOptions::default()); let mut headers HashMap::new(); headers.insert(X-Service.into(), iii-worker.into()); worker.register_function(( RegisterFunctionMessage::with_id(notifications::send.into()) .with_description(POST a notification to the provider webhook.into()), HttpInvocationConfig { url: https://hooks.provider.example.com/notify.into(), method: HttpMethod::Post, timeout_ms: Some(5000), headers, auth: Some(HttpAuthConfig::Bearer { token_key: PROVIDER_API_TOKEN.into(), }), }, ));普通函数接收id Handler而 HTTP 可调用函数接收idHttpInvocationConfig。HttpInvocationConfig字段说明字段类型默认值说明urlstring必填函数被调用时引擎请求的端点。methodGET \| POST \| PUT \| PATCH \| DELETEPOSTHTTP 方法。timeout_msnumber30000单次请求超时时间毫秒。headersRecordstring, string{}每次调用都会附加的请求头。authHttpAuthConfig无认证配置对象bearer、hmac或api_key配合token_key、secret_key或value_key使用。三套 SDK 的类型定义与该表完全一致Node 见 sdk/packages/node/helpers/src/http/index.tsPython 见 sdk/packages/python/helpers/src/iii_helpers/http/init.pyRust 见 sdk/packages/rust/helpers/src/http.rs。其中HttpMethod仅包含 GET/POST/PUT/PATCH/DELETE 五种与内置 HTTP 触发器的完整方法枚举不同后者还覆盖 HEAD/OPTIONSHttpAuthConfig有三种形态hmac共享密钥做 HMAC 签名校验字段secret_key、bearerBearer Token字段token_key、api_key自定义头携带 API Key字段headervalue_key。安全要点token_key、secret_key、value_key这三个字段指定的都是环境变量的名字而不是密钥本身。引擎在注册时从自身的进程环境中解析这些变量因此密钥始终停留在引擎主机上永远不会通过 SDK 的 WebSocket 连接传输。HTTP 错误处理引擎会把调用 payload 作为 JSON 请求体发送给端点并将任何非 2xx 响应或网络错误视为调用失败错误会传播回调用方。HTTP 可调用函数与进程内 Handler 一样会出现在engine::functions::list中引擎侧该内置函数实现在 engine/src/workers/engine_fn/mod.rs支持按prefix/search/worker过滤并可通过 Console 被发现。仓库的集成测试对这套行为做了完整验证Node 侧 sdk/packages/node/iii/tests/http-external-functions.test.ts 覆盖了投递事件到外部 HTTP 函数注册与注销 HTTP 函数自定义请求头多外部函数注销后停止投递PUT 方法等场景Rust 侧 sdk/packages/rust/iii/tests/http_external_functions.rs 与 Python 侧 sdk/packages/python/iii/tests/test_http_external_functions_integration.py 提供了等价覆盖包括 Bearer 认证。返回值与错误处理函数要么返回一个值要么返回一个错误返回值由 Handler 负责将结果塑造成与其声明的 response schema 匹配的形态错误Handler 内部抛出的错误会作为调用错误传播给调用方并附带 Worker 侧的堆栈信息Node 转发error.stackPython 转发traceback.format_exc()Rust 转发底层错误的堆栈跟踪。引擎不会吞掉这些错误。建议利用这一区分来表达两类失败预期的失败expected failures——返回结构化的错误值意外的失败unexpected ones——直接 throw / raise / 返回Err让引擎把完整堆栈带给调用方用于诊断。注销一个函数registerFunction及对应语言的注册方法返回一个句柄handle其上带有unregister()方法可在运行时把函数从引擎中移除。此外当 Worker 断开连接时它的全部函数都会被自动移除尚未完成的调用pending invocations会报错。Node / TypeScriptconst add worker.registerFunction(math::add, async (payload) { return { c: payload.a payload.b }; }); add.unregister();Pythonadd worker.register_function(math::add, add_handler) add.unregister()Rustlet add worker.register_function(RegisterFunction::new(math::add, |input: AddInput| { Ok(serde_json::json!({ c: input.a input.b })) })); add.unregister();从实现上看unregister()会向引擎发送UnregisterFunction消息并从本地函数表中移除该 idNode 实现见 sdk/packages/node/iii/src/iii.tsPython 见 sdk/packages/python/iii/src/iii/iii.py。Rust 侧测试registers_and_unregisters_external_http_function与 Node 侧的stops delivering events after unregister都验证了注销后事件不再投递。小结函数是 iii 系统的能力单元id采用service::name命名通过 SDK 注册到引擎供worker.trigger、iii trigger与各类事件触发器按function_id路由调用request_format/response_format是契约文档型元数据当前不做运行时校验会同步到 Console 与 Agent 可读的 skill 中Python 可从类型注解自动抽取Rust 可经schemars::JsonSchema自动派生Node 需显式声明HttpInvocationConfig可以把任意 HTTP 端点注册为 iii 函数支持方法、超时、自定义头与三种认证bearer / hmac / api_key密钥以环境变量名形式交给引擎在注册时解析错误会连同 Worker 堆栈Nodeerror.stack、Pythontraceback.format_exc()、Rust 底层堆栈传播给调用方预期失败请返回结构化错误值unregister()句柄支持运行时注销Worker 断开时函数自动清理。如需深入了解调用方视角如何触发函数、TriggerAction、队列/定时/状态/HTTP 触发器的绑定方式请继续阅读 使用 iii / Triggers 与 使用 iii / Functions。【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考