Add to Cursor Add to Claude Copy for LLM View as MD

Events

Tasks can pause until an external event arrives before continuing. This is the foundation for human-in-the-loop workflows, webhook-driven pipelines, and any flow that depends on signals from outside the task.

Both durable tasks and DAGs support waiting for events. Durable tasks call WaitForEvent dynamically at runtime, while DAGs declare event conditions upfront on the task definition.

Events are delivered by pushing events into Hatchet using the event client. The event key you push must match the key your task is waiting for.

Wait For Events

Wait For Events lets a durable task pause until an external event arrives. Even if the task is interrupted and requeued while waiting, the event will still be processed. When it resumes, it reads the event from the durable event log and continues.

Waiting for an event puts the task into an evictable state, the worker slot is freed and the task is re-queued when the event arrives.

Declaring a wait for event

Wait For Event is declared using the context method WaitFor (or utility method WaitForEvent) on the DurableContext object.

examples/python/durable_event/worker.py

@hatchet.durable_task(name="DurableEventTask")
async def durable_event_task(input: EmptyModel, ctx: DurableContext) -> None:
    res = await ctx.aio_wait_for(
        "event",
        UserEventCondition(event_key="user:update"),
    )

    print("got event", res)

examples/typescript/durable_event/workflow.ts

export const durableEvent = hatchet.durableTask({
  name: 'durable-event',
  executionTimeout: '10m',
  fn: async (_, ctx) => {
    const res = ctx.waitFor({
      eventKey: 'user:update',
    });

    console.log('res', res);

    return {
      Value: 'done',
    };
  },
});

examples/go/durable/event/main.go

task := client.NewStandaloneDurableTask("long-running-task", func(ctx hatchet.DurableContext, input DurableInput) (DurableOutput, error) {
	log.Printf("Starting task, will sleep for %d seconds", input.Delay)

	if _, err := ctx.WaitForEvent("user:updated", ""); err != nil {
		return DurableOutput{}, err
	}

	log.Printf("Finished waiting for event, processing message: %s", input.Message)

	return DurableOutput{
		ProcessedAt: time.Now().Format(time.RFC3339),
		Message:     "Processed: " + input.Message,
	}, nil
})

The Ruby SDK is in early access, and may change. We'd love your feedback!

examples/ruby/durable_event/worker.rb

DURABLE_EVENT_TASK = HATCHET.durable_task(name: "DurableEventTask") do |input, ctx|
  res = ctx.wait_for(
    "event",
    Hatchet::UserEventCondition.new(event_key: "user:update")
  )

  puts "got event #{res}"
end

DURABLE_EVENT_TASK_WITH_FILTER = HATCHET.durable_task(name: "DurableEventWithFilterTask") do |input, ctx|

Event filters

Events can be filtered using CEL expressions. For example, to only receive user:update events for a specific user:

examples/python/durable_event/worker.py

res = await ctx.aio_wait_for(
    "event",
    UserEventCondition(
        event_key="user:update", expression="input.user_id == '1234'"
    ),
)

Pushing events

For a waiting task to resume, something must push an event into Hatchet with a matching key. You can do this from any service that has access to the Hatchet client.

examples/python/events/event.py

hatchet.event.push("user:create", {"should_skip": False})

When the pushed event’s key matches what a durable task is waiting for (and passes any CEL filter), the task is re-queued and resumes from its checkpoint.

Event Conditions

Event conditions let a DAG task react to external events. A task can wait for an event before running, be skipped when an event arrives, or be cancelled by an event.

Unlike durable tasks (where WaitForEvent is called dynamically at runtime), DAG event conditions are declared upfront on the task definition.

Usage modes

Event conditions can be used with three operators:

wait_for — the task waits for the event before starting.
skip_if — the task is skipped if the event arrives.
cancel_if — the task is cancelled if the event arrives.

⚠️

A task cancelled by cancel_if behaves like any other cancellation in Hatchet — downstream tasks will be cancelled as well.

Waiting for an event

Declare a task with a wait_for event condition. The task will not start until the specified event is pushed into Hatchet.

examples/python/conditions/worker.py

@task_condition_workflow.task(
    parents=[start],
    wait_for=[
        or_(
            SleepCondition(duration=timedelta(minutes=1)),
            UserEventCondition(event_key="wait_for_event:start"),
        )
    ],
)
def wait_for_event(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))

examples/typescript/dag_match_condition/complex-workflow.ts

const waitForEvent = taskConditionWorkflow.task({
  name: 'waitForEvent',
  parents: [start],
  waitFor: [Or(new SleepCondition('1m'), new UserEventCondition('wait_for_event:start', 'true'))],
  fn: () => {
    return {
      randomNumber: Math.floor(Math.random() * 100) + 1,
    };
  },
});

examples/go/conditions/main.go

waitForEvent := workflow.NewTask("wait-for-event", func(ctx hatchet.Context, _ any) (StepOutput, error) {
	return StepOutput{RandomNumber: rand.Intn(100) + 1}, nil //nolint:gosec
},
	hatchet.WithParents(start),
	hatchet.WithWaitFor(hatchet.OrCondition(
		hatchet.SleepCondition(1*time.Minute),
		hatchet.UserEventCondition("wait_for_event:start", ""),
	)),
)

The Ruby SDK is in early access, and may change. We'd love your feedback!

examples/ruby/conditions/worker.rb

WAIT_FOR_EVENT = TASK_CONDITION_WORKFLOW.task(
  :wait_for_event,
  parents: [COND_START],
  wait_for: [
    Hatchet.or_(
      Hatchet::SleepCondition.new(60),
      Hatchet::UserEventCondition.new(event_key: "wait_for_event:start")
    )
  ]
) do |input, ctx|
  { "random_number" => rand(1..100) }
end

Skipping on an event

Declare a task with a skip_if event condition. The task will be skipped if the event arrives before the task starts.

examples/python/conditions/worker.py

@task_condition_workflow.task(
    parents=[start],
    wait_for=[SleepCondition(timedelta(seconds=30))],
    skip_if=[UserEventCondition(event_key="skip_on_event:skip")],
)
def skip_on_event(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))

Event filters

Events can be filtered using CEL expressions. The CEL expression is evaluated against the event payload, and the condition only matches if the expression returns true. This works identically to event filters in durable tasks.

Pushing events

For a waiting task to proceed, something must push an event into Hatchet with a matching key. You can do this from any service that has access to the Hatchet client.

examples/python/events/event.py

hatchet.event.push("user:create", {"should_skip": False})

Combining with other conditions

Event conditions can be combined with parent and sleep conditions using or groups. For example, you can wait for either an event or a timeout (whichever comes first). See Conditions & Branching for details.

Sleep & Delays Conditions & Branching

We use cookies

Events

Wait For Events

Declaring a wait for event

Event filters

Pushing events

Event Conditions

Usage modes

Waiting for an event

Skipping on an event

Event filters

Pushing events

Combining with other conditions