Hatchet Python v2.0.0 Migration Guide

Hatchet Python V2 Migration Guide

This guide will help you migrate Hatchet workflows from the V1 SDK to the V2 SDK.

V2 is a focused cleanup release: it removes everything that was deprecated in V1 (primarily in V1.31.0), enforces types that were previously coerced or warned about, and makes async a first-class requirement for durable tasks. It also makes optional arguments keyword-only, takes every duration as a timedelta, replaces the SDK's enums with plain strings, and returns SDK models instead of generated REST models from the crons, scheduled, and runs clients. There are no new concepts to learn. The goal is to arrive at a codebase with no deprecation warnings.

How to migrate

1. Upgrade to the latest V1 release first and run your test suite with deprecation warnings turned into errors (python -W error::DeprecationWarning -m pytest). Most of the removals in V2 emitted a DeprecationWarning in V1, so this finds the bulk of the work while your code still runs.
2. Upgrade to V2 (pip install "hatchet-sdk>=2,<3") on Python 3.11 or newer.
3. Apply the renames in the quick reference below. These are mechanical and safe to apply with find and replace.
4. Work through the removals and the detailed sections that follow. These need a small code change at each call site.
5. Run a type checker (mypy or pyright) over your code. The SDK is fully typed, so nearly every remaining V1 usage shows up as a type error.
6. Run your tests and check the behavior changes section, which lists the changes a type checker cannot catch.

Migrating with a coding agent

This page is written so that it can be handed to a coding agent as is. A prompt along these lines works well:

Migrate this codebase from hatchet-sdk V1 to V2 by following the attached migration guide.

1. Search the codebase for every pattern in the "Quick reference" and
   "Removed APIs and their replacements" tables and apply the listed replacement.
2. Read each detailed section and fix any remaining usages it describes.
3. Run the type checker and the test suite, and fix what they report.
4. Go through the "Behavior changes with no type error" section and check each
   item by reading the code, since nothing will fail loudly for those.
5. List anything you could not migrate mechanically so I can review it.

Do not change any behavior beyond what the guide requires.

Quick reference

These are one-to-one renames. The left column is safe to search for directly.

Removed APIs and their replacements

These have no one-to-one rename. Each row links to the section that explains the change.

Behavior changes with no type error

A type checker will not flag these, so check them by hand:

• Tasks with no input validator receive None, not an EmptyModel() instance. Code that reads attributes off input in such a task, or checks isinstance(input, EmptyModel), needs updating.

• Running a task that returns nothing gives None, not an EmptyModel() instance.

• ctx.task_output(parent) returns a dict for a parent task with no return annotation, and so does running a hatchet.stubs.task with no output_validator, not an EmptyModel. Replace attribute access such as output.field with output["field"], or annotate the parent's return type.

• Passing an input to a task or workflow with no input_validator raises a TypeError. V1 dropped the input silently. Declare input_validator=dict[str, Any] if you need to pass untyped input.

• runs.list, runs.aio_list, workflow.list_runs, and workflow.aio_list_runs with no limit return every matching run. In V1, runs.list returned the first 50, and workflow.list_runs returned up to 100 runs for each day in the range. Pass a limit or a narrow time range if you only need recent runs. See the section on runs.list.

• from hatchet_sdk import Workflow is a different class. It is now the workflow class returned by hatchet.workflow(), not the REST model.

• An explicit retries=0, execution_timeout=timedelta(seconds=60), or schedule_timeout=timedelta(minutes=5) on Workflow.task and Workflow.durable_task is no longer replaced by the workflow's task_defaults. V1 could not tell these values apart from the parameter defaults. Remove the argument if you were relying on task_defaults taking over.

• Priorities are strings, so comparing them compares strings. In V1 Priority was an int enum, so ctx.priority >= Priority.MEDIUM worked. "HIGH" >= "MEDIUM" is still valid Python, but compares alphabetically. Convert with priority_to_int from hatchet_sdk.types.priority before comparing.

• Former enum values are plain strings when serialized or logged. Priority and WorkerLabelComparator were int enums, so a priority written into additional_metadata or JSON was 1, 2, or 3. It is now "LOW", "MEDIUM", or "HIGH". A status formats as "FAILED", not "RunStatus.FAILED".

• Worker.register_workflow raises on failure instead of calling sys.exit(1).

• WorkflowRunRef.result and TaskRunRef.result take timeout as their first positional parameter. In V1.41 the only parameter was poll_interval, so a positional call such as ref.result(2.0) must become ref.result(poll_interval=timedelta(seconds=2)).

• from hatchet_sdk import Event is a different class. It is now the SDK's own event class, returned by every method on the events client, not the V0 REST model.

• Event.payload and Event.additional_metadata are dictionaries, not JSON strings. A leftover json.loads(event.payload) raises a TypeError at runtime. See the Event class.

• ctx.task_run_errors holds TaskRunError objects, not strings. Code that logs or string-matches the values still runs, but sees the formatted error rather than the raw serialized one. See task run errors.

• BulkPushEventWithMetadata.key is required. It previously defaulted to "".

• ClientTLSConfig.strategy is validated. A value other than "tls", "mtls", or "none" now fails at construction time instead of being treated as insecure.

• The HATCHET_CLIENT_WORKER_HEALTHCHECK_EVENT_LOOP_BLOCK_THRESHOLD_SECONDS environment variable is no longer read. Rename it to HATCHET_CLIENT_WORKER_HEALTHCHECK_EVENT_LOOP_BLOCK_THRESHOLD.

• grpcio-tools is no longer installed with the SDK. Add it to your own dependencies if you import it.

• ref.result() and ref.aio_result() raise RuntimeError instead of ValueError when fetching the run's status fails repeatedly. A cancelled run still raises ValueError.

• parent_id, parent_step_run_id, child_index, and desired_worker_id are no longer valid in OpenTelemetryConfig.excluded_attributes or HATCHET_CLIENT_OPENTELEMETRY_EXCLUDED_ATTRIBUTES. Remove them, or building the ClientConfig fails.

• from hatchet_sdk import RunStatus, Priority, and the other former enums still succeed. The names remain as Literal type aliases, so the import works but member access such as RunStatus.COMPLETED raises AttributeError, and isinstance(value, RunStatus) raises TypeError. See enums replaced by strings.

• FailedTaskRunExceptionGroup is no longer a ValueError. It is an ExceptionGroup[TaskRunError], so an except ValueError around run() or result() no longer catches a failed run. Catch FailedTaskRunExceptionGroup or use except* TaskRunError. See the exception group.

• ctx.refresh_timeout, ctx.aio_refresh_timeout, ctx.put_stream, and ctx.aio_put_stream raise when the request fails. V1 logged the error and carried on. An unhandled error now fails the task. See context methods that raise.

• worker.start() returns after a graceful shutdown instead of calling sys.exit(0), so code after it now runs. A failure to start the action listener process raises the original exception instead of calling sys.exit(1). See worker startup and shutdown.

• on_failure_task and on_success_task now inherit the workflow's task_defaults for schedule_timeout, execution_timeout, retries, backoff_factor, and backoff_max. In V1 they always used 5 minutes, 60 seconds, and 0 retries. See on_failure_task and on_success_task.

• A ParentCondition's default readable_data_key changed from the parent's name plus a timestamp to the parent's name plus the expression. Two identical ParentConditions on one task now collide, and registering the workflow raises a ValueError. See ParentCondition.

