Hatchet Python 快速入门实战用 Python 编写并运行第一个后台任务工作流【免费下载链接】hatchet An orchestration engine for background tasks, AI agents, and durable workflows项目地址: https://gitcode.com/GitHub_Trending/ha/hatchetHatchet 是一个面向后台任务、AI Agent 与持久化工作流durable workflows的编排引擎。本指南以仓库内 examples/python/quickstart 示例项目为主线完整讲解从环境准备、工程初始化、Worker 启动到任务触发执行的端到端流程并结合 Python SDK 源码剖析Hatchet()客户端初始化、hatchet.task任务声明、worker启动等底层机制。读完本文你将能够独立搭建一个最小可用的 Hatchet Python 工程并理解任务输入输出校验、Worker 槽位slots与 Token/TLS 配置背后的原理。前置条件在运行示例项目之前需要准备以下两项基础环境Python v3.10 或更高版本SDK 与示例工程的类型注解、语法特性依赖 3.10。示例工程 pyproject.toml 中声明了python ^3.10的版本约束。Poetry用于依赖管理与虚拟环境隔离。示例工程采用 Poetry 作为包管理器其pyproject.toml通过[tool.poetry.dependencies]声明了hatchet-sdk 1.33.6与python-dotenv 1.2.2两个核心依赖其中python-dotenv用于从.env文件加载环境变量。了解示例项目结构先用cd examples/python/quickstart进入示例目录其结构如下quickstart/ ├── README.md # 官方快速入门说明 ├── hatchet_client.py # 初始化 Hatchet 客户端全局单例 ├── worker.py # 定义并启动 Worker ├── run.py # 触发任务执行的入口 ├── pyproject.toml # Poetry 工程配置 ├── poetry.lock # 依赖锁定文件 └── workflows/ ├── __init__.py └── first_task.py # 第一个任务的声明与实现该工程刻意保持极简一个任务声明文件、一个 Worker 启动文件、一个任务触发文件覆盖了「声明任务 → 注册到 Worker → 触发执行」的完整链路非常适合作为上手模板。第一步创建项目并安装依赖1.1 使用 Hatchet CLI 脚手架创建工程官方推荐通过 Hatchet CLI 生成该示例工程hatchet quickstart该命令会拉取当前仓库examples/python/quickstart同源的模板工程Python、TypeScript、Go、Ruby 等语言均有对应版本。若你已经在本地克隆了本仓库也可以直接复用examples/python/quickstart目录下的现成代码。1.2 配置环境变量创建工程后需要设置HATCHET_CLIENT_TOKEN环境变量其值来自 Hatchet 控制台Dashboard中创建的 API Tokenexport HATCHET_CLIENT_TOKENtoken自托管提示如果你是自托管部署且未启用 TLS需要额外设置HATCHET_CLIENT_TLS_STRATEGYnone来关闭 TLSexport HATCHET_CLIENT_TLS_STRATEGYnone从 SDK 配置源码 可以看到ClientTLSConfig以HATCHET_CLIENT_TLS_为环境变量前缀strategy字段默认值为tls自托管 HTTP 场景必须显式覆盖为none。同时 SDK 会在初始化时通过 token 解析逻辑 校验 Token 是否已设置并从 JWT 中解析出tenant_id与 gRPC 广播地址host_portget_tenant_id_from_jwt/grpc_broadcast_address_from_jwt这意味着只要提供合法的HATCHET_CLIENT_TOKENSDK 即可自动推导出需要连接的引擎地址无需手工配置主机端口。另外SDK 的配置加载基于 pydantic-settings支持从.env、.env.hatchet、.env.dev、.env.local等环境文件读取配置见 config.py因此也可以将 Token 写入项目根目录的.env文件配合python-dotenv自动加载。1.3 安装依赖poetry install该命令会根据pyproject.toml与poetry.lock创建虚拟环境并安装hatchet-sdk、python-dotenv等全部依赖。工程还在[tool.poetry.scripts]中注册了两个快捷命令[tool.poetry.scripts] simple src.run:main worker src.worker:main安装完成后可直接通过poetry run simple与poetry run worker调用等价于执行src/run.py与src/worker.py的main函数注意脚本入口路径中的src目录即示例工程代码所在包。第二步理解客户端初始化与任务声明2.1 初始化 Hatchet 客户端hatchet_client.py 是整个工程的基石仅有两行from hatchet_sdk import Hatchet # Initialize Hatchet client hatchet Hatchet()Hatchet()构造时会从环境变量加载配置包括上一节的 Token 与 TLS 策略并组装出 admin、dispatcher、event、listener 等一系列子客户端。从 hatchet.py 源码 可以看到hatchet实例还暴露了workers、workflows、dispatcher、event、listener、stubs等属性分别对应 Worker 管理、工作流声明、任务分发、事件推送与运行结果监听等能力后续进阶开发都会用到。2.2 声明第一个任务workflows/first_task.py 演示了「任务 输入模型 装饰器 处理函数」的声明范式from pydantic import BaseModel from hatchet_sdk import Context from ..hatchet_client import hatchet # Simple task class SimpleInput(BaseModel): message: str class SimpleOutput(BaseModel): transformed_message: str # Declare the task to run hatchet.task(namefirst-task, input_validatorSimpleInput) def first_task(input: SimpleInput, ctx: Context) - SimpleOutput: print(first-task task called) return SimpleOutput(transformed_messageinput.message.lower())关键点拆解hatchet.task(namefirst-task, input_validatorSimpleInput)将普通 Python 函数注册为 Hatchet 任务。name是任务在引擎侧的标识input_validator声明输入模型SDK 会在任务执行前用 Pydantic 对入参做反序列化与校验。input: SimpleInput任务函数接收经过校验的输入对象。ctx: Context任务上下文封装了 worker 信息、任务运行 ID、日志、步骤输出、重试与并发控制等能力见 context.py 中worker、worker_id、task_run_id等属性。本例虽然未使用但它是后续编写真实业务逻辑的核心参数。返回值SimpleOutput返回的 Pydantic 模型会被序列化保存为任务输出供下游任务或调用方读取。从 SDK 的 Stubs 客户端实现stubs.py可以推断hatchet.task这类声明式 API 最终会构建为Standalone独立任务或Workflow类型的可运行对象——这也是hatchet.worker(workflows[...])能够直接接收first_task的原因。第三步启动 Worker 并触发任务3.1 启动 Worker终端一worker.py 创建并启动一个名为first-worker的 Workerfrom .hatchet_client import hatchet from .workflows.first_task import first_task def main() - None: worker hatchet.worker( first-worker, slots10, workflows[first_task], ) worker.start() if __name__ __main__: main()运行命令poetry run python src/worker.py参数说明对应 hatchet.py 中worker方法签名slots10Worker 同时执行标准任务的并发槽位数决定该 Worker 可并行承接的任务数量上限。此外还支持durable_slots持久化任务槽位、labelsWorker 标签用于运行时亲和性调度与lifespanWorker 生命周期钩子。workflows[first_task]将任务注册到该 Worker。SDK 会在内部校验传入对象必须是BaseWorkflow子类Workflow或Standalone否则抛出TypeError并提示常见误用例如误传了Workflow.task的返回值。worker.start()阻塞式启动Worker 会持续向引擎发起长连接等待并执行被派发的任务。3.2 触发任务终端二保持 Worker 终端运行新开一个终端执行poetry run python src/run.pyrun.py 使用异步 API 直接运行任务import asyncio from .workflows.first_task import SimpleInput, first_task async def main() - None: # Run a Task result await first_task.aio_run(SimpleInput(messageHello World!)) print( Finished running task, and got the transformed message! The transformed message is:, result.transformed_message, ) if __name__ __main__: asyncio.run(main())执行流程如下构造SimpleInput(messageHello World!)输入对象await first_task.aio_run(...)将任务提交给引擎由引擎调度到已注册该任务的 Worker即终端一中运行的first-worker执行Worker 上的first_task函数将消息转为小写并返回SimpleOutput(transformed_messagehello world!)触发端收到结果打印first-task task called # 来自 Worker 终端 Finished running task, and got the transformed message! The transformed message is: hello world!这就完成了一次完整的「声明任务 → Worker 注册 → 引擎调度 → 执行 → 回传结果」闭环。运行模式与关键配置速查为便于日常使用将本示例涉及的核心配置整理如下配置项环境变量 / 参数说明默认值API TokenHATCHET_CLIENT_TOKEN认证凭证JWT 中携带 tenant 与引擎地址信息必填无默认TLS 策略HATCHET_CLIENT_TLS_STRATEGYtls默认面向云端/HTTPS自托管 HTTP 场景设为nonetlsWorker 名称hatchet.worker(first-worker, ...)Worker 在引擎中的唯一标识必填并发槽位worker(slots10)标准任务并发上限由 SDK 计算任务名称hatchet.task(namefirst-task)任务在引擎侧的标识函数名输入校验input_validatorSimpleInputPydantic 模型执行前校验入参无依赖管理Poetry安装hatchet-sdk与python-dotenv—除标准任务外SDK 还提供hatchet.durable_task、hatchet.workflow等声明 API 以支持持久化工作流、DAG 编排与定时/事件触发等高级场景本示例的aio_run对应后台管理 API 中的aio_run_workflow见 admin.py属于任务触发的底层实现之一。进阶方向从 Quickstart 到真实项目掌握 Quickstart 后可以基于仓库内更丰富的 Python 示例继续深入见 examples/python 目录例如并发与限流concurrency_limit、concurrency_cancel_in_progress、rate_limit等目录演示任务级/工作流级并发控制定时与事件触发cron、scheduled、events目录演示按 Cron 表达式或事件驱动任务重试与超时retries、timeout、non_retryable目录演示失败重试策略子任务与扇出child、fanout、fanout_sync目录演示父任务派生子任务与批量扇出模式持久化任务durable、durable_sleep、durable_event目录演示可暂停/恢复的持久化执行。同时可以阅读 Python SDK 测试 与 SDK 文档目录进一步理解引擎行为约定。至此你已具备从零搭建、运行并扩展 Hatchet Python 任务的基础能力。【免费下载链接】hatchet An orchestration engine for background tasks, AI agents, and durable workflows项目地址: https://gitcode.com/GitHub_Trending/ha/hatchet创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Hatchet Python 快速入门实战：用 Python 编写并运行第一个后台任务工作流
Hatchet Python 快速入门实战用 Python 编写并运行第一个后台任务工作流【免费下载链接】hatchet An orchestration engine for background tasks, AI agents, and durable workflows项目地址: https://gitcode.com/GitHub_Trending/ha/hatchetHatchet 是一个面向后台任务、AI Agent 与持久化工作流durable workflows的编排引擎。本指南以仓库内 examples/python/quickstart 示例项目为主线完整讲解从环境准备、工程初始化、Worker 启动到任务触发执行的端到端流程并结合 Python SDK 源码剖析Hatchet()客户端初始化、hatchet.task任务声明、worker启动等底层机制。读完本文你将能够独立搭建一个最小可用的 Hatchet Python 工程并理解任务输入输出校验、Worker 槽位slots与 Token/TLS 配置背后的原理。前置条件在运行示例项目之前需要准备以下两项基础环境Python v3.10 或更高版本SDK 与示例工程的类型注解、语法特性依赖 3.10。示例工程 pyproject.toml 中声明了python ^3.10的版本约束。Poetry用于依赖管理与虚拟环境隔离。示例工程采用 Poetry 作为包管理器其pyproject.toml通过[tool.poetry.dependencies]声明了hatchet-sdk 1.33.6与python-dotenv 1.2.2两个核心依赖其中python-dotenv用于从.env文件加载环境变量。了解示例项目结构先用cd examples/python/quickstart进入示例目录其结构如下quickstart/ ├── README.md # 官方快速入门说明 ├── hatchet_client.py # 初始化 Hatchet 客户端全局单例 ├── worker.py # 定义并启动 Worker ├── run.py # 触发任务执行的入口 ├── pyproject.toml # Poetry 工程配置 ├── poetry.lock # 依赖锁定文件 └── workflows/ ├── __init__.py └── first_task.py # 第一个任务的声明与实现该工程刻意保持极简一个任务声明文件、一个 Worker 启动文件、一个任务触发文件覆盖了「声明任务 → 注册到 Worker → 触发执行」的完整链路非常适合作为上手模板。第一步创建项目并安装依赖1.1 使用 Hatchet CLI 脚手架创建工程官方推荐通过 Hatchet CLI 生成该示例工程hatchet quickstart该命令会拉取当前仓库examples/python/quickstart同源的模板工程Python、TypeScript、Go、Ruby 等语言均有对应版本。若你已经在本地克隆了本仓库也可以直接复用examples/python/quickstart目录下的现成代码。1.2 配置环境变量创建工程后需要设置HATCHET_CLIENT_TOKEN环境变量其值来自 Hatchet 控制台Dashboard中创建的 API Tokenexport HATCHET_CLIENT_TOKENtoken自托管提示如果你是自托管部署且未启用 TLS需要额外设置HATCHET_CLIENT_TLS_STRATEGYnone来关闭 TLSexport HATCHET_CLIENT_TLS_STRATEGYnone从 SDK 配置源码 可以看到ClientTLSConfig以HATCHET_CLIENT_TLS_为环境变量前缀strategy字段默认值为tls自托管 HTTP 场景必须显式覆盖为none。同时 SDK 会在初始化时通过 token 解析逻辑 校验 Token 是否已设置并从 JWT 中解析出tenant_id与 gRPC 广播地址host_portget_tenant_id_from_jwt/grpc_broadcast_address_from_jwt这意味着只要提供合法的HATCHET_CLIENT_TOKENSDK 即可自动推导出需要连接的引擎地址无需手工配置主机端口。另外SDK 的配置加载基于 pydantic-settings支持从.env、.env.hatchet、.env.dev、.env.local等环境文件读取配置见 config.py因此也可以将 Token 写入项目根目录的.env文件配合python-dotenv自动加载。1.3 安装依赖poetry install该命令会根据pyproject.toml与poetry.lock创建虚拟环境并安装hatchet-sdk、python-dotenv等全部依赖。工程还在[tool.poetry.scripts]中注册了两个快捷命令[tool.poetry.scripts] simple src.run:main worker src.worker:main安装完成后可直接通过poetry run simple与poetry run worker调用等价于执行src/run.py与src/worker.py的main函数注意脚本入口路径中的src目录即示例工程代码所在包。第二步理解客户端初始化与任务声明2.1 初始化 Hatchet 客户端hatchet_client.py 是整个工程的基石仅有两行from hatchet_sdk import Hatchet # Initialize Hatchet client hatchet Hatchet()Hatchet()构造时会从环境变量加载配置包括上一节的 Token 与 TLS 策略并组装出 admin、dispatcher、event、listener 等一系列子客户端。从 hatchet.py 源码 可以看到hatchet实例还暴露了workers、workflows、dispatcher、event、listener、stubs等属性分别对应 Worker 管理、工作流声明、任务分发、事件推送与运行结果监听等能力后续进阶开发都会用到。2.2 声明第一个任务workflows/first_task.py 演示了「任务 输入模型 装饰器 处理函数」的声明范式from pydantic import BaseModel from hatchet_sdk import Context from ..hatchet_client import hatchet # Simple task class SimpleInput(BaseModel): message: str class SimpleOutput(BaseModel): transformed_message: str # Declare the task to run hatchet.task(namefirst-task, input_validatorSimpleInput) def first_task(input: SimpleInput, ctx: Context) - SimpleOutput: print(first-task task called) return SimpleOutput(transformed_messageinput.message.lower())关键点拆解hatchet.task(namefirst-task, input_validatorSimpleInput)将普通 Python 函数注册为 Hatchet 任务。name是任务在引擎侧的标识input_validator声明输入模型SDK 会在任务执行前用 Pydantic 对入参做反序列化与校验。input: SimpleInput任务函数接收经过校验的输入对象。ctx: Context任务上下文封装了 worker 信息、任务运行 ID、日志、步骤输出、重试与并发控制等能力见 context.py 中worker、worker_id、task_run_id等属性。本例虽然未使用但它是后续编写真实业务逻辑的核心参数。返回值SimpleOutput返回的 Pydantic 模型会被序列化保存为任务输出供下游任务或调用方读取。从 SDK 的 Stubs 客户端实现stubs.py可以推断hatchet.task这类声明式 API 最终会构建为Standalone独立任务或Workflow类型的可运行对象——这也是hatchet.worker(workflows[...])能够直接接收first_task的原因。第三步启动 Worker 并触发任务3.1 启动 Worker终端一worker.py 创建并启动一个名为first-worker的 Workerfrom .hatchet_client import hatchet from .workflows.first_task import first_task def main() - None: worker hatchet.worker( first-worker, slots10, workflows[first_task], ) worker.start() if __name__ __main__: main()运行命令poetry run python src/worker.py参数说明对应 hatchet.py 中worker方法签名slots10Worker 同时执行标准任务的并发槽位数决定该 Worker 可并行承接的任务数量上限。此外还支持durable_slots持久化任务槽位、labelsWorker 标签用于运行时亲和性调度与lifespanWorker 生命周期钩子。workflows[first_task]将任务注册到该 Worker。SDK 会在内部校验传入对象必须是BaseWorkflow子类Workflow或Standalone否则抛出TypeError并提示常见误用例如误传了Workflow.task的返回值。worker.start()阻塞式启动Worker 会持续向引擎发起长连接等待并执行被派发的任务。3.2 触发任务终端二保持 Worker 终端运行新开一个终端执行poetry run python src/run.pyrun.py 使用异步 API 直接运行任务import asyncio from .workflows.first_task import SimpleInput, first_task async def main() - None: # Run a Task result await first_task.aio_run(SimpleInput(messageHello World!)) print( Finished running task, and got the transformed message! The transformed message is:, result.transformed_message, ) if __name__ __main__: asyncio.run(main())执行流程如下构造SimpleInput(messageHello World!)输入对象await first_task.aio_run(...)将任务提交给引擎由引擎调度到已注册该任务的 Worker即终端一中运行的first-worker执行Worker 上的first_task函数将消息转为小写并返回SimpleOutput(transformed_messagehello world!)触发端收到结果打印first-task task called # 来自 Worker 终端 Finished running task, and got the transformed message! The transformed message is: hello world!这就完成了一次完整的「声明任务 → Worker 注册 → 引擎调度 → 执行 → 回传结果」闭环。运行模式与关键配置速查为便于日常使用将本示例涉及的核心配置整理如下配置项环境变量 / 参数说明默认值API TokenHATCHET_CLIENT_TOKEN认证凭证JWT 中携带 tenant 与引擎地址信息必填无默认TLS 策略HATCHET_CLIENT_TLS_STRATEGYtls默认面向云端/HTTPS自托管 HTTP 场景设为nonetlsWorker 名称hatchet.worker(first-worker, ...)Worker 在引擎中的唯一标识必填并发槽位worker(slots10)标准任务并发上限由 SDK 计算任务名称hatchet.task(namefirst-task)任务在引擎侧的标识函数名输入校验input_validatorSimpleInputPydantic 模型执行前校验入参无依赖管理Poetry安装hatchet-sdk与python-dotenv—除标准任务外SDK 还提供hatchet.durable_task、hatchet.workflow等声明 API 以支持持久化工作流、DAG 编排与定时/事件触发等高级场景本示例的aio_run对应后台管理 API 中的aio_run_workflow见 admin.py属于任务触发的底层实现之一。进阶方向从 Quickstart 到真实项目掌握 Quickstart 后可以基于仓库内更丰富的 Python 示例继续深入见 examples/python 目录例如并发与限流concurrency_limit、concurrency_cancel_in_progress、rate_limit等目录演示任务级/工作流级并发控制定时与事件触发cron、scheduled、events目录演示按 Cron 表达式或事件驱动任务重试与超时retries、timeout、non_retryable目录演示失败重试策略子任务与扇出child、fanout、fanout_sync目录演示父任务派生子任务与批量扇出模式持久化任务durable、durable_sleep、durable_event目录演示可暂停/恢复的持久化执行。同时可以阅读 Python SDK 测试 与 SDK 文档目录进一步理解引擎行为约定。至此你已具备从零搭建、运行并扩展 Hatchet Python 任务的基础能力。【免费下载链接】hatchet An orchestration engine for background tasks, AI agents, and durable workflows项目地址: https://gitcode.com/GitHub_Trending/ha/hatchet创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
