{"id":2966,"library":"hatchet-sdk","title":"Hatchet Python SDK","description":"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.","status":"active","version":"1.32.1","language":"en","source_language":"en","source_url":"https://github.com/hatchet-dev/hatchet-python","tags":["task queue","workflow orchestration","distributed","fault-tolerant","AI agents","background tasks"],"install":[{"cmd":"pip install hatchet-sdk","lang":"bash","label":"Install latest version"}],"dependencies":[],"imports":[{"symbol":"Hatchet","correct":"from hatchet_sdk import Hatchet"},{"symbol":"Context","correct":"from hatchet_sdk import Context"},{"symbol":"DurableContext","correct":"from hatchet_sdk import DurableContext"},{"symbol":"EmptyModel","correct":"from hatchet_sdk import EmptyModel"}],"quickstart":{"code":"import os\nfrom hatchet_sdk import Hatchet, Context, EmptyModel\n\n# Initialize the Hatchet client. HATCHET_CLIENT_TOKEN is required.\n# HATCHET_CLIENT_HOST_PORT defaults to 'localhost:7077' if not set.\nhatchet = Hatchet(\n    token=os.environ.get('HATCHET_CLIENT_TOKEN', 'YOUR_HATCHET_CLIENT_TOKEN'),\n    host_port=os.environ.get('HATCHET_CLIENT_HOST_PORT', 'localhost:7077')\n)\n\n# Define a simple Hatchet task using the decorator\n@hatchet.task()\ndef hello_world(input: EmptyModel, context: Context) -> dict[str, str]:\n    print(f\"Executing task hello_world with input: {input.model_dump_json()}\")\n    return {\"message\": \"Hello, world from Hatchet!\"}\n\n# Create a worker and register the task\nworker = hatchet.worker(\"my-python-worker\")\nworker.register_task(hello_world)\n\nprint(\"Starting Hatchet worker... Press Ctrl+C to stop.\")\n# Start the worker to listen for and execute tasks\nworker.start()\nprint(\"Hatchet worker stopped.\")","lang":"python","description":"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."},"warnings":[{"fix":"Update string-based timeouts to `datetime.timedelta` objects.","message":"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)`).","severity":"breaking","affected_versions":"All V1.x.x versions when migrating from V0 SDK."},{"fix":"Replace direct usage of protobuf objects with their corresponding Python enum equivalents.","message":"Migration from V0 to V1 SDK: External-facing protobuf objects (e.g., `StickyStrategy`, `ConcurrencyLimitStrategy`) have been replaced by native Python enums.","severity":"breaking","affected_versions":"All V1.x.x versions when migrating from V0 SDK."},{"fix":"Prefix asynchronous method calls with `aio_`.","message":"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).","severity":"breaking","affected_versions":"All V1.x.x versions when migrating from V0 SDK."},{"fix":"Update `max_runs` parameter to `slots` when configuring workers.","message":"Migration from V0 to V1 SDK: The `max_runs` parameter on the worker has been renamed to `slots`.","severity":"breaking","affected_versions":"All V1.x.x versions when migrating from V0 SDK."},{"fix":"Update task function signatures to accept `input` and `context` arguments and use the `@hatchet.task()` decorator.","message":"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.","severity":"breaking","affected_versions":"All V1.x.x versions when migrating from V0 SDK."},{"fix":"Upgrade your Hatchet engine to V1 or later to ensure compatibility with the V1 Python SDK.","message":"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.","severity":"gotcha","affected_versions":"V1.x.x SDK with V0 engine."},{"fix":"Ensure your task's `input_validator` Pydantic model explicitly defines all fields used in concurrency keys.","message":"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).","severity":"gotcha","affected_versions":"All V1.x.x versions."}],"env_vars":null,"last_verified":"2026-04-11T00:00:00.000Z","next_check":"2026-07-10T00:00:00.000Z"}