尧图网站建设 
iii-sdk Python 快速上手用 iii 引擎注册函数、绑定触发器与调用工作流【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii导读本文以 sdk/packages/python/iii/README.md 为主体系统讲解 iii-sdkPyPI 包名iii-sdk的安装、Worker 初始化、函数注册、触发器绑定与函数调用全流程。iii-sdk 是 iii 引擎的 Python 客户端Worker 通过 WebSocket 接入引擎注册可在引擎侧按名称调用的函数并将 HTTP、cron、队列等触发器绑定到这些函数上。读完本文你将掌握用 Python 编写可被引擎调度、可被其他 Worker 调用、可接入实时触发器的完整 Worker 的最小实现以及同步/异步 API、命名空间、连接管理与开发测试配套工具链。安装与项目环境iii-sdk 是一个发布在 PyPI 上的标准 Python 包通过 pip 即可安装pip install iii-sdk从 pyproject.toml 可以看到包的工程约束要求 Python3.10核心依赖为websockets12.0WebSocket 通信、pydantic2.0消息与配置模型、opentelemetry-api1.25可观测性以及同版本的iii-helpers公共辅助类型如 HTTP 调用配置与 OTel 配置。包名与许可证为 Apache-2.0源码位于 src/iii 目录顶层导出集中在init.py。引擎地址的解析规则见 iii.py 中的resolve_engine_url显式传入的 address 参数优先其次是环境变量III_URL最后回退到内置默认值ws://127.0.0.1:49134见 iii_constants.py刻意使用 IPv4 回环地址以避免localhost在部分主机上解析为::1的问题。因此本地开发时register_worker()可以不传任何参数直接连接默认引擎。Hello World完整的最小 WorkerREADME 给出的 Hello World 是理解 SDK 工作流的最佳入口from iii import register_worker iii register_worker(ws://localhost:49134) def greet(data): return {message: fHello, {data[name]}!} iii.register_function(hello::greet, greet) iii.register_trigger({ type: http, function_id: hello::greet, config: {api_path: /greet, http_method: POST}, }) iii.connect() result iii.trigger({function_id: hello::greet, payload: {name: world}}) print(result) # {message: Hello, world!}这段代码体现了 SDK 的四个核心步骤初始化连接register_worker(url)创建 SDK 实例并自动连接引擎注册函数iii.register_function(id, handler)注册一个可被按名称调用的函数绑定触发器iii.register_trigger({...})把 HTTP、cron、队列等触发器绑定到函数上调用与关闭iii.trigger(...)同步等待调用结果iii.shutdown()优雅断开连接。从源码看register_workeriii.py会创建III客户端实例并阻塞等待连接建立最多 30 秒若超时未连上只记录警告并返回客户端连接会在后台持续重试注册的消息会先进入发送队列、连接成功后再统一冲刷见_on_connected与_queue逻辑。III构造时启动一个非守护后台线程运行独立事件循环测试 test_sync_api.py 专门验证了这一行为。API 总览README 用一张表概括了 SDK 的核心操作下表在此基础上补充了对应源码位置与关键行为说明操作签名说明初始化register_worker(url, options?)创建 SDK 实例并自动连接引擎address 可省略依次从III_URL、默认地址解析注册函数iii.register_function(id, handler)注册可被按名称调用的函数返回带unregister()的FunctionRef注册触发器iii.register_trigger({type: ..., function_id: ..., config: ...})将 HTTP、cron、队列等触发器绑定到函数返回带unregister()的Trigger同步调用等待结果iii.trigger({function_id: id, payload: data})发送调用并等待函数返回结果异步调用fire-and-forgetiii.trigger({..., action: TriggerAction.Void()})只发送不等待响应返回None队列路由调用iii.trigger({..., action: TriggerAction.Enqueue(queuename)})通过命名队列路由调用返回包含messageReceiptId的字典关闭iii.shutdown()断开连接并停止后台线程此外register_worker()的options参数接受InitOptions定义于 iii_constants.py常用字段包括worker_nameWorker 显示名III_WORKER_NAME环境变量可覆盖缺省为hostname:pid、worker_description一行人类/LLM 可读的职责描述会出现在engine::workers::list等引擎内建函数中、namespaceWorker 所属命名空间回退到III_NAMESPACE环境变量再缺省由引擎使用default、invocation_timeout_ms调用超时默认 30000、reconnection_config重连策略与otelOpenTelemetry 配置。注册函数同步、异步与 HTTP 转发README 的示例展示了返回 HTTP 风格响应的函数注册方式def create_order(data): return {status_code: 201, body: {id: 123, item: data[body][item]}} iii.register_function(orders::create, create_order)从 register_function 的实现 可以看到几处值得注意的行为同步与异步处理器均可协程处理器asyncio.iscoroutinefunction直接await同步处理器会被包装到独立线程中执行run_in_executor的等价实现避免阻塞 SDK 的事件循环按需透传调用元数据只有处理器显式声明名为metadata的参数时SDK 才会把每次调用的元数据传给它支持def handler(data, metadataNone)或def handler(data, *, metadataNone)两种签名见_metadata_passing_modeiii.py既有的一参处理器无需改动请求/响应格式自动提取省略request_format/response_format时SDK 会从处理器的类型注解自动提取 schemaPydantic 模型会被转换为 JSON SchemaNode SDK 因 TypeScript 类型在运行时被擦除而必须显式传 schema这是 Python SDK 的独有便利HTTP 外部函数除了本地可调用处理器还可以传入iii_helpers.http.HttpInvocationConfig把函数注册为 HTTP 调用的远端函数如 Lambda、Cloudflare Workers引擎会通过 HTTP 转发调用重复注册校验function_id为空或已注册会抛出ValueError非字符串会抛出TypeError返回句柄可注销register_function返回的FunctionRef带有unregister()可编程移除函数对应引擎侧UnregisterFunctionMessage消息。注册触发器把事件源接到函数上README 展示了 HTTP 触发器的注册方式iii.register_trigger({ type: http, function_id: orders::create, config: {api_path: /orders, http_method: POST}, })触发器描述符由三要素构成type触发器类型如http、cron、queue等、function_id触发时调用的目标函数、config类型相关的配置HTTP 类型下为api_path与http_method。源码实现iii.py中SDK 会为每个触发器自动生成 UUID 作为触发器 ID发送RegisterTriggerMessage给引擎并返回带unregister()方法的Trigger对象。值得注意的是命名空间语义触发器默认解析到当前 Worker 的命名空间而非引擎的default因为触发器指向的函数注册在 Worker 自己的命名空间里默认到别处会导致触发器触发了却解析不到函数的问题需要明确指向其他命名空间时可通过RegisterTriggerInput.namespace显式声明。自定义触发器类型除了内置类型SDK 还支持注册自定义触发器类型。通过register_trigger_type传入类型定义id、description、可选的trigger_request_format/call_request_formatPydantic 模型与实现了TriggerHandler抽象基类的处理器需实现register_trigger/unregister_trigger两个抽象方法见 triggers.py即可获得带类型约束的TriggerTypeRef句柄webhook worker.register_trigger_type( RegisterTriggerTypeInput( idwebhook, descriptionWebhook trigger, trigger_request_formatWebhookConfig, call_request_formatWebhookCallRequest, ), WebhookHandler(), ) webhook.register_function(handler, handle_webhook) webhook.register_trigger(handler, WebhookConfig(url/hook))调用函数同步、Void 与队列三种路由方式README 中同步调用的写法为result iii.trigger({function_id: orders::create, payload: {body: {item: widget}}})trigger的完整行为取决于请求中的action字段源码见 iii.py不带 action同步调用SDK 生成invocation_id并等待引擎返回结果支持通过timeout_ms缺省取InitOptions.invocation_timeout_ms默认 30000ms控制超时超时抛出InvocationError(codeTIMEOUT)TriggerAction.Void()fire-and-forget只发送不等待立即返回NoneTriggerAction.Enqueue(queuename)通过命名队列路由调用返回包含messageReceiptId的字典队列需在队列 Worker 的queue_configs中预先声明。worker.trigger({function_id: process, payload: {}, action: TriggerAction.Enqueue(queuejobs)}) worker.trigger({function_id: notify, payload: {}, action: TriggerAction.Void()})调用失败时抛出InvocationError可通过其code字段区分原因TIMEOUT表示超时FORBIDDEN表示 RBAC 拒绝。所有同步方法均有对应的trigger_async异步版本III内部通过asyncio.run_coroutine_threadsafe把同步调用桥接到后台事件循环iii.py因此既能在普通脚本中同步调用也能在 asyncio 程序中直接await worker.trigger_async(...)配套测试见 test_async_api.py。连接生命周期重连、状态监听与命名空间iii-sdk 的连接管理并非一次连接、断开即死的简单模型其关键机制包括自动重连连接失败或异常断开后SDK 按ReconnectionConfiginitial_delay_ms、backoff_multiplier、max_delay_ms、jitter_factor、max_retriesmax_retries-1表示无限重试执行带指数退避与抖动的重连循环_reconnect_loopiii.py断线重注册重连成功后_on_connected会把此前注册的触发器类型、函数、触发器全部重放给引擎并冲刷连接期间积压在队列中的消息保证 Worker 断线期间发起的调用不丢失Reattach 身份重续若已有worker_id重连时会先发送REATTACH消息携带上轮的reattach_token作为身份凭证让引擎退役旧连接、避免身份冲突状态机与监听连接状态为disconnected/connecting/connected/reconnecting/failed五态iii_constants.py通过add_connection_state_listener订阅状态迁移返回幂等退订函数get_connection_state()随时可查致命拒绝不再重连若引擎以REGISTRATION_REJECTED拒绝注册如WORKER_NAMESPACE_CONFLICT表示同名 Worker 已在存活SDK 记录致命错误、立即失败所有在途调用并停止重连而FUNCTION_NAMESPACE_CONFLICT单个函数 ID 被他人占用只跳过该函数、不影响 Worker 其余服务命名空间解析namespace的优先级为InitOptions.namespace→III_NAMESPACE环境变量 → 引擎default声明为空白字符串会被视为错误抛出因为未设置与设为空语义相反见 iii.py。这些环境变量III_URL、III_NAMESPACE、III_WORKER_NAME通常由 iii 的编排器iii compose、容器运行时或 systemd注入让同一份 Worker 代码在不同部署环境下无需改动即可接入不同引擎。开发、类型检查与测试README 给出包自身的开发工作流适用于对 SDK 本身做二次开发或本地调试的场景# 开发模式安装 pip install -e . # 类型检查 mypy src # Lint ruff check src工程在 pyproject.toml 中配置了mypystrict 模式与ruff选择 E/F/I/W 规则组行宽 120测试采用 pytest 且默认开启分支覆盖率统计--covsrc/iii --cov-branch。仓库 tests 目录下有超过 40 个测试文件覆盖同步/异步 API、触发器注册、重连如 test_reconnect_sends_reattach.py、命名空间、RBAC Worker、OpenTelemetry 遥测、流Streams、通道Channels与状态管理等功能是理解 SDK 行为契约的补充资料。总结iii-sdkPython以register_worker→register_function→register_trigger→trigger四步为核心工作流通过 WebSocket 把 Python Worker 接入 iii 引擎引擎负责函数路由、触发器分发与跨 Worker 编排SDK 负责连接管理、消息协议与同步/异步适配。若要进一步深入可阅读 docs/quickstart.mdx 了解引擎安装与启动方式参考 sdk/packages/python/iii/README.md 同目录下的源码与测试或对照 Node、Rust、Go 等其他语言的 SDK 实现理解跨语言一致性设计。【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 iii-sdk Python 快速上手：用 iii 引擎注册函数、绑定触发器与调用工作流
iii-sdk Python 快速上手用 iii 引擎注册函数、绑定触发器与调用工作流【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii导读本文以 sdk/packages/python/iii/README.md 为主体系统讲解 iii-sdkPyPI 包名iii-sdk的安装、Worker 初始化、函数注册、触发器绑定与函数调用全流程。iii-sdk 是 iii 引擎的 Python 客户端Worker 通过 WebSocket 接入引擎注册可在引擎侧按名称调用的函数并将 HTTP、cron、队列等触发器绑定到这些函数上。读完本文你将掌握用 Python 编写可被引擎调度、可被其他 Worker 调用、可接入实时触发器的完整 Worker 的最小实现以及同步/异步 API、命名空间、连接管理与开发测试配套工具链。安装与项目环境iii-sdk 是一个发布在 PyPI 上的标准 Python 包通过 pip 即可安装pip install iii-sdk从 pyproject.toml 可以看到包的工程约束要求 Python3.10核心依赖为websockets12.0WebSocket 通信、pydantic2.0消息与配置模型、opentelemetry-api1.25可观测性以及同版本的iii-helpers公共辅助类型如 HTTP 调用配置与 OTel 配置。包名与许可证为 Apache-2.0源码位于 src/iii 目录顶层导出集中在init.py。引擎地址的解析规则见 iii.py 中的resolve_engine_url显式传入的 address 参数优先其次是环境变量III_URL最后回退到内置默认值ws://127.0.0.1:49134见 iii_constants.py刻意使用 IPv4 回环地址以避免localhost在部分主机上解析为::1的问题。因此本地开发时register_worker()可以不传任何参数直接连接默认引擎。Hello World完整的最小 WorkerREADME 给出的 Hello World 是理解 SDK 工作流的最佳入口from iii import register_worker iii register_worker(ws://localhost:49134) def greet(data): return {message: fHello, {data[name]}!} iii.register_function(hello::greet, greet) iii.register_trigger({ type: http, function_id: hello::greet, config: {api_path: /greet, http_method: POST}, }) iii.connect() result iii.trigger({function_id: hello::greet, payload: {name: world}}) print(result) # {message: Hello, world!}这段代码体现了 SDK 的四个核心步骤初始化连接register_worker(url)创建 SDK 实例并自动连接引擎注册函数iii.register_function(id, handler)注册一个可被按名称调用的函数绑定触发器iii.register_trigger({...})把 HTTP、cron、队列等触发器绑定到函数上调用与关闭iii.trigger(...)同步等待调用结果iii.shutdown()优雅断开连接。从源码看register_workeriii.py会创建III客户端实例并阻塞等待连接建立最多 30 秒若超时未连上只记录警告并返回客户端连接会在后台持续重试注册的消息会先进入发送队列、连接成功后再统一冲刷见_on_connected与_queue逻辑。III构造时启动一个非守护后台线程运行独立事件循环测试 test_sync_api.py 专门验证了这一行为。API 总览README 用一张表概括了 SDK 的核心操作下表在此基础上补充了对应源码位置与关键行为说明操作签名说明初始化register_worker(url, options?)创建 SDK 实例并自动连接引擎address 可省略依次从III_URL、默认地址解析注册函数iii.register_function(id, handler)注册可被按名称调用的函数返回带unregister()的FunctionRef注册触发器iii.register_trigger({type: ..., function_id: ..., config: ...})将 HTTP、cron、队列等触发器绑定到函数返回带unregister()的Trigger同步调用等待结果iii.trigger({function_id: id, payload: data})发送调用并等待函数返回结果异步调用fire-and-forgetiii.trigger({..., action: TriggerAction.Void()})只发送不等待响应返回None队列路由调用iii.trigger({..., action: TriggerAction.Enqueue(queuename)})通过命名队列路由调用返回包含messageReceiptId的字典关闭iii.shutdown()断开连接并停止后台线程此外register_worker()的options参数接受InitOptions定义于 iii_constants.py常用字段包括worker_nameWorker 显示名III_WORKER_NAME环境变量可覆盖缺省为hostname:pid、worker_description一行人类/LLM 可读的职责描述会出现在engine::workers::list等引擎内建函数中、namespaceWorker 所属命名空间回退到III_NAMESPACE环境变量再缺省由引擎使用default、invocation_timeout_ms调用超时默认 30000、reconnection_config重连策略与otelOpenTelemetry 配置。注册函数同步、异步与 HTTP 转发README 的示例展示了返回 HTTP 风格响应的函数注册方式def create_order(data): return {status_code: 201, body: {id: 123, item: data[body][item]}} iii.register_function(orders::create, create_order)从 register_function 的实现 可以看到几处值得注意的行为同步与异步处理器均可协程处理器asyncio.iscoroutinefunction直接await同步处理器会被包装到独立线程中执行run_in_executor的等价实现避免阻塞 SDK 的事件循环按需透传调用元数据只有处理器显式声明名为metadata的参数时SDK 才会把每次调用的元数据传给它支持def handler(data, metadataNone)或def handler(data, *, metadataNone)两种签名见_metadata_passing_modeiii.py既有的一参处理器无需改动请求/响应格式自动提取省略request_format/response_format时SDK 会从处理器的类型注解自动提取 schemaPydantic 模型会被转换为 JSON SchemaNode SDK 因 TypeScript 类型在运行时被擦除而必须显式传 schema这是 Python SDK 的独有便利HTTP 外部函数除了本地可调用处理器还可以传入iii_helpers.http.HttpInvocationConfig把函数注册为 HTTP 调用的远端函数如 Lambda、Cloudflare Workers引擎会通过 HTTP 转发调用重复注册校验function_id为空或已注册会抛出ValueError非字符串会抛出TypeError返回句柄可注销register_function返回的FunctionRef带有unregister()可编程移除函数对应引擎侧UnregisterFunctionMessage消息。注册触发器把事件源接到函数上README 展示了 HTTP 触发器的注册方式iii.register_trigger({ type: http, function_id: orders::create, config: {api_path: /orders, http_method: POST}, })触发器描述符由三要素构成type触发器类型如http、cron、queue等、function_id触发时调用的目标函数、config类型相关的配置HTTP 类型下为api_path与http_method。源码实现iii.py中SDK 会为每个触发器自动生成 UUID 作为触发器 ID发送RegisterTriggerMessage给引擎并返回带unregister()方法的Trigger对象。值得注意的是命名空间语义触发器默认解析到当前 Worker 的命名空间而非引擎的default因为触发器指向的函数注册在 Worker 自己的命名空间里默认到别处会导致触发器触发了却解析不到函数的问题需要明确指向其他命名空间时可通过RegisterTriggerInput.namespace显式声明。自定义触发器类型除了内置类型SDK 还支持注册自定义触发器类型。通过register_trigger_type传入类型定义id、description、可选的trigger_request_format/call_request_formatPydantic 模型与实现了TriggerHandler抽象基类的处理器需实现register_trigger/unregister_trigger两个抽象方法见 triggers.py即可获得带类型约束的TriggerTypeRef句柄webhook worker.register_trigger_type( RegisterTriggerTypeInput( idwebhook, descriptionWebhook trigger, trigger_request_formatWebhookConfig, call_request_formatWebhookCallRequest, ), WebhookHandler(), ) webhook.register_function(handler, handle_webhook) webhook.register_trigger(handler, WebhookConfig(url/hook))调用函数同步、Void 与队列三种路由方式README 中同步调用的写法为result iii.trigger({function_id: orders::create, payload: {body: {item: widget}}})trigger的完整行为取决于请求中的action字段源码见 iii.py不带 action同步调用SDK 生成invocation_id并等待引擎返回结果支持通过timeout_ms缺省取InitOptions.invocation_timeout_ms默认 30000ms控制超时超时抛出InvocationError(codeTIMEOUT)TriggerAction.Void()fire-and-forget只发送不等待立即返回NoneTriggerAction.Enqueue(queuename)通过命名队列路由调用返回包含messageReceiptId的字典队列需在队列 Worker 的queue_configs中预先声明。worker.trigger({function_id: process, payload: {}, action: TriggerAction.Enqueue(queuejobs)}) worker.trigger({function_id: notify, payload: {}, action: TriggerAction.Void()})调用失败时抛出InvocationError可通过其code字段区分原因TIMEOUT表示超时FORBIDDEN表示 RBAC 拒绝。所有同步方法均有对应的trigger_async异步版本III内部通过asyncio.run_coroutine_threadsafe把同步调用桥接到后台事件循环iii.py因此既能在普通脚本中同步调用也能在 asyncio 程序中直接await worker.trigger_async(...)配套测试见 test_async_api.py。连接生命周期重连、状态监听与命名空间iii-sdk 的连接管理并非一次连接、断开即死的简单模型其关键机制包括自动重连连接失败或异常断开后SDK 按ReconnectionConfiginitial_delay_ms、backoff_multiplier、max_delay_ms、jitter_factor、max_retriesmax_retries-1表示无限重试执行带指数退避与抖动的重连循环_reconnect_loopiii.py断线重注册重连成功后_on_connected会把此前注册的触发器类型、函数、触发器全部重放给引擎并冲刷连接期间积压在队列中的消息保证 Worker 断线期间发起的调用不丢失Reattach 身份重续若已有worker_id重连时会先发送REATTACH消息携带上轮的reattach_token作为身份凭证让引擎退役旧连接、避免身份冲突状态机与监听连接状态为disconnected/connecting/connected/reconnecting/failed五态iii_constants.py通过add_connection_state_listener订阅状态迁移返回幂等退订函数get_connection_state()随时可查致命拒绝不再重连若引擎以REGISTRATION_REJECTED拒绝注册如WORKER_NAMESPACE_CONFLICT表示同名 Worker 已在存活SDK 记录致命错误、立即失败所有在途调用并停止重连而FUNCTION_NAMESPACE_CONFLICT单个函数 ID 被他人占用只跳过该函数、不影响 Worker 其余服务命名空间解析namespace的优先级为InitOptions.namespace→III_NAMESPACE环境变量 → 引擎default声明为空白字符串会被视为错误抛出因为未设置与设为空语义相反见 iii.py。这些环境变量III_URL、III_NAMESPACE、III_WORKER_NAME通常由 iii 的编排器iii compose、容器运行时或 systemd注入让同一份 Worker 代码在不同部署环境下无需改动即可接入不同引擎。开发、类型检查与测试README 给出包自身的开发工作流适用于对 SDK 本身做二次开发或本地调试的场景# 开发模式安装 pip install -e . # 类型检查 mypy src # Lint ruff check src工程在 pyproject.toml 中配置了mypystrict 模式与ruff选择 E/F/I/W 规则组行宽 120测试采用 pytest 且默认开启分支覆盖率统计--covsrc/iii --cov-branch。仓库 tests 目录下有超过 40 个测试文件覆盖同步/异步 API、触发器注册、重连如 test_reconnect_sends_reattach.py、命名空间、RBAC Worker、OpenTelemetry 遥测、流Streams、通道Channels与状态管理等功能是理解 SDK 行为契约的补充资料。总结iii-sdkPython以register_worker→register_function→register_trigger→trigger四步为核心工作流通过 WebSocket 把 Python Worker 接入 iii 引擎引擎负责函数路由、触发器分发与跨 Worker 编排SDK 负责连接管理、消息协议与同步/异步适配。若要进一步深入可阅读 docs/quickstart.mdx 了解引擎安装与启动方式参考 sdk/packages/python/iii/README.md 同目录下的源码与测试或对照 Node、Rust、Go 等其他语言的 SDK 实现理解跨语言一致性设计。【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考