Declaring Your First Task

In Hatchet, the fundamental unit of invocable work is a Task. Each task is an atomic function.

As we continue to build with Hatchet, we'll add additional configuration options to compose tasks into workflows with DAGs or procedural child spawning.

Defining a Task

Start by declaring a task with a name. The task object can declare additional task-level configuration options which we'll cover later.

The returned object is an instance of the StandaloneTaskWorkflow class, which is the primary interface for interacting with the task (i.e. running, enqueuing, scheduling, etc).

from hatchet_sdk import Context, EmptyModel, Hatchet
from pydantic import BaseModel
 
hatchet = Hatchet(debug=True)
 
class SimpleInput(BaseModel):
    message: str
 
@hatchet.task(name="SimpleWorkflow")
def simple(input: SimpleInput, ctx: Context) -> dict[str, str]:
    return {
      "transformed_message": input.message.lower(),
    }

Running a Task

With your task defined, you can import it wherever you need to use it and invoke it with the run method.

⚠️

NOTE: You must first register the task on a worker before you can run it. Calling your_task.run will enqueue a task to be executed by a worker but it will wait indefinitely for the task to be executed.

simple.run(SimpleInput(message="HeLlO WoRlD"))

There are many ways to run a task, including:

Now that you have defined a complete task, you can move on to creating a worker to execute the task.

Setup Workers