Hatchet Python SDK

1.32.1 · active · verified Sat Apr 11

This is the official Python SDK for Hatchet, a distributed, fault-tolerant task queue. The SDK allows you to easily integrate Hatchet's task scheduling and workflow orchestration capabilities into your Python applications, supporting the development of mission-critical AI agents, durable workflows, and background tasks. It is actively maintained, with regular updates and a focus on durability and scalability.

Warnings

breaking Migration from V0 to V1 SDK: `timeout` and `schedule_timeout` fields are now `datetime.timedelta` objects instead of strings (e.g., `"10s"` becomes `timedelta(seconds=10)`).
Fix: Update string-based timeouts to `datetime.timedelta` objects.
breaking Migration from V0 to V1 SDK: External-facing protobuf objects (e.g., `StickyStrategy`, `ConcurrencyLimitStrategy`) have been replaced by native Python enums.
Fix: Replace direct usage of protobuf objects with their corresponding Python enum equivalents.
breaking Migration from V0 to V1 SDK: Asynchronous methods throughout the SDK are now prefixed by `aio_` (e.g., `workflow.run()` is now `workflow.aio_run()` for async execution).
Fix: Prefix asynchronous method calls with `aio_`.
breaking Migration from V0 to V1 SDK: The `max_runs` parameter on the worker has been renamed to `slots`.
Fix: Update `max_runs` parameter to `slots` when configuring workers.
breaking Migration from V0 to V1 SDK: Tasks now have a new signature, taking `input` and `context` as arguments, and are typically declared using `@hatchet.task()` instead of requiring an explicit workflow definition first.
Fix: Update task function signatures to accept `input` and `context` arguments and use the `@hatchet.task()` decorator.
gotcha V0 Hatchet engine is not compatible with V1 workflows. The V0 engine entered End-of-Life on September 30, 2025. Ensure your Hatchet engine is upgraded to V1 or later when using the V1 Python SDK.
Fix: Upgrade your Hatchet engine to V1 or later to ensure compatibility with the V1 Python SDK.
gotcha Concurrency keys that reference fields in a task's `input` will be checked for validity at runtime. The `input_validator` Pydantic model for the task must explicitly contain any fields used in a concurrency key (e.g., if key is `input.user_id`, `input_validator` must have a `user_id` field).
Fix: Ensure your task's `input_validator` Pydantic model explicitly defines all fields used in concurrency keys.

Install

pip install hatchet-sdk Install latest version

Imports

Hatchet
```
from hatchet_sdk import Hatchet
```
Context
```
from hatchet_sdk import Context
```
DurableContext
```
from hatchet_sdk import DurableContext
```
EmptyModel
```
from hatchet_sdk import EmptyModel
```

Quickstart

This quickstart initializes the Hatchet client using environment variables for the API token and host, defines a simple task with the `@hatchet.task()` decorator, registers it with a worker, and starts the worker to process incoming tasks. Replace 'YOUR_HATCHET_CLIENT_TOKEN' with your actual token.

import os
from hatchet_sdk import Hatchet, Context, EmptyModel

# Initialize the Hatchet client. HATCHET_CLIENT_TOKEN is required.
# HATCHET_CLIENT_HOST_PORT defaults to 'localhost:7077' if not set.
hatchet = Hatchet(
    token=os.environ.get('HATCHET_CLIENT_TOKEN', 'YOUR_HATCHET_CLIENT_TOKEN'),
    host_port=os.environ.get('HATCHET_CLIENT_HOST_PORT', 'localhost:7077')
)

# Define a simple Hatchet task using the decorator
@hatchet.task()
def hello_world(input: EmptyModel, context: Context) -> dict[str, str]:
    print(f"Executing task hello_world with input: {input.model_dump_json()}")
    return {"message": "Hello, world from Hatchet!"}

# Create a worker and register the task
worker = hatchet.worker("my-python-worker")
worker.register_task(hello_world)

print("Starting Hatchet worker... Press Ctrl+C to stop.")
# Start the worker to listen for and execute tasks
worker.start()
print("Hatchet worker stopped.")

view raw JSON →