Scheduling tasks

Hatchet supports three scheduling modes:

Mode	When to use
Cron	Repeat a task on a fixed schedule (e.g. every hour, every Monday at 9 AM)
One-time	Trigger a run at a specific future datetime
Event	Trigger a run whenever a named event is pushed from your application

For long-running tasks that need to sleep mid-execution, use a durable task with ctx.aio_sleep_for() instead of scheduling a new workflow.

Cron schedules

Declare a cron in the workflow definition

Add a cron expression directly to the workflow to have every registered worker run it on schedule.

Each cron expression declared on a workflow definition runs on the schedule as long as at least one worker with that workflow is connected. You do not need to manage the trigger separately.

Python
TypeScript
Go

worker.py

from hatchet_sdk import Context, EmptyModel, Hatchet

hatchet = Hatchet(debug=True)

cron_workflow = hatchet.workflow(name="CronWorkflow", on_crons=["* * * * *"])

@cron_workflow.task()
def step1(input: EmptyModel, ctx: Context) -> dict[str, str]:
    return {"time": "step1"}

def main() -> None:
    worker = hatchet.worker("test-worker", slots=1, workflows=[cron_workflow])
    worker.start()

if __name__ == "__main__":
    main()

workflow.ts

import { hatchet } from './hatchet-client';

export const onCron = hatchet.workflow({
  name: 'on-cron-workflow',
  on: {
    // Run every 15 minutes
    cron: '*/15 * * * *',
  },
});

onCron.task({
  name: 'job',
  fn: (input) => {
    return { TransformedMessage: 'done' };
  },
});

worker.ts

import { hatchet } from './hatchet-client';
import { onCron } from './workflow';

async function main() {
  const worker = await hatchet.worker('on-cron-worker', {
    workflows: [onCron],
  });

  await worker.start();
}

if (require.main === module) {
  main();
}

main.go

package main

import (
    "log"
    "time"
    "github.com/hatchet-dev/hatchet/pkg/cmdutils"
    hatchet "github.com/hatchet-dev/hatchet/sdks/go"
)

type CronInput struct {
    Timestamp string `json:"timestamp"`
}

type CronOutput struct {
    JobName    string `json:"job_name"`
    ExecutedAt string `json:"executed_at"`
}

func main() {
    client, err := hatchet.NewClient()
    if err != nil {
        log.Fatalf("failed to create hatchet client: %v", err)
    }

    dailyCleanup := client.NewStandaloneTask("cleanup-temp-files",
        func(ctx hatchet.Context, input CronInput) (CronOutput, error) {
            log.Printf("Running daily cleanup at %s", input.Timestamp)
            return CronOutput{
                JobName:    "daily-cleanup",
                ExecutedAt: time.Now().Format(time.RFC3339),
            }, nil
        },
        hatchet.WithWorkflowCron("0 2 * * *"),
        hatchet.WithWorkflowCronInput(CronInput{
            Timestamp: time.Now().Format(time.RFC3339),
        }),
    )

    worker, err := client.NewWorker("cron-worker",
        hatchet.WithWorkflows(dailyCleanup),
    )
    if err != nil {
        log.Fatalf("failed to create worker: %v", err)
    }

    interruptCtx, cancel := cmdutils.NewInterruptContext()
    defer cancel()

    if err := worker.StartBlocking(interruptCtx); err != nil {
        log.Fatalf("failed to start worker: %v", err)
    }
}

Create and manage crons programmatically

You can also create named cron triggers at runtime via the Hatchet client. This is useful when each customer or tenant needs their own schedule.

Cron names must be unique per workflow. Creating a cron with a name that already exists will update the existing trigger rather than create a duplicate.

Python (sync)
Python (async)
Go

programatic-sync.py

from pydantic import BaseModel
from hatchet_sdk import Hatchet

hatchet = Hatchet()

class DynamicCronInput(BaseModel):
    name: str

dynamic_cron_workflow = hatchet.workflow(
    name="DynamicCronWorkflow", input_validator=DynamicCronInput
)

# Create a named cron trigger
cron_trigger = dynamic_cron_workflow.create_cron(
    cron_name="customer-a-daily-report",
    expression="0 12 * * *",
    input=DynamicCronInput(name="John Doe"),
    additional_metadata={"customer_id": "customer-a"},
)

id = cron_trigger.metadata.id  # the id of the cron trigger

# List all cron triggers
cron_triggers = hatchet.cron.list()

# Get a specific cron trigger
cron_trigger = hatchet.cron.get(cron_id=id)

# Delete a cron trigger
hatchet.cron.delete(cron_id=id)

programatic-async.py

from pydantic import BaseModel
from hatchet_sdk import Hatchet

hatchet = Hatchet()

class DynamicCronInput(BaseModel):
    name: str