• TaskDefaults rejects unknown fields. A leftover priority or backoff_max_seconds, including one passed by unpacking a dict, raises a ValidationError instead of being ignored.

• ScheduledRun.workflow_run_status is a RunStatus. A comparison against the V0 strings "SUCCEEDED" or "PENDING" is silently False. Compare against "COMPLETED" and "QUEUED". See the new models.

• input and additional_metadata on CronTrigger and ScheduledRun, and additional_metadata on TaskRunSummary, are {} when unset, not None. A check such as if cron.input is None is never true.

• The renamed environment variables are no longer read. Rename HATCHET_CLIENT_LISTENER_V2_TIMEOUT to HATCHET_CLIENT_ACTION_LISTENER_TIMEOUT, HATCHET_CLIENT_EMBEDDED_READY_TIMEOUT_SECONDS to HATCHET_CLIENT_EMBEDDED_READY_TIMEOUT, and HATCHET_CLIENT_OPENTELEMETRY_INCLUDE_TASK_NAME_IN_START_STEP_RUN_SPAN_NAME to HATCHET_CLIENT_OPENTELEMETRY_INCLUDE_TASK_NAME_IN_SPAN_NAME. Their values are unchanged.

• ctx.put_stream, ctx.aio_put_stream, and runs.aio_subscribe_to_stream emit a DeprecationWarning. A test suite run with -W error::DeprecationWarning fails on them. See run streams.

• The timedelta config fields still accept numbers at runtime. sync_result_poll_interval, action_listener_timeout, and ready_timeout read a number as seconds, so ClientConfig(sync_result_poll_interval=2) runs, but a type checker reports it.

The rest of this page covers each change in detail.

Python 3.11 or newer is required

V2 requires Python 3.11+. Python 3.10 reaches end-of-life in October 2026 and is no longer supported.

The SDK also no longer depends on grpcio-tools at runtime, since it was only ever needed for code generation. If your project relied on the SDK pulling it in transitively, add it to your own dependencies.

Running without wait

In V1, running a workflow without waiting for the result was done with dedicated _no_wait methods:

# V1
ref = workflow.run_no_wait(input)
ref = await workflow.aio_run_no_wait(input)
refs = workflow.run_many_no_wait([input1, input2])
refs = await workflow.aio_run_many_no_wait([input1, input2])

These methods have been removed in V2. Pass wait_for_result=False to the standard run methods instead:

# V2
ref = workflow.run(input, wait_for_result=False)
ref = await workflow.aio_run(input, wait_for_result=False)
refs = workflow.run_many([input1, input2], wait_for_result=False)
refs = await workflow.aio_run_many([input1, input2], wait_for_result=False)

Options objects removed in favor of inline parameters

PushEventOptions, BulkPushEventOptions, and passing a TriggerWorkflowOptions or ScheduleTriggerWorkflowOptions instance to run(), schedule(), create_bulk_run_item(), and related methods have all been removed. Pass the options as keyword arguments directly instead.

Events:

# V1
from hatchet_sdk import PushEventOptions

hatchet.event.push("my-event", payload, options=PushEventOptions(additional_metadata={"key": "value"}))

# V2
hatchet.events.push("my-event", payload, additional_metadata={"key": "value"})

events.push and events.aio_push accept additional_metadata, priority, and scope as keyword arguments.

BulkPushEventOptions has also been removed. Use BulkPushEventWithMetadata directly, which now requires a key (it previously defaulted to "").

Running and scheduling workflows:

# V1
from hatchet_sdk import ScheduleTriggerWorkflowOptions, TriggerWorkflowOptions

workflow.run(input, options=TriggerWorkflowOptions(additional_metadata={"key": "value"}, priority=1))
workflow.schedule(run_at, input, options=ScheduleTriggerWorkflowOptions(additional_metadata={"key": "value"}))

# V2
workflow.run(input, additional_metadata={"key": "value"}, priority="LOW")
workflow.schedule(run_at, input, additional_metadata={"key": "value"})

The keyword arguments accepted by run, aio_run, and create_bulk_run_item are child_key, additional_metadata, priority, sticky, and desired_worker_labels. run and aio_run also take wait_for_result.

desired_worker_id has been removed from run, aio_run, create_bulk_run_item, and TriggerWorkflowOptions. To run a child on the same worker as the task that triggers it, pass sticky=True. To steer a run toward particular workers, use desired_worker_labels.

Bulk run items:

# V1
workflow.create_bulk_run_item(
    input=input,
    key="child-1",
    options=TriggerWorkflowOptions(additional_metadata={"key": "value"}),
)

# V2
workflow.create_bulk_run_item(
    input=input,
    child_key="child-1",
    additional_metadata={"key": "value"},
)

The internal parent_id, parent_step_run_id, child_index, and key fields on TriggerWorkflowOptions / ScheduleTriggerWorkflowOptions have also been removed, along with the key field on WorkflowRunTriggerConfig. These were never intended for external use.

The namespace field on these options has also been removed. The namespace always comes from the ClientConfig now.

The desired_worker_label field on TriggerWorkflowOptions has been renamed to desired_worker_labels and only accepts a list[DesiredWorkerLabel] (previously it also accepted a dict[str, DesiredWorkerLabel]).

Optional arguments are keyword-only

Optional parameters that follow a method's main arguments are now keyword-only. Passing one positionally raises a TypeError, and a type checker reports it. The fix is to pass the argument by keyword:

# V1
@workflow.task("process", timedelta(minutes=5))
def process(input: MyInput, ctx: Context) -> None: ...

workflow.run(input, False)
workflow.schedule(trigger_at, input, "child-1")
hatchet.cron.create("my-workflow", "nightly", "0 0 * * *", {"user_id": "1"}, {})
hatchet.rate_limits.put("my-key", 10, "MINUTE")
SleepCondition(timedelta(seconds=10), "wait-10s")
await ctx.aio_sleep_for(timedelta(seconds=10), "wait-10s")

# V2
@workflow.task("process", schedule_timeout=timedelta(minutes=5))
def process(input: MyInput, ctx: Context) -> None: ...

workflow.run(input, wait_for_result=False)
workflow.schedule(trigger_at, input, child_key="child-1")
hatchet.crons.create(
    "my-workflow", "nightly", "0 0 * * *", input={"user_id": "1"}, additional_metadata={}
)
hatchet.rate_limits.put("my-key", 10, duration="MINUTE")
SleepCondition(timedelta(seconds=10), readable_data_key="wait-10s")
await ctx.aio_sleep_for(timedelta(seconds=10), label="wait-10s")

These are the parameters that can still be passed positionally. Everything after them is keyword-only. Each aio_ variant matches its sync method.

input and additional_metadata on crons.create and scheduled.create are still required, and now must be passed by keyword.

hatchet.event renamed to hatchet.events, hatchet.cron renamed to hatchet.crons

The event and cron clients are now accessed as hatchet.events and hatchet.crons, consistent with the other (plural) feature clients:

# V1
hatchet.event.push("my-event", payload)
hatchet.cron.create(...)

# V2
hatchet.events.push("my-event", payload)
hatchet.crons.create(...)

The internal log, stream, and aio_stream methods and the token and namespace attributes on the events client are no longer public. Use hatchet.namespace to read the namespace. Use ctx.log and the streaming methods on Context instead.

