Migrating from Temporal to Hatchet

Temporal and Hatchet are both platforms which allow developers to write durable workflows. These are workflows whose intermediate state is persisted, which means that if your worker crashes or fails halfway through a workflow, it automatically picks up where it left off. This can be particularly useful for AI agents, long-running jobs, or business-critical workflows.

This guide assumes you have already decided to migrate a project from Temporal to Hatchet and want to understand what code and configuration need to change. Each section starts with a common Temporal pattern, then shows what replaces it in Hatchet and what to watch out for.

Conceptual similarities and differences

Let's start with what's the same. The Temporal and Hatchet long-lived worker models are very similar. They both utilize client-instantiated, long-lived gRPC connections to connect to the Temporal or Hatchet engine, respectively. They also both implement primarily push-based scheduling: work is assigned from the engine to the worker, not the other way around.

However, there are a few important conceptual differences when you decide to migrate to Hatchet. In Temporal, everything operates as either a workflow or an activity. The equivalent concepts in Hatchet are durable tasks and tasks:

• A task is ordinary code. It runs on a worker, retries on failure, and has no determinism constraints. This is what most Temporal activities and most short Temporal workflows become. It is @hatchet.task() in Python, hatchet.task() in TypeScript, client.NewStandaloneTask in Go, and HATCHET.task in Ruby.
• A durable task is the durable-execution primitive. It checkpoints every time it waits or spawns a child, and it carries the same determinism rules a Temporal workflow does. It is @hatchet.durable_task(), hatchet.durableTask(), client.NewStandaloneDurableTask, and HATCHET.durable_task respectively.

Second, workers in Hatchet have a built-in notion of slots, instead of per-activity or per-workflow rate limits. This makes it very easy to reason about the amount of work that a worker is able to accept.

Migration Steps

Step 1: Dependencies and connection

Install the Hatchet SDK:

pip install hatchet-sdk

Temporal clients are configured with a server address, a namespace, and, on Temporal Cloud, mTLS certificates or an API key. Hatchet replaces all of that with a single token, read from the environment by every SDK:

export HATCHET_CLIENT_TOKEN="your-token-here"

The token encodes the tenant and the engine address, so there is no separate namespace or endpoint setting for Hatchet Cloud. For self-hosted deployments you may need environment-specific host and TLS settings.

Temporal and Hatchet can run side by side during a migration. They share no runtime, so each workflow must be moved as a unit: once a workflow runs on Hatchet, update the callers that start it.

Step 2: Replace the client and worker bootstrap

Temporal separates the client from the worker, and routes work with a task queue name:

from temporalio.client import Client
from temporalio.worker import Worker

client = await Client.connect("localhost:7233", namespace="default")

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[ProcessOrderWorkflow],
    activities=[validate_order, charge_order, fulfill_order],
)
await worker.run()

In Hatchet, one client object is both the client and the registry, and workers are named rather than addressed through a queue:

from hatchet_sdk import Hatchethatchet = Hatchet()

from hatchet_sdk import Hatchet

hatchet = Hatchet()

def main() -> None:    worker = hatchet.worker(        "order-worker",        slots=10,        workflows=[validate_order, charge_order, fulfill_order, process_order],    )    worker.start()

def main() -> None:
    worker = hatchet.worker(
        "order-worker",
        slots=10,
        workflows=[validate_order, charge_order, fulfill_order, process_order],
    )
    worker.start()

Two differences worth noting:

• You register both durable tasks and tasks in the same workflows argument
• slots bounds how many task runs this worker will accept at once. Temporal's nearest equivalents are its worker-side executor limits (max_concurrent_activity_task_executions and friends); in Hatchet slot control is a first-class scheduling input, not just a worker-side limit. See workers.

Routing that you would have done with multiple task queues is done with separate workers and, when you need finer control, worker affinity. Note that each separate task definition in Hatchet get its own queue.

Step 3: Convert activities to tasks

A Temporal activity:

from temporalio import activity