async def create_cron() -> None:
    dynamic_cron_workflow = hatchet.workflow(
        name="DynamicCronWorkflow", input_validator=DynamicCronInput
    )

    # Create a named cron trigger
    cron_trigger = await dynamic_cron_workflow.aio_create_cron(
        cron_name="customer-a-daily-report",
        expression="0 12 * * *",
        input=DynamicCronInput(name="John Doe"),
        additional_metadata={"customer_id": "customer-a"},
    )

    id = cron_trigger.metadata.id

    # List all cron triggers
    await hatchet.cron.aio_list()

    # Get a specific cron trigger
    cron_trigger = await hatchet.cron.aio_get(cron_id=id)

    # Delete a cron trigger
    await hatchet.cron.aio_delete(cron_id=id)

main.go

package main

import (
    "context"
    "log"
    "time"

    "github.com/hatchet-dev/hatchet/pkg/client/rest"
    hatchet "github.com/hatchet-dev/hatchet/sdks/go"
    "github.com/hatchet-dev/hatchet/sdks/go/features"
)

// Create a named cron trigger
createdCron, err := client.Crons().Create(
    context.Background(),
    "cleanup-temp-files",
    features.CreateCronTrigger{
        Name:       "daily-cleanup",
        Expression: "0 0 * * *",
        Input: map[string]interface{}{
            "timestamp": time.Now().Format(time.RFC3339),
        },
        AdditionalMetadata: map[string]interface{}{
            "description": "Daily cleanup and maintenance tasks",
        },
    },
)

// Delete a cron trigger
err = client.Crons().Delete(context.Background(), createdCron.Metadata.Id)

// List cron triggers (with optional filters)
cronList, err := client.Crons().List(
    context.Background(),
    rest.CronWorkflowListParams{
        AdditionalMetadata: &[]string{"description:Daily cleanup and maintenance tasks"},
    },
)

One-time scheduled runs

Schedule a single run to execute at a specific future datetime. Hatchet stores the trigger and fires it at the given time regardless of whether a worker is currently connected.

Python (sync)
Python (async)
Go

programatic-sync.py

from datetime import datetime, timedelta, timezone
from hatchet_sdk import Hatchet

hatchet = Hatchet()

# Schedule a run 10 seconds from now
scheduled_run = hatchet.scheduled.create(
    workflow_name="simple-workflow",
    trigger_at=datetime.now(tz=timezone.utc) + timedelta(seconds=10),
    input={"data": "simple-workflow-data"},
    additional_metadata={"customer_id": "customer-a"},
)

id = scheduled_run.metadata.id  # the id of the scheduled run trigger

# Reschedule to a later time
hatchet.scheduled.update(
    scheduled_id=id,
    trigger_at=datetime.now(tz=timezone.utc) + timedelta(hours=1),
)

# List all scheduled runs
scheduled_runs = hatchet.scheduled.list()

# Get a specific scheduled run
scheduled_run = hatchet.scheduled.get(scheduled_id=id)

# Delete a scheduled run
hatchet.scheduled.delete(scheduled_id=id)

programatic-async.py

from datetime import datetime, timedelta, timezone
from hatchet_sdk import Hatchet

hatchet = Hatchet()

async def create_scheduled() -> None:
    scheduled_run = await hatchet.scheduled.aio_create(
        workflow_name="simple-workflow",
        trigger_at=datetime.now(tz=timezone.utc) + timedelta(seconds=10),
        input={"data": "simple-workflow-data"},
        additional_metadata={"customer_id": "customer-a"},
    )

    id = scheduled_run.metadata.id

    # Delete it
    await hatchet.scheduled.aio_delete(scheduled_id=id)

    # List all
    await hatchet.scheduled.aio_list()

    # Get by id
    scheduled_run = await hatchet.scheduled.aio_get(scheduled_id=id)

main.go

package main

import (
    "context"
    "log"
    "time"

    "github.com/hatchet-dev/hatchet/pkg/client/rest"
    hatchet "github.com/hatchet-dev/hatchet/sdks/go"
    "github.com/hatchet-dev/hatchet/sdks/go/features"
)

// Schedule a run 1 minute from now
scheduledRun, err := client.Schedules().Create(
    context.Background(),
    "scheduled",
    features.CreateScheduledRunTrigger{
        TriggerAt: time.Now().Add(1 * time.Minute),
        Input:     map[string]interface{}{"message": "Hello, World!"},
    },
)
if err != nil {
    log.Fatalf("failed to create scheduled run: %v", err)
}

// Delete the scheduled run
err = client.Schedules().Delete(
    context.Background(),
    scheduledRun.Metadata.Id,
)

// List all scheduled runs
scheduledRuns, err := client.Schedules().List(
    context.Background(),
    rest.WorkflowScheduledListParams{},
)

You can also schedule a run from inside a task. This is useful for chaining workflows across long time gaps:

worker.py

from datetime import datetime, timedelta, timezone
from hatchet_sdk import Context, Hatchet

hatchet = Hatchet(debug=True)

print_printer_wf = hatchet.workflow(name="PrintPrinterWorkflow")
print_schedule_wf = hatchet.workflow(name="PrintScheduleWorkflow")