Context property renames

Several properties on Context have been renamed or removed.

Note that ctx.done() was a method and ctx.is_cancelled is a property, so drop the parentheses. ctx.task_output is synchronous, so drop the await when replacing ctx.aio_task_output.

ctx.priority now returns "LOW", "MEDIUM", "HIGH", or None instead of an int | None, consistent with how priorities are passed everywhere else.

ctx.worker_labels is now a list[WorkerLabel] instead of a dict, and ctx.upsert_worker_labels and ctx.aio_upsert_worker_labels take a list[WorkerLabel], matching the labels argument to hatchet.worker. See worker labels.

ctx.pause_rate_limit and ctx.aio_pause_rate_limit have been removed. They did the same thing as hatchet.rate_limits.pause and hatchet.rate_limits.aio_pause, which are still available.

The raise_on_error parameter on ctx.log and ctx.aio_log has been removed. Logging never raises.

Internal context properties (ctx.data, ctx.action, ctx.dispatcher_client, ctx.admin_client, ctx.event_client, ctx.runs_client, ctx.log_sender, ctx.durable_event_listener) have also been removed. These were never intended for external use.

The WorkerContext class backing ctx.worker has also been removed and is no longer exported from hatchet_sdk.

Hatchet client changes

Hatchet.get_current_context() has been removed. Use the current_context property instead:

# V1
ctx = hatchet.get_current_context()

# V2
ctx = hatchet.current_context

The debug constructor parameter has been removed. Enable debug mode via the HATCHET_CLIENT_DEBUG environment variable instead:

HATCHET_CLIENT_DEBUG=true

The client constructor parameter has also been removed, along with the internal hatchet_sdk.client.Client class. Pass a ClientConfig via the config parameter instead.

Durations must be timedelta objects

Passing durations as strings (e.g. "10s") is no longer supported. Use datetime.timedelta objects:

from datetime import timedelta

# V1
@workflow.task(execution_timeout="10s", schedule_timeout="30m")
def my_task(input: MyInput, ctx: Context) -> None:
    ctx.refresh_timeout("15s")

# V2
@workflow.task(execution_timeout=timedelta(seconds=10), schedule_timeout=timedelta(minutes=30))
def my_task(input: MyInput, ctx: Context) -> None:
    ctx.refresh_timeout(timedelta(seconds=15))

This applies everywhere a duration is accepted:

• execution_timeout and schedule_timeout on every task decorator (task, durable_task, batch_task, on_failure_task, on_success_task) and on TaskDefaults
• ctx.refresh_timeout and ctx.aio_refresh_timeout
• ctx.aio_sleep_for
• SleepCondition

The suffixes map as s to seconds, m to minutes, and h to hours.

Durations with a unit in their name are now timedelta

Parameters and config fields that took a number of seconds, with the unit in their name, now take a timedelta and drop the unit:

# V1
@workflow.task(retries=5, backoff_factor=2.0, backoff_max_seconds=60)
def my_task(input: MyInput, ctx: Context) -> None: ...

# V2
@workflow.task(retries=5, backoff_factor=2.0, backoff_max=timedelta(seconds=60))
def my_task(input: MyInput, ctx: Context) -> None: ...

backoff_max is sent to the engine in whole seconds, so any fraction of a second is dropped. The environment variables for the config fields still take a number of seconds. See configuration changes for the renamed fields and environment variables.

Priority is a string

Passing priority as an integer is no longer supported, and neither is the V1 Priority enum. Pass "LOW", "MEDIUM", or "HIGH":

# V1
from hatchet_sdk import Priority

workflow = hatchet.workflow(name="MyWorkflow", default_priority=3)
workflow.run(input, priority=Priority.LOW)

# V2
workflow = hatchet.workflow(name="MyWorkflow", default_priority="HIGH")
workflow.run(input, priority="LOW")

"LOW", "MEDIUM", and "HIGH" correspond to the integer values 1, 2, and 3. Priority is still exported from hatchet_sdk as a type alias for annotations. To compare or sort priorities, convert them with priority_to_int from hatchet_sdk.types.priority, since comparing the strings directly is alphabetical.

This applies everywhere a priority is accepted: default_priority on workflow() and task() decorators, the priority argument to run(), schedule(), create_cron(), hatchet.crons.create(), hatchet.runs.create(), and hatchet.events.push(), and the priority field on BulkPushEventWithMetadata.

TaskDefaults.priority has been removed. It was never read, so setting it had no effect. Set default_priority on the workflow instead.

ConcurrencyExpression renamed to ConcurrencyStrategy

ConcurrencyExpression has been renamed to ConcurrencyStrategy, and its limit_strategy field is now strategy. The field takes a plain string instead of a member of the ConcurrencyLimitStrategy enum, which has been removed:

# V1
from hatchet_sdk import ConcurrencyExpression, ConcurrencyLimitStrategy

workflow = hatchet.workflow(
    name="MyWorkflow",
    concurrency=ConcurrencyExpression(
        expression="input.user_id",
        max_runs=5,
        limit_strategy=ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,
    ),
)

# V2
from hatchet_sdk import ConcurrencyStrategy

workflow = hatchet.workflow(
    name="MyWorkflow",
    concurrency=ConcurrencyStrategy(
        expression="input.user_id",
        max_runs=5,
        strategy="CANCEL_IN_PROGRESS",
    ),
)

The accepted values are "CANCEL_IN_PROGRESS", "GROUP_ROUND_ROBIN", "CANCEL_NEWEST", "CANCEL_QUEUED_EXCEPT_NEWEST", and "CANCEL_QUEUED_EXCEPT_OLDEST", which are the names of the old enum members.

The strategy field is typed as a Literal, so type checkers still catch invalid values, and Pydantic validates them at construction time. The QUEUE_NEWEST and QUEUE_OLDEST members of the old enum have no replacement: the engine no longer supports those strategies.

Enums replaced by strings

Every fixed set of choices in the SDK is now a plain string, typed as a Literal so that type checkers and editors still validate and autocomplete the values. The enums have been removed. Where a name is still useful in annotations (RunStatus, Priority, SlotType, WorkerLabelComparator, HTTPMethod, OTelAttribute, WorkerStatus), it survives as a Literal type alias, so an import such as from hatchet_sdk import RunStatus still succeeds, but member access such as RunStatus.COMPLETED raises AttributeError:

# V1
from hatchet_sdk import RateLimit, RateLimitDuration, RunStatus, StickyStrategy
from hatchet_sdk.runnables.workflow import MCPProvider

workflow = hatchet.workflow(name="MyWorkflow", sticky=StickyStrategy.SOFT)
hatchet.rate_limits.put("my-key", 10, RateLimitDuration.MINUTE)
limit = RateLimit(static_key="my-key", units=1, duration=RateLimitDuration.MINUTE)
tool = my_task.mcp_tool(MCPProvider.CLAUDE)
failed = hatchet.runs.list(statuses=[RunStatus.FAILED])

# V2
from hatchet_sdk import RateLimit

workflow = hatchet.workflow(name="MyWorkflow", sticky="SOFT")
hatchet.rate_limits.put("my-key", 10, duration="MINUTE")
limit = RateLimit(static_key="my-key", units=1, duration="MINUTE")
tool = my_task.mcp_tool("CLAUDE")
failed = hatchet.runs.list(statuses=["FAILED"])