@activity.defn
async def charge_order(order_id: str) -> bool:
    return await payments.charge(order_id)

becomes a Hatchet task. Inputs are a single structured value rather than positional arguments, and every task receives a context:

class OrderInput(BaseModel):    order_id: strclass ChargeOutput(BaseModel):    charged: bool    charge_id: str@hatchet.task(name="charge-order", input_validator=OrderInput)async def charge_order(input: OrderInput, ctx: Context) -> ChargeOutput:    charge_id = await submit_charge(input.order_id)    return ChargeOutput(charged=True, charge_id=charge_id)

class OrderInput(BaseModel):
    order_id: str


class ChargeOutput(BaseModel):
    charged: bool
    charge_id: str


@hatchet.task(name="charge-order", input_validator=OrderInput)
async def charge_order(input: OrderInput, ctx: Context) -> ChargeOutput:
    charge_id = await submit_charge(input.order_id)

    return ChargeOutput(charged=True, charge_id=charge_id)

Note that a Hatchet task is directly runnable. Running charge_order with an order id is a valid, retryable, observable unit of work with no workflow wrapper around it. In Temporal an activity cannot be invoked on its own; it needs a workflow to orchestrate it, which is why Temporal projects accumulate single-activity workflows. Those wrappers can eventually be deleted, though not yet: migrate them in step 4 like any other workflow, then collapse them in step 13.

Serialization moves from Temporal's converters (JSON by default, pickle or custom converters if configured) to JSON. Any workflow argument that relied on a custom data converter needs a JSON-serializable representation. The shape of the input type differs by SDK: Python uses Pydantic models, TypeScript uses TypeScript types (with optional Zod validation via inputValidator), Go uses structs with json tags, and Ruby passes plain string-keyed hashes.

Step 4: Convert workflows to durable tasks

Next, we'll convert every Temporal workflow to a Hatchet durable task. The workflow body becomes the durable task body, and each execute_activity(...) call becomes a call to the Hatchet task you defined in step 3.

A Temporal workflow that runs three activities in order:

@workflow.defn
class ProcessOrderWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> dict:
        valid = await workflow.execute_activity(
            validate_order, order_id, start_to_close_timeout=timedelta(seconds=30)
        )
        charged = await workflow.execute_activity(
            charge_order, order_id, start_to_close_timeout=timedelta(seconds=30)
        )
        return await workflow.execute_activity(
            fulfill_order, order_id, start_to_close_timeout=timedelta(seconds=30)
        )

becomes a durable task that runs three tasks in order:

@hatchet.durable_task(name="ProcessOrder", input_validator=OrderInput)async def process_order(input: OrderInput, ctx: DurableContext) -> FulfillOutput:    await validate_order.aio_run(input)    await charge_order.aio_run(input)    return await fulfill_order.aio_run(input)

@hatchet.durable_task(name="ProcessOrder", input_validator=OrderInput)
async def process_order(input: OrderInput, ctx: DurableContext) -> FulfillOutput:
    await validate_order.aio_run(input)
    await charge_order.aio_run(input)

    return await fulfill_order.aio_run(input)

Every guarantee of the Temporal workflow is preserved in Hatchet. The durable task has a run id and a history, it survives a worker crash, and it resumes from its last checkpoint rather than from the top. workflow.execute_activity(fn, arg, ...) becomes fn.aio_run(input) (.run(...) in TypeScript, Go and Ruby), results come back as ordinary return values rather than futures, and the per-call retry and timeout options move onto the task definition instead, which is step 6.

Workflows that wait, rather than only calling activities, translate the same way. A workflow that sends an email, sleeps for three days, then sends another:

@hatchet.durable_task(    name="OnboardingFlow",    input_validator=OnboardingInput,    # The timeout has to cover the whole wall-clock span of the run, sleeps included.    execution_timeout=timedelta(days=7),)async def onboarding_flow(input: OnboardingInput, ctx: DurableContext) -> None:    await send_welcome_email.aio_run(input)    await ctx.aio_sleep_for(timedelta(days=3))    await send_followup_email.aio_run(input)

@hatchet.durable_task(
    name="OnboardingFlow",
    input_validator=OnboardingInput,
    # The timeout has to cover the whole wall-clock span of the run, sleeps included.
    execution_timeout=timedelta(days=7),
)
async def onboarding_flow(input: OnboardingInput, ctx: DurableContext) -> None:
    await send_welcome_email.aio_run(input)

    await ctx.aio_sleep_for(timedelta(days=3))

    await send_followup_email.aio_run(input)

The determinism rules here are the same: a durable task may only wait on the durable context or spawn children, and the code between checkpoints must be reproducible on replay. Anything that touches the outside world belongs in a child task. See durable tasks.

Note that a durable task's execution timeout has to cover the whole wall-clock span of the run, including its sleeps, so set it accordingly.

Step 5: Invoke work

The mapping is the same in every SDK; names below use the Python spelling.

Fire and forget, then collect the result later:

async def trigger_process_order(order_id: str) -> FulfillOutput:    ref = await process_order.aio_run(        OrderInput(order_id=order_id),        wait_for_result=False,    )    # Available immediately. Store it if you need to reattach to the run later.    print(ref.workflow_run_id)    return await ref.aio_result()

async def trigger_process_order(order_id: str) -> FulfillOutput:
    ref = await process_order.aio_run(
        OrderInput(order_id=order_id),
        wait_for_result=False,
    )

    # Available immediately. Store it if you need to reattach to the run later.
    print(ref.workflow_run_id)

    return await ref.aio_result()

Note what is not required: there is no workflow id and no task queue at the call site. Deduplication that you would have achieved with a Temporal workflow id reuse policy is expressed with an idempotency key instead.

Step 6: Retries and timeouts

Temporal attaches a RetryPolicy to activity and workflow options. Hatchet attaches retry settings to the task definition:

@hatchet.task(    name="charge-order-with-retries",    input_validator=OrderInput,    retries=10,    backoff_factor=2.0,    backoff_max_seconds=10,    execution_timeout=timedelta(seconds=30),    schedule_timeout=timedelta(minutes=10),)async def charge_order_with_retries(input: OrderInput, ctx: Context) -> ChargeOutput:    # Raising `NonRetryableException` here would stop Hatchet from retrying at all.    if ctx.retry_count < 2:        raise RuntimeError(f"payment provider unavailable for {input.order_id}")    return ChargeOutput(charged=True, charge_id=await submit_charge(input.order_id))

@hatchet.task(
    name="charge-order-with-retries",
    input_validator=OrderInput,
    retries=10,
    backoff_factor=2.0,
    backoff_max_seconds=10,
    execution_timeout=timedelta(seconds=30),
    schedule_timeout=timedelta(minutes=10),
)
async def charge_order_with_retries(input: OrderInput, ctx: Context) -> ChargeOutput:
    # Raising `NonRetryableException` here would stop Hatchet from retrying at all.
    if ctx.retry_count < 2:
        raise RuntimeError(f"payment provider unavailable for {input.order_id}")

    return ChargeOutput(charged=True, charge_id=await submit_charge(input.order_id))

The examples give this task its own name, charge-order-with-retries, only so that it can be registered alongside the plain charge-order task from step 3. In your own code these options go straight onto the existing task and the name does not change.

The concept mapping is the same in every SDK. Names below use the Python spelling; the code above shows each SDK's exact form.

The non-retryable error type is NonRetryableException in Python (from hatchet_sdk.exceptions), NonRetryableError in TypeScript, worker.NewNonRetryableError in Go, and Hatchet::NonRetryableError in Ruby. The retry count is ctx.retry_count in Python and Ruby, ctx.retryCount() in TypeScript, and ctx.RetryCount() in Go.

Defaults for every task in a workflow go in the workflow's task defaults (task_defaults in Python and Ruby, taskDefaults in TypeScript). See retry policies and timeouts.

Step 7: Timers and sleeps

A durable sleep inside a Temporal workflow becomes a durable sleep on the Hatchet durable context:

Temporal:

await asyncio.sleep(60 * 60 * 24)

Hatchet:

await ctx.aio_sleep_for(timedelta(days=1))

await ctx.aio_sleep_for(timedelta(days=1))

While a Hatchet durable task is sleeping it is evicted and its worker slot is released, so a million sleeping runs cost no worker capacity. See durable sleep.

For a task (not a durable task) that should simply start later, create a scheduled run rather than sleeping. There is no reason to hold a durable task open for a fixed future start.

Step 8: Signals, queries, and updates

Temporal signals become Hatchet events, which are pushed to the tenant rather than to a specific run:

Temporal:

handle = client.get_workflow_handle(workflow_id)
await handle.signal(ApprovalWorkflow.approve, "approved")

Hatchet:

async def grant_approval(correlation_id: str) -> None:    await hatchet.event.aio_push(        "approval:granted",        {"correlation_id": correlation_id},    )

async def grant_approval(correlation_id: str) -> None:
    await hatchet.event.aio_push(
        "approval:granted",
        {"correlation_id": correlation_id},
    )

The waiting side uses a durable event wait, with a CEL expression to select the event that belongs to this run:

class ApprovalInput(BaseModel):    order_id: str    # Generate this yourself, and keep it CEL-safe: the expression below is compiled    # as-is, so a value containing a quote would fail to compile and the run would    # wait until it timed out.    correlation_id: strclass ApprovalOutput(BaseModel):    approved: bool@hatchet.durable_task(    name="ApprovalFlow",    input_validator=ApprovalInput,    execution_timeout=timedelta(minutes=10),)async def approval_flow(input: ApprovalInput, ctx: DurableContext) -> ApprovalOutput:    await ctx.aio_wait_for_event(        "approval:granted",        f"input.correlation_id == '{input.correlation_id}'",    )    return ApprovalOutput(approved=True)

class ApprovalInput(BaseModel):
    order_id: str
    # Generate this yourself, and keep it CEL-safe: the expression below is compiled
    # as-is, so a value containing a quote would fail to compile and the run would
    # wait until it timed out.
    correlation_id: str


class ApprovalOutput(BaseModel):
    approved: bool


@hatchet.durable_task(
    name="ApprovalFlow",
    input_validator=ApprovalInput,
    execution_timeout=timedelta(minutes=10),
)
async def approval_flow(input: ApprovalInput, ctx: DurableContext) -> ApprovalOutput:
    await ctx.aio_wait_for_event(
        "approval:granted",
        f"input.correlation_id == '{input.correlation_id}'",
    )

    return ApprovalOutput(approved=True)

Note that anything can emit a Hatchet event, and any number of durable tasks can wait on the same event. The filter expression is doing the work that the workflow id used to do in Temporal, so give runs a stable, CEL-safe correlation id in their input and filter on that.

Three related mappings:

• A wait-for-condition inside a workflow (workflow.wait_condition, wf.condition, workflow.Await) is usually a wait condition on a DAG task instead: a sleep condition, a user-event condition, or a parent condition, composed with an or-group. That moves the wait out of your code and into the workflow definition, where the engine can see it. See conditions.
• Queries have no direct equivalent, because Hatchet does not run your code to answer a question about a run. Run inputs, outputs, and history are already queryable through the API and dashboard. For live progress, stream it from the task (ctx.aio_put_stream in Python, ctx.putStream in TypeScript, ctx.PutStream in Go, ctx.put_stream in Ruby) and consume it from the caller. See streaming.
• Updates, a synchronous and validated mutation of running workflow state, have no equivalent. Model this as an event the durable task waits on, plus a child task that performs the mutation and returns a result.

Step 9: Child workflows and fan-out

Convert workflows that spawn one child per item into a durable task that runs one child task per item:

Temporal:

results = await asyncio.gather(*[
    workflow.execute_child_workflow(ProcessItem.run, item) for item in items
])

Hatchet:

class ItemInput(BaseModel):    item_id: strclass ItemOutput(BaseModel):    item_id: str    result: strclass FanOutInput(BaseModel):    item_ids: list[str]class FanOutOutput(BaseModel):    results: list[ItemOutput]@hatchet.task(name="process-item", input_validator=ItemInput)async def process_item(input: ItemInput, ctx: Context) -> ItemOutput:    return ItemOutput(item_id=input.item_id, result=await handle_item(input.item_id))@hatchet.durable_task(name="ProcessItems", input_validator=FanOutInput)async def process_items(input: FanOutInput, ctx: DurableContext) -> FanOutOutput:    results = await asyncio.gather(        *[            process_item.aio_run(ItemInput(item_id=item_id))            for item_id in input.item_ids        ]    )    return FanOutOutput(results=results)

class ItemInput(BaseModel):
    item_id: str


class ItemOutput(BaseModel):
    item_id: str
    result: str


class FanOutInput(BaseModel):
    item_ids: list[str]


class FanOutOutput(BaseModel):
    results: list[ItemOutput]


@hatchet.task(name="process-item", input_validator=ItemInput)
async def process_item(input: ItemInput, ctx: Context) -> ItemOutput:
    return ItemOutput(item_id=input.item_id, result=await handle_item(input.item_id))


@hatchet.durable_task(name="ProcessItems", input_validator=FanOutInput)
async def process_items(input: FanOutInput, ctx: DurableContext) -> FanOutOutput:
    results = await asyncio.gather(
        *[
            process_item.aio_run(ItemInput(item_id=item_id))
            for item_id in input.item_ids
        ]
    )

    return FanOutOutput(results=results)

Spawning from a durable task checkpoints each child, so a crash mid-fan-out resumes without re-running completed children. See child spawning.

Hatchet places no per-run cap on the number of children spawned from a durable run. Temporal enforces a limit on pending child executions and pending activities per run (51,200 by default, tunable in server dynamic config), which is the constraint behind most Temporal "continue-as-new to keep the history small" batching code. That batching code can usually be deleted. For large fan-outs, see bulk run.

Step 10: Schedules and crons

A Temporal schedule created through the schedule client is usually a declaration on the task itself:

class ReportInput(BaseModel):    kind: strclass ReportOutput(BaseModel):    kind: str    rows: int@hatchet.task(    name="weekly-report",    input_validator=ReportInput,    on_crons=["0 9 * * 1"],    cron_input=ReportInput(kind="weekly"),)async def weekly_report(input: ReportInput, ctx: Context) -> ReportOutput:    return ReportOutput(kind=input.kind, rows=await count_report_rows(input.kind))

class ReportInput(BaseModel):
    kind: str


class ReportOutput(BaseModel):
    kind: str
    rows: int


@hatchet.task(
    name="weekly-report",
    input_validator=ReportInput,
    on_crons=["0 9 * * 1"],
    cron_input=ReportInput(kind="weekly"),
)
async def weekly_report(input: ReportInput, ctx: Context) -> ReportOutput:
    return ReportOutput(kind=input.kind, rows=await count_report_rows(input.kind))

For schedules created at runtime (the equivalent of programmatic Temporal schedules), use the API:

async def create_schedules(customer_id: str) -> tuple[str, str]:    cron = await hatchet.cron.aio_create(        workflow_name=weekly_report.name,        cron_name=f"weekly-report-{customer_id}",        expression="0 9 * * 1",        input={"kind": "weekly"},        additional_metadata={"customer_id": customer_id},    )    scheduled = await hatchet.scheduled.aio_create(        workflow_name=weekly_report.name,        trigger_at=datetime.now(tz=timezone.utc) + timedelta(days=1),        input={"kind": "weekly"},        additional_metadata={"customer_id": customer_id},    )    return cron.metadata.id, scheduled.metadata.id

async def create_schedules(customer_id: str) -> tuple[str, str]:
    cron = await hatchet.cron.aio_create(
        workflow_name=weekly_report.name,
        cron_name=f"weekly-report-{customer_id}",
        expression="0 9 * * 1",
        input={"kind": "weekly"},
        additional_metadata={"customer_id": customer_id},
    )

    scheduled = await hatchet.scheduled.aio_create(
        workflow_name=weekly_report.name,
        trigger_at=datetime.now(tz=timezone.utc) + timedelta(days=1),
        input={"kind": "weekly"},
        additional_metadata={"customer_id": customer_id},
    )

    return cron.metadata.id, scheduled.metadata.id

See cron runs and scheduled runs.

Step 11: Versioning and determinism

Temporal gives you patching APIs (workflow.patched / GetVersion and their per-SDK equivalents) to branch on workflow version, so a deployment can change the code path of workflows that are already running. Hatchet has no equivalent: for breaking changes to durable tasks, we recommend deploying a new durable task definition and letting the old work drain.

Step 12: Flow control

Several things that Temporal projects build by hand (partitioned task queues, worker-side semaphores, external rate limiters in front of activities) are engine features in Hatchet:

class SyncInput(BaseModel):    customer_id: strclass SyncOutput(BaseModel):    records_synced: int# One in-flight run per customer, newest cancels the oldest.sync_customer = hatchet.workflow(    name="SyncCustomer",    input_validator=SyncInput,    concurrency=ConcurrencyExpression(        expression="input.customer_id",        max_runs=1,        limit_strategy=ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,    ),)@sync_customer.task()async def sync(input: SyncInput, ctx: Context) -> SyncOutput:    return SyncOutput(records_synced=await sync_customer_data(input.customer_id))class PromptInput(BaseModel):    prompt: strclass ModelOutput(BaseModel):    completion: str# A global budget shared by every worker, not per-process. The static key has to be# declared once with `hatchet.rate_limits.put` before a task can consume it.@hatchet.task(    name="call-model",    input_validator=PromptInput,    rate_limits=[RateLimit(static_key="openai", units=1)],)async def call_model(input: PromptInput, ctx: Context) -> ModelOutput:    return ModelOutput(completion=await complete_prompt(input.prompt))

class SyncInput(BaseModel):
    customer_id: str


class SyncOutput(BaseModel):
    records_synced: int


# One in-flight run per customer, newest cancels the oldest.
sync_customer = hatchet.workflow(
    name="SyncCustomer",
    input_validator=SyncInput,
    concurrency=ConcurrencyExpression(
        expression="input.customer_id",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,
    ),
)


@sync_customer.task()
async def sync(input: SyncInput, ctx: Context) -> SyncOutput:
    return SyncOutput(records_synced=await sync_customer_data(input.customer_id))


class PromptInput(BaseModel):
    prompt: str


class ModelOutput(BaseModel):
    completion: str


# A global budget shared by every worker, not per-process. The static key has to be
# declared once with `hatchet.rate_limits.put` before a task can consume it.
@hatchet.task(
    name="call-model",
    input_validator=PromptInput,
    rate_limits=[RateLimit(static_key="openai", units=1)],
)
async def call_model(input: PromptInput, ctx: Context) -> ModelOutput:
    return ModelOutput(completion=await complete_prompt(input.prompt))

See concurrency, rate limits, priority, and worker slots. Auditing what your Temporal deployment does with task queues and external limiters usually deletes a meaningful amount of infrastructure.

Step 13: Simplification

Now that workflows have been converted to durable tasks, you can simplify them in two ways. Both are optional. Neither changes what your code does, so nothing here has to happen on the migration's critical path.

A fixed sequence of task calls becomes a DAG

The ProcessOrder durable task from step 4 never waits and never decides at runtime what to run next. It calls three tasks in a fixed order. That shape is a DAG, and Hatchet can express it declaratively with parents=[...], which deletes the orchestration code entirely:

order_workflow = hatchet.workflow(name="ProcessOrderDag", input_validator=OrderInput)@order_workflow.task(execution_timeout=timedelta(seconds=30))async def validate(input: OrderInput, ctx: Context) -> ValidateOutput:    return ValidateOutput(valid=await check_inventory(input.order_id))@order_workflow.task(parents=[validate], execution_timeout=timedelta(seconds=30))async def charge(input: OrderInput, ctx: Context) -> ChargeOutput:    validated = ctx.task_output(validate)    if not validated.valid:        raise ValueError(f"order {input.order_id} failed validation")    return ChargeOutput(charged=True, charge_id=await submit_charge(input.order_id))@order_workflow.task(parents=[charge], execution_timeout=timedelta(seconds=30))async def fulfill(input: OrderInput, ctx: Context) -> FulfillOutput:    charged = ctx.task_output(charge)    return FulfillOutput(        order_id=input.order_id,        tracking_number=await ship(charged.charge_id),    )

order_workflow = hatchet.workflow(name="ProcessOrderDag", input_validator=OrderInput)


@order_workflow.task(execution_timeout=timedelta(seconds=30))
async def validate(input: OrderInput, ctx: Context) -> ValidateOutput:
    return ValidateOutput(valid=await check_inventory(input.order_id))


@order_workflow.task(parents=[validate], execution_timeout=timedelta(seconds=30))
async def charge(input: OrderInput, ctx: Context) -> ChargeOutput:
    validated = ctx.task_output(validate)

    if not validated.valid:
        raise ValueError(f"order {input.order_id} failed validation")

    return ChargeOutput(charged=True, charge_id=await submit_charge(input.order_id))


@order_workflow.task(parents=[charge], execution_timeout=timedelta(seconds=30))
async def fulfill(input: OrderInput, ctx: Context) -> FulfillOutput:
    charged = ctx.task_output(charge)

    return FulfillOutput(
        order_id=input.order_id,
        tracking_number=await ship(charged.charge_id),
    )

The examples register the DAG as ProcessOrderDag so that it and the ProcessOrder durable task it replaces can both be served while the cutover is in progress. When you make the switch for real, give the DAG the name the durable task had and delete the durable task, so callers do not have to change.

Upstream results are read off the context (ctx.task_output in Python and Ruby, ctx.parentOutput in TypeScript, ctx.ParentOutput in Go) rather than returned through local variables. Tasks with no parents run in parallel; that is the replacement for whatever you used to run activities concurrently inside a workflow. See DAGs.

There is no determinism constraint on a DAG task. Each one is ordinary code that can read a database, call an API, or use the clock.

That is worth doing for three reasons. The orchestration code is gone, leaving only task bodies and their parent declarations. The parallelism is free: you declare the graph, and Hatchet runs whatever it does not order concurrently, with no asyncio.gather, Promise.all or goroutines of your own. And the dashboard renders the graph before the run starts, so a stuck run points at the node it is stuck on rather than at a flat list of events.

A wrapper around a single task becomes that task

A Temporal workflow that existed only to invoke one activity became, in step 4, a durable task that does nothing but call one task. Delete the wrapper and call the task directly.

Nothing is lost, because a Hatchet task is independently runnable. The call site changes from the wrapper's name to the task's name, and the run in the dashboard is the work itself rather than a wrapper around it. You get one run per invocation instead of two, no durable checkpoint on a call that never needed one, and no determinism constraint on code that never had one to respect.

Step 14: Observability

Hatchet contains a fully-featured web UI which aims to be a superset of the Temporal Web UI. It is available in both Hatchet Cloud and self-hosted deployments. The UI provides run history, workflow event views, and additional observability features such as:

• Logging from the context writes to a built-in, searchable log sink attached to the run. See logging.
• An OpenTelemetry collector ships with the platform; you can also instrument it yourself. See OpenTelemetry.
• Prometheus metrics are exposed per tenant. See Prometheus metrics.

ctx.log("charging order")

ctx.log("charging order")

Where to go next

• Tasks and running your task for the basics
• Durable execution to understand the durable context and its guarantees
• DAGs for the declarative replacement for orchestration code
• Architecture and guarantees if you are evaluating rather than migrating