@print_schedule_wf.task()
def schedule(input, ctx: Context) -> None:
    now = datetime.now(tz=timezone.utc)
    future_time = now + timedelta(seconds=15)
    # Schedule the second workflow to run 15 seconds from now
    print_printer_wf.schedule(future_time, input=input)

Durable sleep

Inside a durable task, call ctx.aio_sleep_for() to pause execution for a fixed duration without holding a worker slot. Hatchet persists the checkpoint and resumes the task on any available worker when the timer fires.

Python
TypeScript

worker.py

from datetime import timedelta
from hatchet_sdk import DurableContext, EmptyModel, Hatchet

hatchet = Hatchet(debug=True)

durable_workflow = hatchet.workflow(name="DurableWorkflow")

@durable_workflow.durable_task()
async def durable_task(input: EmptyModel, ctx: DurableContext) -> dict:
    sleep = await ctx.aio_sleep_for(duration=timedelta(seconds=5))
    return {"sleep_duration_seconds": sleep.duration.seconds}

workflow.ts

import { hatchet } from './hatchet-client';

export const durableSleep = hatchet.workflow({ name: 'durable-sleep' });

durableSleep.durableTask({
  name: 'durable-sleep',
  executionTimeout: '10m',
  fn: async (_, ctx) => {
    console.log('sleeping for 5s');
    const sleepRes = await ctx.sleepFor('5s');
    console.log('done sleeping for 5s', sleepRes);
    return { Value: 'done' };
  },
});

See Durable execution for the full reference, including waiting for external events and combining conditions.

Event triggers

A workflow can declare one or more event keys. Whenever your application pushes a matching event, Hatchet starts a new run of the workflow automatically.

Define an event-triggered workflow

Python
TypeScript
Go

worker.py

from pydantic import BaseModel
from hatchet_sdk import Context, Hatchet

hatchet = Hatchet()

EVENT_KEY = "user:create"

class EventWorkflowInput(BaseModel):
    should_skip: bool

event_workflow = hatchet.workflow(
    name="EventWorkflow",
    on_events=[EVENT_KEY],
    input_validator=EventWorkflowInput,
)

@event_workflow.task()
def task(input: EventWorkflowInput, ctx: Context) -> dict[str, str]:
    print("event received")
    return {}

def main() -> None:
    worker = hatchet.worker(name="EventWorker", workflows=[event_workflow])
    worker.start()

if __name__ == "__main__":
    main()

workflow.ts

import { hatchet } from './hatchet-client';

export const SIMPLE_EVENT = 'simple-event:create';

export const lower = hatchet.workflow({
  name: 'lower',
  onEvents: ['simple-event:create'],
});

lower.task({
  name: 'lower',
  fn: (input: { Message: string }) => ({
    TransformedMessage: input.Message.toLowerCase(),
  }),
});

main.go

package main

import (
    "strings"
    hatchet "github.com/hatchet-dev/hatchet/sdks/go"
)

const SimpleEvent = "simple-event:create"

type EventInput struct{ Message string }
type LowerTaskOutput struct{ TransformedMessage string }

func Lower(client *hatchet.Client) *hatchet.StandaloneTask {
    return client.NewStandaloneTask(
        "lower",
        func(ctx hatchet.Context, input EventInput) (*LowerTaskOutput, error) {
            return &LowerTaskOutput{
                TransformedMessage: strings.ToLower(input.Message),
            }, nil
        },
        hatchet.WithWorkflowEvents(SimpleEvent),
    )
}

Push an event

Use the Hatchet client from anywhere in your application to fire an event:

Python
TypeScript
Go

event.py

from hatchet_sdk import Hatchet

hatchet = Hatchet()

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

event.ts

import { hatchet } from './hatchet-client';

await hatchet.events.push('simple-event:create', {
  Message: 'hello',
  ShouldSkip: false,
});

main.go

import (
    "context"
    hatchet "github.com/hatchet-dev/hatchet/sdks/go"
)

err := client.Events().Push(
    context.Background(),
    "simple-event:create",
    EventInput{Message: "Hello, World!"},
)

Get Started

Running Tasks

Flow Control

Workers

Observability

Self-Hosting

Scheduling tasks

Cron schedules

Declare a cron in the workflow definition

Create and manage crons programmatically

One-time scheduled runs

Durable sleep

Event triggers

Define an event-triggered workflow

Push an event

Next steps

Durable execution

Concurrency limits

Build docs developers (and LLMs) love

Get Started

Running Tasks

Flow Control

Workers

Observability

Self-Hosting

​Cron schedules

​Declare a cron in the workflow definition

​Create and manage crons programmatically

​One-time scheduled runs

​Durable sleep

​Event triggers

​Define an event-triggered workflow

​Push an event

​Next steps

Durable execution

Concurrency limits

Build docs developers (and LLMs) love

Cron schedules

Declare a cron in the workflow definition

Create and manage crons programmatically

One-time scheduled runs

Durable sleep

Event triggers

Define an event-triggered workflow

Push an event

Next steps