In most cases the string is the name of the old enum member:

Two take the old member's value rather than its name:

The ordering parameters on crons.list, scheduled.list, and rate_limits.list use snake_case field names and lowercase directions on every client, instead of the REST API's mix of spellings:

The ordering and webhook enums (CronWorkflowsOrderByField, ScheduledWorkflowsOrderByField, RateLimitOrderByField, RateLimitOrderByDirection, WorkflowRunOrderByDirection, V1WebhookSourceName, V1WebhookHMACAlgorithm, and V1WebhookHMACEncoding) are no longer exported from hatchet_sdk. The generated REST models still use their own enums, so a field other than status read off a raw REST model may still be a REST enum member. Those enums subclass str, so comparing them against a string works.

Worker labels must use typed objects

Passing desired_worker_labels as a dict is no longer supported. Use a list[DesiredWorkerLabel] with the key field set:

from hatchet_sdk import DesiredWorkerLabel

# V1
@workflow.task(desired_worker_labels={"region": {"value": "us-east-1", "comparator": 4}})
def my_task(input: MyInput, ctx: Context) -> None: ...

# V2
@workflow.task(
    desired_worker_labels=[
        DesiredWorkerLabel(
            key="region",
            value="us-east-1",
            comparator="EQUAL",
        )
    ]
)
def my_task(input: MyInput, ctx: Context) -> None: ...

The comparator field on DesiredWorkerLabel takes the comparator's name as a string, such as "EQUAL" or "LESS_THAN", rather than an integer.

The same applies to the labels a worker advertises. The labels argument to hatchet.worker(...) no longer accepts a dict. Use a list[WorkerLabel]:

from hatchet_sdk import WorkerLabel

# V1
worker = hatchet.worker(
    "my-worker",
    labels={"model": "fancy-ai-model-v2", "memory": 512},
)

# V2
worker = hatchet.worker(
    "my-worker",
    labels=[
        WorkerLabel(key="model", value="fancy-ai-model-v2"),
        WorkerLabel(key="memory", value=512),
    ],
)

Each dict key becomes the key of a WorkerLabel, and each dict value becomes its value.

The labels a task reads and updates through its context use the same type. ctx.worker_labels is a list[WorkerLabel], and ctx.upsert_worker_labels and ctx.aio_upsert_worker_labels take one:

# V1
if ctx.worker_labels.get("model") != "fancy-ai-model-v2":
    await ctx.aio_upsert_worker_labels({"model": "fancy-ai-model-v2"})

# V2
model_labels = [label for label in ctx.worker_labels if label.key == "model"]

if not model_labels or model_labels[0].value != "fancy-ai-model-v2":
    await ctx.aio_upsert_worker_labels(
        [WorkerLabel(key="model", value="fancy-ai-model-v2")]
    )

Upserting a label replaces any existing label with the same key and keeps the others.

WorkerLabel.key is now required. In V1 it defaulted to None, and a label without a key was silently dropped. DesiredWorkerLabel no longer subclasses WorkerLabel, so isinstance(label, WorkerLabel) is False for a DesiredWorkerLabel.

Durable tasks must be async

Synchronous durable tasks are no longer supported, and declaring one raises a TypeError. Convert durable task functions to async def:

# V1
@workflow.durable_task()
def my_durable_task(input: MyInput, ctx: DurableContext) -> dict[str, str]:
    ...

# V2
@workflow.durable_task()
async def my_durable_task(input: MyInput, ctx: DurableContext) -> dict[str, str]:
    ...

Tasks with no input validator now receive None

EmptyModel has been removed. Tasks and workflows declared without an input type previously received an EmptyModel() instance as their input; they now receive None:

# V1
from hatchet_sdk import EmptyModel

@hatchet.task()
def my_task(input: EmptyModel, ctx: Context) -> None: ...

my_task.run(EmptyModel())

# V2
@hatchet.task()
def my_task(input: None, ctx: Context) -> None: ...

my_task.run()

The same applies to results: running a task that returns nothing now gives None rather than an EmptyModel() instance.

If you used EmptyModel as a base class for your own models, subclass pydantic.BaseModel instead.

For a workflow declared without an input_validator, the workflow.input_validator and workflow.input_validator_type properties now return None.

Durable wait results are now typed

ctx.aio_wait_for previously returned a raw dict[str, Any]. It now returns a list[SleepResult | EventWaitResult[Any] | OrGroupResult], one entry per satisfied condition (an OrGroup contributes a single OrGroupResult). ctx.aio_sleep_for returns a SleepResult, and ctx.aio_wait_for_event returns an EventWaitResult, whose .payload holds the event data (validated against payload_validator if one was provided):

# V1
result = await ctx.aio_wait_for("my-signal", UserEventCondition(event_key="my-event"))
payload = result["my-event"]

event = await ctx.aio_wait_for_event("my-event")
user_id = event["user_id"]

# V2
from hatchet_sdk import EventWaitResult

results = await ctx.aio_wait_for(UserEventCondition(event_key="my-event"))
event_result = next(r for r in results if isinstance(r, EventWaitResult))
payload = event_result.payload

event = await ctx.aio_wait_for_event("my-event")
user_id = event.payload["user_id"]

An OrGroupResult wraps whichever condition in the group was satisfied, available as .result.

Each condition passed to ctx.aio_wait_for, including the conditions inside each OrGroup, must have a distinct readable_data_key, or the call raises a ValueError. The default key is built from the condition itself (sleep:30s for a 30 second sleep, or the event key), so two OrGroups that both contain SleepCondition(timedelta(seconds=30)) need an explicit readable_data_key on at least one of them.

ctx.aio_wait_for no longer takes a signal key as its first argument. Pass the conditions directly. If you need a specific key, pass signal_key by keyword. When it is left out, the key is generated from the order of the wait within the task run, which is stable across replays:

# V1
await ctx.aio_wait_for("my-signal", SleepCondition(timedelta(minutes=5)))

# V2
await ctx.aio_wait_for(SleepCondition(timedelta(minutes=5)))
await ctx.aio_wait_for(SleepCondition(timedelta(minutes=5)), signal_key="my-signal")

label on ctx.aio_sleep_for and ctx.aio_sleep_until is keyword-only. ctx.aio_sleep_until with a time in the past now completes immediately.

DurableContext.wait_index has been removed. It was internal.

Internal Worker properties removed

Several internal Worker properties that were never intended for external use have been removed: worker.config, worker.client, worker.slot_config, worker.slots, worker.durable_slots, worker.debug, worker.labels, worker.lifespan, worker.loop, worker.handle_kill, worker.owned_loop, worker.action_registry, worker.killing, worker.action_listener_process, worker.durable_action_listener_process, worker.action_listener_health_check, worker.action_runner, worker.ctx, worker.action_queue, worker.event_queue, worker.durable_action_queue, worker.durable_event_queue, and worker.lifespan_stack. If your code accessed any of these, remove those usages.

The Worker constructor no longer accepts debug (set HATCHET_CLIENT_DEBUG=true instead), and its labels parameter only accepts a list[WorkerLabel], as described in worker labels.

worker.start() no longer accepts an options argument, and WorkerStartOptions has been removed. worker.exit_gracefully() is no longer public. Send the process SIGTERM to shut a worker down.

TaskRunError.serialize, TaskRunError.deserialize, and TaskRunError.from_exception are no longer public. Use str(error) for the formatted error, or read error.exc, error.exc_type, and error.trace.

The config, tracer_provider, and meter_provider attributes and the extract_bound_args method on HatchetInstrumentor are no longer public. Keep your own reference to the providers you pass in. All of HatchetInstrumentor's parameters are keyword-only.

The same applies to internals on the workflow and task objects: workflow.config, workflow.client, workflow.service_name, workflow.create_action_name, workflow.to_proto, task.fn, task.workflow, task.is_durable, task.is_async_function, task.validators, task.slot_requests, task.call, task.aio_call, task.to_proto, Standalone.is_async_function, and Depends(...).fn are no longer public. Neither are the task settings: task.type, task.retries, task.execution_timeout, task.schedule_timeout, task.parents, task.rate_limits, task.desired_worker_labels, task.backoff_factor, task.backoff_max_seconds, task.concurrency, task.wait_for, task.skip_if, task.cancel_if, task.batch, and task.eviction_policy. task.name is still public. On conditions, the base attribute, the to_proto method, and the hatchet_sdk.conditions.Action enum are internal too. The workflow_run_listener, workflow_run_event_listener, and admin_client properties on hatchet.runs, WorkflowRunRef, and TaskRunRef have been removed too. Use workflow.name for the workflow's name, and mock_run / aio_mock_run to call a task's underlying function in unit tests.

List methods now return plain lists

All list and aio_list methods on feature clients previously returned paginated wrapper types (e.g. V1EventList, WorkerList, WorkflowList). In V2, they return plain Python lists. Update any code that accessed .rows, .pagination, or other wrapper properties:

# V1
result = hatchet.runs.list(...)
runs = result.rows  # V1TaskSummaryList

# V2
runs = hatchet.runs.list(...)  # list[TaskRunSummary] directly

This applies to all feature clients:

Event, CronTrigger, TaskRunSummary, and ScheduledRun are SDK models exported from hatchet_sdk. See the new models. RateLimit, Worker, and Workflow in this table are the REST models from hatchet_sdk.clients.rest.models, not the top-level hatchet_sdk classes of the same name.

Since the wrappers are gone, patterns such as result.rows or [] become just result, and len(result.rows) becomes len(result).

Cron triggers, scheduled runs, and task runs use SDK models

The crons, scheduled, and runs clients returned generated REST models. They now return the SDK's own models, exported from hatchet_sdk. Like Event, these flatten metadata into top-level fields and use SDK types: RunStatus, Priority, timedelta, and dictionaries that are empty instead of None.

The fields that changed:

ScheduledRun.workflow_run_status maps the V0 statuses to RunStatus like this: PENDING, QUEUED, and BACKOFF become QUEUED, RUNNING stays RUNNING, SUCCEEDED becomes COMPLETED, and FAILED and CANCELLED are unchanged.

# V1
cron = hatchet.cron.create("my-workflow", "nightly", "0 0 * * *", {}, {})
print(cron.cron)
hatchet.cron.delete(cron.metadata.id)

scheduled = workflow.schedule(trigger_at, input)
hatchet.scheduled.delete(scheduled.scheduled_workflows[0].id)

for run in hatchet.runs.list(since=since):
    print(run.metadata.id, run.task_external_id, run.duration)  # duration in ms

# V2
cron = hatchet.crons.create(
    "my-workflow", "nightly", "0 0 * * *", input={}, additional_metadata={}
)
print(cron.expression)
hatchet.crons.delete(cron.id)

scheduled = workflow.schedule(trigger_at, input)
hatchet.scheduled.delete(scheduled.id)

for run in hatchet.runs.list(since=since):
    print(run.id, run.task_run_id, run.duration)  # duration is a timedelta

V1TaskSummary is no longer exported from hatchet_sdk. Use TaskRunSummary in type annotations.

runs.list_with_pagination and aio_list_with_pagination removed

The list_with_pagination and aio_list_with_pagination methods on the runs client have been removed. Use list and aio_list instead. They now handle pagination internally and return list[TaskRunSummary] directly:

# V1
runs = await hatchet.runs.aio_list_with_pagination(since=..., until=...)

# V2
runs = await hatchet.runs.aio_list(since=..., until=...)

limit and offset now count across the whole since to until range, newest run first. In V1, list_with_pagination applied them to each day in the range separately, so limit=10 over three days could return 30 runs. Leaving limit unset returns every matching run, so pass a limit or a narrow time range when you only need the most recent runs.

workflow.list_runs and workflow.aio_list_runs also default to limit=None, so they return every matching run instead of up to 100 per day. Their parameters are keyword-only, and parent_task_external_id and triggering_event_external_id are now parent_task_run_id and triggering_event_id.

ref.stream() and the streaming event types removed

The stream() method on WorkflowRunRef and TaskRunRef has been removed. It was marked internal in V1. The closest replacement is hatchet.runs.aio_subscribe_to_stream, which yields the payload of each stream chunk as a str. It is deprecated along with ctx.put_stream, so new code should use durable streams instead:

# V1
from hatchet_sdk import StepRunEventType

async for event in ref.stream():
    if event.type == StepRunEventType.STEP_RUN_EVENT_TYPE_STREAM:
        print(event.payload)

# V2
async for chunk in hatchet.runs.aio_subscribe_to_stream(ref.workflow_run_id):
    print(chunk)

aio_subscribe_to_stream only yields stream chunks (the values sent with ctx.put_stream), so the check on the event type is no longer needed. Run state changes such as started, completed, and failed are not part of the stream. Use ref.aio_result() or hatchet.runs.aio_get_status(run_id) to observe those.

Since nothing public yields them anymore, StepRunEventType and WorkflowRunEventType are no longer exported from hatchet_sdk, and the StepRunEvent class is gone. Remove the imports along with the event type checks.

hatchet.listener (with its stream, stream_by_run_id, stream_by_additional_metadata, and on methods) is no longer public either. Use hatchet.runs.aio_subscribe_to_stream in its place.

Iterating over a stream from a synchronous task is no longer supported. Make the task async and use aio_subscribe_to_stream.

OpenTelemetry helper functions removed

The create_traceparent, parse_carrier_from_metadata, and inject_traceparent_into_metadata functions in hatchet_sdk.opentelemetry.instrumentor have been removed. They have not been needed since V1.11.0: HatchetInstrumentor creates the traceparent and propagates it through the additional metadata of runs and events automatically. Delete the calls, and pass your metadata through unchanged:

# V1
from hatchet_sdk.opentelemetry.instrumentor import (
    create_traceparent,
    inject_traceparent_into_metadata,
)

workflow.run(
    input,
    additional_metadata=inject_traceparent_into_metadata(
        {"key": "value"}, create_traceparent()
    ),
)

# V2
workflow.run(input, additional_metadata={"key": "value"})

Top-level exports cleaned up

hatchet_sdk no longer re-exports the V0-era REST API models (StepRun, PullRequest, Job, JobRun, WorkflowRun, Tenant, User, the invite/token request models, and similar), the pagination wrapper types (EventList, WorkerList, WorkflowList, ...), the raw CreateWorkflowVersionOpts protobuf, or WorkerStartOptions. If you need a REST model directly, import it from hatchet_sdk.clients.rest.models.

Worker.register_workflow_from_opts, which took a raw protobuf, has been removed. Use worker.register_workflow(...) with a workflow object instead. The internal hatchet.dispatcher and hatchet.listener properties are also no longer public.

from hatchet_sdk import Event now resolves to the SDK's own event class (with seen_at, additional_metadata, etc.) rather than the V0 REST model. Every method on the events client returns it. See the Event class. TaskRunSummary, CronTrigger, and ScheduledRun, the types returned by the runs, crons, and scheduled clients, are exported from the top level, as are TaskRunDetail, WorkflowRunDetail, TaskMetrics, NotFoundError, the webhook auth models (WebhookBasicAuth, WebhookAPIKeyAuth, WebhookHMACAuth), and the webhook type aliases (WebhookSourceName, WebhookHMACAlgorithm, WebhookHMACEncoding).

from hatchet_sdk import Workflow now resolves to the workflow class returned by hatchet.workflow(), and Standalone, the class returned by the hatchet.task family of decorators, is exported alongside it. The Workflow and WorkflowVersion REST models are no longer exported from the top level; import them from hatchet_sdk.clients.rest.models.

The hatchet_sdk.labels, hatchet_sdk.rate_limit, and hatchet_sdk.utils.priority modules (aliases for hatchet_sdk.types.labels, hatchet_sdk.types.rate_limit, and hatchet_sdk.types.priority) have been removed. Import DesiredWorkerLabel, WorkerLabel, RateLimit, and Priority from hatchet_sdk directly. (RateLimitDuration is no longer exported from hatchet_sdk; pass the duration as a plain string, as described above.)

The hatchet_sdk.types package no longer re-exports DesiredWorkerLabel, WorkerLabelComparator, Priority, RateLimit, and SlotType. Import them from hatchet_sdk.

The PARENT_ID, PARENT_STEP_RUN_ID, CHILD_INDEX, and DESIRED_WORKER_ID members of OTelAttribute have been removed, since the SDK no longer sets those span attributes. Remove parent_id, parent_step_run_id, child_index, and desired_worker_id from OpenTelemetryConfig.excluded_attributes (or HATCHET_CLIENT_OPENTELEMETRY_EXCLUDED_ATTRIBUTES) if you set them, since they now fail validation. The run, bulk run, and schedule spans also no longer set hatchet.namespace or hatchet.key. The child key is still reported as hatchet.child_key.

TriggerWorkflowOptions, ScheduleTriggerWorkflowOptions, RunEventListener, and WorkflowConfig are no longer exported from hatchet_sdk. No public method accepts them now that the options parameters are gone. WorkflowRunTriggerConfig still holds a TriggerWorkflowOptions, so build one with workflow.create_bulk_run_item(...) rather than by hand.

The events client module has moved from hatchet_sdk.clients.events to hatchet_sdk.features.events, alongside the other feature clients. Event and BulkPushEventWithMetadata are still exported from hatchet_sdk, which is the recommended place to import them from.

RunStatus is the only run status type

V1 had two enums with the same five members: RunStatus, used by runs.get_details(), and V1TaskStatus, used by the REST models and the statuses filters. V2 keeps only RunStatus, and it is a string ("QUEUED", "RUNNING", "COMPLETED", "CANCELLED", or "FAILED") rather than an enum. Every status the SDK accepts or returns is a RunStatus, including TaskRunSummary.status, ScheduledRun.workflow_run_status, and the status field on REST models.

# V1
from hatchet_sdk import RunStatus, V1TaskStatus

runs = hatchet.runs.list(statuses=[V1TaskStatus.FAILED])
details = hatchet.runs.get_details(run_id)
is_done = details.status == RunStatus.COMPLETED

# V2
runs = hatchet.runs.list(statuses=["FAILED"])
details = hatchet.runs.get_details(run_id)
is_done = details.status == "COMPLETED"

V1TaskStatus is no longer exported from hatchet_sdk, and the RunStatus.from_proto, RunStatus.from_v1_task_status, and RunStatus.to_v1_task_status conversion methods have been removed, since there is nothing left to convert between.

Scheduled run status filters use RunStatus

scheduled.list and scheduled.aio_list took statuses as a list of ScheduledRunStatus, the engine's V0 run statuses. They now take RunStatus values, matching ScheduledRun.workflow_run_status, plus "SCHEDULED" for schedules that have not triggered a run yet. ScheduledRunStatus is no longer exported from hatchet_sdk.

# V1
from hatchet_sdk import ScheduledRunStatus

hatchet.scheduled.list(statuses=[ScheduledRunStatus.SUCCEEDED, ScheduledRunStatus.SCHEDULED])

# V2
hatchet.scheduled.list(statuses=["COMPLETED", "SCHEDULED"])

"QUEUED" matches runs the engine reports as PENDING or QUEUED, and "COMPLETED" matches SUCCEEDED.

Exceptions renamed to use the Error suffix

NonRetryableException is now NonRetryableError, and RetryAfterException is now RetryAfterError, matching the rest of the SDK's exceptions and the names used by the other Hatchet SDKs. NotFoundException, which the REST client raises for a 404, is now NotFoundError. Their behavior is unchanged.

# V1
from hatchet_sdk import NonRetryableException, NotFoundException, RetryAfterException

raise NonRetryableException("Bad request")
raise RetryAfterException("Rate limited", after=timedelta(seconds=30))

try:
    hatchet.runs.get(run_id)
except NotFoundException:
    ...

# V2
from hatchet_sdk import NonRetryableError, NotFoundError, RetryAfterError

raise NonRetryableError("Bad request")
raise RetryAfterError("Rate limited", after=timedelta(seconds=30))

try:
    hatchet.runs.get(run_id)
except NotFoundError:
    ...

ctx.task_run_errors returns parsed errors

In an on-failure task, ctx.task_run_errors now returns a dict[str, TaskRunError] keyed by task name, the same type ctx.get_task_run_error(task) returns. In V1 the values were serialized error strings. It also no longer logs an error when no upstream task failed, and returns an empty dictionary instead.

# V1
for task_name, serialized_error in ctx.task_run_errors.items():
    print(task_name, serialized_error)

# V2
for task_name, error in ctx.task_run_errors.items():
    print(task_name, error.exc_type, error.exc, error.trace)

Parameter names made consistent

A few parameters were renamed so that the same concept has the same name everywhere. These only affect calls that pass the argument by keyword.

• Options. BulkCancelReplayOpts is now BulkCancelReplayOptions, and the opts parameter on runs.bulk_cancel and runs.bulk_replay is now options.
• Scheduling. The run_at parameter on schedule and aio_schedule is now trigger_at, matching hatchet.scheduled.create(trigger_at=...) and the trigger_at field on scheduled runs.
• Run IDs. Methods that take a workflow run's ID call it workflow_run_id: runs.get_result, runs.get_details, and get_run_ref, get_result, and aio_get_result on a standalone task. Methods that take a task run's ID call it task_run_id, which now includes runs.reset_durable_task. runs.cancel and runs.replay keep run_id, since they accept either kind.
• ID filters. parent_task_external_id is now parent_task_run_id, and triggering_event_external_id is now triggering_event_id, on runs.list, metrics.get_task_metrics, workflow.list_runs, and their aio_ variants. scheduled.bulk_delete takes parent_task_run_id too.

"Step" names renamed to "task"

The remaining public names that used the V0 word "step" now say "task":

• The parent_step_run_id parameter on scheduled.bulk_delete and scheduled.aio_bulk_delete is now parent_task_run_id.
• The span attribute names (hatchet.step_name, hatchet.step_id, hatchet.step_run_id) are unchanged, as is the hatchet.start_step_run span name, so existing dashboards and queries keep working. OTelAttribute is now a string, so exclude these attributes as "step_name", "step_id", and "step_run_id". See enums replaced by strings.
• The ctx_step_run_id context variable in hatchet_sdk.runnables.contextvars, which some custom log filters read, is now ctx_task_run_external_id.
• Worker log messages and error messages refer to tasks instead of steps. Update any log parsing that matched the old wording, such as Waiting Steps.

runs.subscribe_to_stream renamed to runs.aio_subscribe_to_stream

The async generator for subscribing to a run's stream chunks now carries the aio_ prefix, consistent with every other async method in the SDK:

# V1
async for chunk in hatchet.runs.subscribe_to_stream(run_id): ...

# V2
async for chunk in hatchet.runs.aio_subscribe_to_stream(run_id): ...

aio_subscribe_to_stream is deprecated. See run streams.

Waiting for results accepts a timeout

result and aio_result on WorkflowRunRef and TaskRunRef accept an optional timeout (a timedelta) and raise TimeoutError when it elapses:

from datetime import timedelta

ref = workflow.run(input, wait_for_result=False)
result = ref.result(timeout=timedelta(seconds=30))

timeout is the first positional parameter of result. In V1.41 the only parameter was poll_interval. It is now keyword-only and takes a timedelta:

# V1
result = ref.result(2.0)

# V2
result = ref.result(poll_interval=timedelta(seconds=2))

If fetching the run's status fails repeatedly, result and aio_result now raise RuntimeError instead of ValueError. A cancelled run still raises ValueError.

Worker startup failures raise and start() returns on shutdown

Worker.register_workflow previously called sys.exit(1) if registering the workflow with the engine failed. It now raises the underlying exception, so registration failures can be caught and handled (and no longer kill the host process, e.g. a web server embedding a worker). The same applies when the action listener process fails to start: worker.start() raises the original exception instead of calling sys.exit(1).

worker.start() now returns after a graceful shutdown (on SIGTERM or SIGINT, or after terminate_worker_after_num_tasks tasks) instead of calling sys.exit(0). Code after worker.start() now runs, so make sure it does not keep the process alive if you rely on the worker exiting. A forced shutdown (a second signal, SIGQUIT, or force_shutdown_on_shutdown_signal) still exits with status 1.

The handle_kill parameter on the Worker constructor has been removed. handle_kill=False was the way to stop the exit in V1, and it is now the only behavior. The Worker constructor's parameters after name are keyword-only.

Configuration changes

• The HealthcheckConfig.event_loop_block_threshold_seconds field has been renamed to event_loop_block_threshold (it is a timedelta; numeric values are interpreted as seconds). The corresponding environment variable is now HATCHET_CLIENT_WORKER_HEALTHCHECK_EVENT_LOOP_BLOCK_THRESHOLD.
• ClientTLSConfig.strategy is now typed as Literal["tls", "mtls", "none"] instead of str, so invalid values fail at construction time instead of being silently treated as insecure.
• ClientConfig.sync_result_poll_interval is a timedelta instead of a float. HATCHET_CLIENT_SYNC_RESULT_POLL_INTERVAL still takes a number of seconds.
• ClientConfig.listener_v2_timeout has been renamed to action_listener_timeout and is a timedelta. The environment variable is now HATCHET_CLIENT_ACTION_LISTENER_TIMEOUT, in seconds.
• EmbeddedHatchetConfig.ready_timeout_seconds has been renamed to ready_timeout and is a timedelta. The environment variable is now HATCHET_CLIENT_EMBEDDED_READY_TIMEOUT, in seconds.
• OpenTelemetryConfig.include_task_name_in_start_step_run_span_name has been renamed to include_task_name_in_span_name. The environment variable is now HATCHET_CLIENT_OPENTELEMETRY_INCLUDE_TASK_NAME_IN_SPAN_NAME.
• The DEFAULT_SYNC_RESULT_POLL_INTERVAL_SECONDS and DEFAULT_READY_TIMEOUT_SECONDS constants in hatchet_sdk.config have been renamed to DEFAULT_SYNC_RESULT_POLL_INTERVAL and DEFAULT_EMBEDDED_READY_TIMEOUT, and are timedelta values.
• HATCHET_CLIENT_TENACITY_* environment variables set after hatchet_sdk is imported are now read when a ClientConfig is created. In V1 they were read once, at import time.

The old environment variable names are no longer read, so rename them along with the fields.

workflow.id replaced by get_id()

The workflow.id property made an API call the first time it was read. It has been replaced by methods, so the call is explicit and has an async version:

# V1
workflow_id = workflow.id

# V2
workflow_id = workflow.get_id()
workflow_id = await workflow.aio_get_id()

Both fetch the ID on the first call and cache it. They raise ValueError if the workflow has not been registered.

The hatchet.workflows methods still take the workflow's ID, so pass workflow.get_id() to them, for example hatchet.workflows.get(workflow.get_id()).

FailedTaskRunExceptionGroup is an ExceptionGroup

FailedTaskRunExceptionGroup, which run() and result() raise when a run fails, now subclasses ExceptionGroup[TaskRunError] instead of ValueError. except ValueError no longer catches it. Catch it by name, or use except* to handle the task errors inside it:

# V1
try:
    workflow.run(input)
except ValueError as e:
    for error in e.exceptions:
        print(error.exc_type, error.exc)

# V2
from hatchet_sdk import TaskRunError

try:
    workflow.run(input)
except* TaskRunError as group:
    for error in group.exceptions:
        print(error.exc_type, error.exc)

.exceptions is a tuple instead of a list. When a run fails without reporting any task errors, it holds a single TaskRunError carrying the failure message instead of being empty. str() is unchanged. .message now holds the same text as str(), the run's failure message followed by each task error, so pytest.raises(FailedTaskRunExceptionGroup, match=...) still matches on a task's error message. TaskRunError now sets args, so both can be pickled.

ctx.refresh_timeout and ctx.put_stream raise on failure

ctx.refresh_timeout, ctx.aio_refresh_timeout, ctx.put_stream, and ctx.aio_put_stream used to log an error and return if the request to the engine failed. They now raise, so an unhandled failure fails the task, which is then retried according to its retry policy. To keep the V1 behavior, catch the error yourself:

try:
    await ctx.aio_put_stream(chunk)
except Exception:
    logger.exception("failed to send stream chunk")

ctx.put_stream and runs.aio_subscribe_to_stream are deprecated

ctx.put_stream, ctx.aio_put_stream, and hatchet.runs.aio_subscribe_to_stream still work, but emit a DeprecationWarning and will be removed in a future release. Use durable streams (hatchet.streams) instead. A durable stream is a topic that stores its messages independently of any run, so a reader that connects late, or reconnects, can resume from a cursor instead of missing chunks:

# Deprecated
@hatchet.task()
async def generate(input: None, ctx: Context) -> None:
    for chunk in chunks:
        await ctx.aio_put_stream(chunk)

ref = await generate.aio_run(wait_for_result=False)
async for chunk in hatchet.runs.aio_subscribe_to_stream(ref.workflow_run_id):
    print(chunk)

# V2
@hatchet.task()
async def generate(input: None, ctx: Context) -> None:
    for chunk in chunks:
        await hatchet.streams.aio_publish(f"generate-{ctx.workflow_run_id}", chunk)

ref = await generate.aio_run(wait_for_result=False)
async for event in hatchet.streams.aio_subscribe(f"generate-{ref.workflow_run_id}"):
    print(event.payload)

on_failure_task and on_success_task use the workflow's task_defaults

In V1, on_failure_task and on_success_task used fixed defaults (a 5 minute schedule timeout, a 60 second execution timeout, and 0 retries) and ignored the workflow's task_defaults, along with its backoff_factor and backoff_max. They now resolve each setting the same way as task does: the explicit argument, then task_defaults, then the built-in default.

workflow = hatchet.workflow(
    name="MyWorkflow",
    task_defaults=TaskDefaults(retries=3, execution_timeout=timedelta(minutes=10)),
)

@workflow.on_failure_task()
def on_failure(input: MyInput, ctx: Context) -> None: ...
# V1: 0 retries and a 60 second timeout. V2: 3 retries and a 10 minute timeout.

To keep the V1 behavior, pass the values explicitly, for example @workflow.on_failure_task(retries=0, execution_timeout=timedelta(seconds=60)).

TaskDefaults rejects unknown fields

TaskDefaults now raises a ValidationError for a field it does not define, instead of ignoring it. This catches the removed priority field and the renamed backoff_max_seconds field (now backoff_max, a timedelta), including when they are passed by unpacking a dict.

ParentCondition has a stable default key

The default readable_data_key of a ParentCondition is the parent task's name, followed by : and the expression when one is set. In V1 it also included the time the condition was created, so the workflow definition changed every time a worker started.

Since the key no longer differs between conditions, two ParentConditions on the same task with the same parent and expression now collide, and registering the workflow raises ValueError: Conditions must have unique readable data keys. Pass a distinct readable_data_key to each one.

Feature client renames and removals

• workers.update and workers.aio_update have been removed. Pausing was the only update they supported, so use workers.pause(worker_id) and workers.unpause(worker_id), or their aio_ variants.

• filters.update and filters.aio_update take the fields to change as keyword arguments instead of a V1UpdateFilterRequest. Fields left as None are not changed:

  # V1
  hatchet.filters.update(filter_id, V1UpdateFilterRequest(expression="input.value > 20"))
  
  # V2
  hatchet.filters.update(filter_id, expression="input.value > 20")

• streams.publish and streams.subscribe have been renamed to streams.aio_publish and streams.aio_subscribe, since both are async.

• runs.bulk_cancel_by_filters_with_pagination and runs.bulk_replay_by_filters_with_pagination, and their aio_ variants, have been renamed to bulk_cancel_by_filters_in_chunks and bulk_replay_by_filters_in_chunks. Their sleep_time parameter (an integer number of seconds) is now delay_between_chunks, a timedelta.

• scheduled.bulk_delete and scheduled.aio_bulk_delete no longer accept statuses. The API has no status filter for bulk deletes, so it was ignored.

• ctx.trigger_data has been removed. Use ctx.filter_payload, which holds its filter_payload entry.

• ctx.pause_rate_limit and ctx.aio_pause_rate_limit have been removed. Use hatchet.rate_limits.pause and hatchet.rate_limits.aio_pause.

Webhook auth models renamed

webhooks.create and webhooks.aio_create take the SDK's own auth models, WebhookBasicAuth, WebhookAPIKeyAuth, and WebhookHMACAuth, instead of the V1WebhookBasicAuth, V1WebhookAPIKeyAuth, and V1WebhookHMACAuth REST models. Their fields are snake_case, and the HMAC algorithm and encoding are strings. source_name is a string too:

# V1
from hatchet_sdk import (
    V1WebhookAPIKeyAuth,
    V1WebhookHMACAlgorithm,
    V1WebhookHMACAuth,
    V1WebhookHMACEncoding,
    V1WebhookSourceName,
)

hatchet.webhooks.create(
    V1WebhookSourceName.GITHUB,
    "github-push",
    event_key_expression="'github:push'",
    auth=V1WebhookHMACAuth(
        algorithm=V1WebhookHMACAlgorithm.SHA256,
        encoding=V1WebhookHMACEncoding.HEX,
        signatureHeaderName="X-Hub-Signature-256",
        signingSecret=secret,
    ),
)
api_key_auth = V1WebhookAPIKeyAuth(headerName="X-API-Key", apiKey=api_key)

# V2
from hatchet_sdk import WebhookAPIKeyAuth, WebhookHMACAuth

hatchet.webhooks.create(
    "GITHUB",
    "github-push",
    event_key_expression="'github:push'",
    auth=WebhookHMACAuth(
        algorithm="SHA256",
        encoding="HEX",
        signature_header_name="X-Hub-Signature-256",
        signing_secret=secret,
    ),
)
api_key_auth = WebhookAPIKeyAuth(header_name="X-API-Key", api_key=api_key)

The source_names filter on webhooks.list takes the same strings. WebhookSourceName, WebhookHMACAlgorithm, and WebhookHMACEncoding are exported from hatchet_sdk as type aliases for annotations.

mock_run takes typed errors and filter_payload

V1.46 added task_run_errors and trigger_data to mock_run and aio_mock_run. In V2 they match the context properties they set:

• task_run_errors takes a dict[str, TaskRunError], the type ctx.task_run_errors returns, instead of serialized error strings.
• trigger_data has been replaced by filter_payload, which sets ctx.filter_payload, since ctx.trigger_data has been removed.

# V1
my_on_failure_task.mock_run(
    task_run_errors={"upstream": "RuntimeError: boom"},
    trigger_data={"filter_payload": {"tier": "gold"}},
)

# V2
from hatchet_sdk import TaskRunError

my_on_failure_task.mock_run(
    task_run_errors={
        "upstream": TaskRunError(
            exc="boom", exc_type="RuntimeError", trace="", task_run_external_id=None
        )
    },
    filter_payload={"tier": "gold"},
)

Event class field and property changes

V1 had two event classes: Event, returned by events.push and events.bulk_push, and the V1Event REST model, returned by events.get and events.list. In V2 every method on the events client returns Event, and V1Event is no longer exported from hatchet_sdk.

The event's ID is now event.id everywhere. It replaces event.event_id on pushed events and event.metadata.id on events read back with get or list. payload and additional_metadata are now dictionaries rather than JSON strings, and additional_metadata is an empty dictionary when the event has none:

# V1
event = hatchet.events.push("user:create", {"user_id": "1"})
payload = json.loads(event.payload)

listed = hatchet.events.list(event_ids=[event.event_id])
listed_id = listed[0].metadata.id

# V2
event = hatchet.events.push("user:create", {"user_id": "1"})
payload = event.payload

listed = hatchet.events.list(event_ids=[event.id])
listed_id = listed[0].id

workflow_run_summary and triggered_runs are only set on events read back with get or list. They are None on the result of a push, since the runs have not been created yet.

The event_timestamp field and its deprecated camelCase aliases (eventTimestamp, additionalMetadata) have also been removed. Use the snake_case fields directly: