Hatchet

https://docs.hatchet.run/v1
Admin
Hatchet is a distributed, durable execution engine and developer platform for building, scheduling,...
Tokens:89,897
Snippets:909
Trust Score:8.3
Update:1 month ago
Show doc for...
Tokens:
Raw
### Run Hatchet Quickstart

Source: https://docs.hatchet.run/v1/quickstart

Initializes the quickstart project environment.

```sh
hatchet quickstart
```

--------------------------------

### Ruby Dockerfile Example

Source: https://docs.hatchet.run/v1/docker

Dockerfile for Ruby applications, including system dependency installation for native gems and bundle configuration.

```dockerfile
FROM ruby:3.3-slim

ENV HATCHET_ENV=production

# Install system dependencies for native gems

RUN apt-get update && \
 apt-get install -y build-essential && \
 apt-get clean && \
 rm -rf /var/lib/apt/lists/*

WORKDIR /app

COPY Gemfile Gemfile.lock ./

RUN bundle config set --local without 'development test' && \
 bundle install

COPY . /app

CMD ["bundle", "exec", "ruby", "worker.rb"]
```

--------------------------------

### Common Environment Variable Examples

Source: https://docs.hatchet.run/v1/compute/environment-variables

Examples of standard environment variable definitions for application configuration.

```text
# Examples of commonly used variables
NODE_ENV=production
LOG_LEVEL=info
API_TIMEOUT=30000
DATABASE_URL=postgresql://user:pass@host:5432/db
```

--------------------------------

### Install Hatchet SDKs

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Commands to install the Hatchet SDK for various programming languages.

```bash
pip install hatchet-sdk
```

```bash
npm install @hatchet-dev/typescript-sdk
```

```bash
go get github.com/hatchet-dev/hatchet/sdks/go
```

```bash
gem install hatchet-sdk
```

--------------------------------

### Install Python OpenTelemetry extra

Source: https://docs.hatchet.run/v1/opentelemetry

Install the necessary dependencies for OpenTelemetry support in the Python SDK.

```bash
pip install hatchet-sdk[otel]
```

--------------------------------

### Optimize Docker library installation

Source: https://docs.hatchet.run/v1/compute/gpu

Avoid meta-packages to keep image sizes small and install only necessary CUDA libraries.

```dockerfile
# DO NOT use meta-packages
# ❌ RUN apt-get install cuda-runtime-*

# ✅ Install only required libraries
RUN apt-get install -y \
    cuda-nvcc-12-2 \
    libcublas-12-2 \
    libcudnn8
```

--------------------------------

### Install Hatchet CLI Skills

Source: https://docs.hatchet.run/v1/using-coding-agents

Installs reference documentation for AI agents to understand Hatchet CLI operations.

```bash
hatchet skills install
```

```bash
hatchet skills install --dir ./my-project
```

--------------------------------

### Example Dockerfile Directory Structure

Source: https://docs.hatchet.run/v1/compute/git-ops

Demonstrates a project structure with multiple Dockerfiles for different worker configurations.

```text
.
├── Dockerfile.worker1        # For worker pool 1
├── Dockerfile.worker2        # For worker pool 2
└── services/
    └── special/
        └── Dockerfile        # For specialized worker
```

--------------------------------

### Install Hatchet CLI via Homebrew

Source: https://docs.hatchet.run/v1/quickstart

Installation method for MacOS users using Homebrew.

```sh
    brew install hatchet-dev/hatchet/hatchet --cask
```

--------------------------------

### Declare and start a worker

Source: https://docs.hatchet.run/v1/workers

Initialize a worker with a unique name and a list of workflows, then start the worker process.

```python
def main() -> None:
    worker = hatchet.worker("dag-worker", workflows=[dag_workflow])

    worker.start()
```

```typescript
import { hatchet } from '../hatchet-client';
import { simple } from './workflow';
import { parent, child } from './workflow-with-child';
import { simpleWithZod } from './zod';

async function main() {
  const worker = await hatchet.worker('simple-worker', {
    // 👀 Declare the workflows that the worker can execute
    workflows: [simple, simpleWithZod, parent, child],
    // 👀 Declare the number of concurrent task runs the worker can accept
    slots: 100,
  });

  await worker.start();
}

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

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

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

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

```ruby
def main
  worker = HATCHET.worker("dag-worker", workflows: [DAG_WORKFLOW])
  worker.start
end
```

--------------------------------

### Install Hatchet CLI via Native Script

Source: https://docs.hatchet.run/v1/quickstart

Recommended installation method for MacOS, Linux, and WSL environments.

```sh
    curl -fsSL https://install.hatchet.run/install.sh | bash
```

--------------------------------

### Install TypeScript OpenTelemetry packages

Source: https://docs.hatchet.run/v1/opentelemetry

Install the required OpenTelemetry core and exporter packages for the TypeScript SDK.

```bash
npm install @opentelemetry/api @opentelemetry/instrumentation @opentelemetry/sdk-trace-base @opentelemetry/exporter-trace-otlp-grpc
```

--------------------------------

### Install Hatchet SDK

Source: https://docs.hatchet.run/v1/from-celery-to-hatchet

Install the Hatchet SDK using pip to begin the migration process alongside existing Celery dependencies.

```bash
pip install hatchet-sdk
```

--------------------------------

### Example Repository Structure

Source: https://docs.hatchet.run/v1/compute/git-ops

A recommended directory layout for managing multiple worker configurations within a single repository.

```text
.
├── Dockerfile.worker1          # Primary worker configuration
├── Dockerfile.worker2          # Secondary worker configuration
├── .dockerignore
├── src/
│   ├── worker1/               # Worker 1 specific code
│   │   └── worker.py
│   └── worker2/               # Worker 2 specific code
│       └── worker.py
└── hatchet.yaml               # Optional Hatchet configuration
```

--------------------------------

### Go Dockerfile Example

Source: https://docs.hatchet.run/v1/docker

Multi-stage build for Go applications, compiling the binary in the first stage and running it in a minimal production image.

```dockerfile
# Stage 1: Build
FROM golang:1.26-alpine3.21 AS builder

WORKDIR /app

COPY . .

RUN go mod download

RUN go build -o hatchet-worker .

# Stage 2: Production

FROM golang:1.26-alpine3.21

WORKDIR /app

COPY --from=builder hatchet-worker .

CMD ["/app/hatchet-worker"]
```

--------------------------------

### Worker connection logs

Source: https://docs.hatchet.run/v1/workers

Example output confirming the worker has successfully connected to the engine.

```text
[INFO]  🪓 -- starting hatchet...
[DEBUG] 🪓 -- 'test-worker' waiting for ['simpletask:step1']
[DEBUG] 🪓 -- acquired action listener: efc4aaf2-...
[DEBUG] 🪓 -- sending heartbeat
```

--------------------------------

### Initialize V1 Reflection SDK Worker

Source: https://docs.hatchet.run/v1/migrating/migration-guide-go

Demonstrates setting up a client, defining input/output types, and starting a blocking worker using the current reflection-based SDK.

```go
package main

import (
	"context"
	"log"

	hatchet "github.com/hatchet-dev/hatchet/sdks/go"
)

func main() {
	client, err := hatchet.NewClient()
	if err != nil {
		log.Fatal(err)
	}

	// Define input/output types
	type Input struct {
		Message string `json:"message"`
	}

	type Output struct {
		Result string `json:"result"`
	}

	// Create a simple task
	task := client.NewStandaloneTask("simple-task", func(ctx hatchet.Context, input Input) (Output, error) {
		return Output{Result: "Processed: " + input.Message}, nil
	})

	// Start worker
	worker, err := client.NewWorker("worker", hatchet.WithWorkflows(task))
	if err != nil {
		log.Fatal(err)
	}

	if err := worker.StartBlocking(context.Background()); err != nil {
		log.Fatal(err)
	}
}
```

--------------------------------

### JavaScript Dockerfile Examples

Source: https://docs.hatchet.run/v1/docker

Multi-stage Docker builds for Node.js applications using npm, pnpm, or yarn package managers.

```dockerfile
# Stage 1: Build
FROM node:18 AS builder

WORKDIR /app

COPY package*.json ./

RUN npm ci

COPY . .

RUN npm run build

# Stage 2: Production
FROM node:22-alpine

WORKDIR /app

COPY package*.json ./

RUN npm ci --omit=dev

COPY --from=builder /app/dist ./dist

ENV NODE_ENV=production

CMD ["node", "dist/worker.js"]
```

```dockerfile
# Stage 1: Build
FROM node:18 AS builder

WORKDIR /app

# Install pnpm
RUN npm install -g pnpm

COPY pnpm-lock.yaml package.json ./

RUN pnpm install --frozen-lockfile

COPY . .

RUN pnpm build

# Stage 2: Production
FROM node:22-alpine

WORKDIR /app

RUN npm install -g pnpm

COPY pnpm-lock.yaml package.json ./

RUN pnpm install --frozen-lockfile --prod

COPY --from=builder /app/dist ./dist

ENV NODE_ENV=production

CMD ["node", "dist/worker.js"]
```

```dockerfile
# Stage 1: Build
FROM node:18 AS builder

WORKDIR /app

COPY package.json yarn.lock ./

RUN yarn install --frozen-lockfile

COPY . .

RUN yarn build

# Stage 2: Production
FROM node:22-alpine

WORKDIR /app

COPY package.json yarn.lock ./

RUN yarn install --frozen-lockfile --production

COPY --from=builder /app/dist ./dist

ENV NODE_ENV=production

CMD ["node", "dist/worker.js"]
```

--------------------------------

### Run worker with TypeScript

Source: https://docs.hatchet.run/v1/workers

Define a start script in package.json and execute it via npm.

```json
"scripts": {
  "start:worker": "ts-node src/worker.ts"
}
```

```bash
npm run start:worker
```

--------------------------------

### Install GPU library dependencies

Source: https://docs.hatchet.run/v1/compute/gpu

Install common CUDA libraries and specific NVIDIA drivers required for GPU operations.

```dockerfile
# Install commonly needed libraries
RUN apt-get install -y \
    libcublas-12-2 \
    libcudnn8 \
    nvidia-driver-525
```

--------------------------------

### Python Dockerfile Examples

Source: https://docs.hatchet.run/v1/docker

Docker configurations for Python applications using either Poetry or pip for dependency management.

```dockerfile
FROM python:3.13-slim

ENV PYTHONUNBUFFERED=1 \
 POETRY_VERSION=1.4.2 \
 HATCHET_ENV=production

# Install system dependencies and Poetry
RUN apt-get update && \
 apt-get install -y curl && \
 curl -sSL https://install.python-poetry.org | python3 - && \
 ln -s /root/.local/bin/poetry /usr/local/bin/poetry && \
 apt-get clean && \
 rm -rf /var/lib/apt/lists/*

WORKDIR /app

COPY pyproject.toml poetry.lock* /app/

RUN poetry config virtualenvs.create false && \
 poetry install --no-interaction --no-ansi

COPY . /app

CMD ["poetry", "run", "python", "worker.py"]
```

```dockerfile
FROM python:3.13-slim

ENV PYTHONUNBUFFERED=1 \
 HATCHET_ENV=production

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY . /app

CMD ["python", "worker.py"]
```

--------------------------------

### Implement a multi-step onboarding flow with sleeps

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Examples of workflows that include waiting periods, requiring execution timeouts to cover the total duration of the task.

```python
@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)
```

```typescript
export const onboardingFlow = hatchet.durableTask({
  name: 'OnboardingFlow',
  // The timeout has to cover the whole wall-clock span of the run, sleeps included.
  executionTimeout: '168h',
  fn: async (input: SignupInput, ctx): Promise => {
    await sendWelcomeEmail.run(input);

    await ctx.sleepFor('72h');

    return sendFollowupEmail.run(input);
  },
});
```

```go
onboardingFlow := client.NewStandaloneDurableTask("OnboardingFlow",
	func(ctx hatchet.DurableContext, input OnboardingInput) (OnboardingOutput, error) {
		if _, err := sendWelcomeEmail.Run(ctx, input); err != nil {
			return OnboardingOutput{}, fmt.Errorf("sending welcome email: %w", err)
		}

		if _, err := ctx.SleepFor(72 * time.Hour); err != nil {
			return OnboardingOutput{}, fmt.Errorf("sleeping between emails: %w", err)
		}

		if _, err := sendFollowupEmail.Run(ctx, input); err != nil {
			return OnboardingOutput{}, fmt.Errorf("sending follow-up email: %w", err)
		}

		return OnboardingOutput{EmailsSent: 2}, nil
	},
	// The execution timeout has to cover the sleep, not just the work.
	hatchet.WithExecutionTimeout(168*time.Hour),
)
```

```ruby
# The execution timeout has to cover the whole wall-clock span of the run,
# sleeps included, so a three-day sleep needs a timeout longer than three days.
ONBOARDING_FLOW = HATCHET.durable_task(
  name: "OnboardingFlow",
  execution_timeout: 604_800
) do |input, ctx|
  welcome = SEND_WELCOME_EMAIL.run(input)

  ctx.sleep_for(duration: 259_200)

  followup = SEND_FOLLOWUP_EMAIL.run(input)

  {
    "user_id" => input["user_id"],
    "welcome_message_id" => welcome["message_id"],
    "followup_message_id" => followup["message_id"]
  }
end
```

--------------------------------

### Define a Temporal Activity

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Examples of defining activities in Temporal across different SDKs.

```python
from temporalio import activity


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

```typescript
// activities.ts — activities are plain exported functions.
export async function chargeOrder(orderId: string): Promise<boolean> {
  return payments.charge(orderId);
}
```

```go
import "context"

func ChargeOrder(ctx context.Context, orderID string) (bool, error) {
	return payments.Charge(ctx, orderID)
}
```

```ruby
require "temporalio/activity"

class ChargeOrder < Temporalio::Activity::Definition
  def execute(order_id)
    Payments.charge(order_id)
  end
end
```

--------------------------------

### Configure Cancel Newest Strategy

Source: https://docs.hatchet.run/v1/concurrency

Examples of implementing the CANCEL_NEWEST concurrency strategy across different supported languages.

```python
class WorkflowInput(BaseModel):
    group: str


concurrency_cancel_newest_workflow = hatchet.workflow(
    name="ConcurrencyCancelNewest",
    concurrency=ConcurrencyExpression(
        expression="input.group",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.CANCEL_NEWEST,
    ),
    input_validator=WorkflowInput,
)
```

```typescript
export const concurrencyCancelNewestWorkflow = hatchet.workflow({
  name: 'concurrencycancelnewest',
  concurrency: {
    expression: 'input.group',
    maxRuns: 1,
    limitStrategy: ConcurrencyLimitStrategy.CANCEL_NEWEST,
  },
});
```

```go
var maxRuns int32 = 1
strategy := types.CancelNewest

return client.NewStandaloneTask("cancel-newest",
	func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet.WithWorkflowConcurrency(types.Concurrency{
		Expression:    "input.GroupKey",
		MaxRuns:       &maxRuns,
		LimitStrategy: &strategy,
	}),
)
```

```ruby
CONCURRENCY_CANCEL_NEWEST_WORKFLOW = HATCHET.workflow(
  name: "ConcurrencyCancelNewest",
  concurrency: Hatchet::ConcurrencyExpression.new(
    expression: "input.group",
    max_runs: 1,
    limit_strategy: :cancel_newest
  )
)
```

--------------------------------

### Prometheus configuration for metrics

Source: https://docs.hatchet.run/v1/worker-healthchecks

Example configuration for scraping Hatchet worker metrics.

```yaml
scrape_configs:
  - job_name: "hatchet"
    scrape_interval: 5s
    static_configs:
      - targets: ["localhost:8001"]
```

--------------------------------

### Start a Celery worker

Source: https://docs.hatchet.run/v1/from-celery-to-hatchet

Standard CLI command for launching a Celery worker with specified concurrency.

```bash
celery -A tasks worker --loglevel=info --concurrency=4
```

--------------------------------

### Implement a Go Standalone Task

Source: https://docs.hatchet.run/v1/inter-service-triggering

Example of defining and running a standalone task in Go, which can be triggered by other services sharing the same name and schema.

```go
package main

import (
	"context"
	"fmt"
	"log"

	hatchet "github.com/hatchet-dev/hatchet/sdks/go"
)

type StubInput struct {
	Message string `json:"message"`
}

type StubOutput struct {
	Ok bool `json:"ok"`
}

func StubWorkflow(client *hatchet.Client) *hatchet.StandaloneTask {
	return client.NewStandaloneTask("stub-workflow", func(ctx hatchet.Context, input StubInput) (StubOutput, error) {
		return StubOutput{
			Ok: true,
		}, nil
	})
}

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

	task := StubWorkflow(client)

	// we are simply running the task here, but it can be implemented in another service / worker
	// and in another language with the same name and input-output types
	result, err := task.Run(context.Background(), StubInput{Message: "Hello, World!"})
	if err != nil {
		log.Fatalf("failed to run task: %v", err)
	}

	fmt.Println(result)
}
```

--------------------------------

### Fetch Metrics via cURL

Source: https://docs.hatchet.run/v1/prometheus-metrics

Example command to retrieve metrics using a Bearer token.

```bash
curl -H "Authorization: Bearer your-api-token-here" \
  https://cloud.onhatchet.run/api/v1/tenants/707d0855-80ab-4e1f-a156-f1c4546cbf52/prometheus-metrics
```

--------------------------------

### Install Hatchet Docs MCP and Agent Skills

Source: https://docs.hatchet.run/v1/quickstart

Commands to integrate Hatchet documentation and CLI skills into AI coding assistants.

```sh
hatchet docs install
```

```sh
hatchet skills install
```

--------------------------------

### Configure Task Retries

Source: https://docs.hatchet.run/v1/retry-policies

Examples of setting the retries property across different supported languages.

```python
@simple_workflow.task(retries=3)
def always_fail(input: EmptyModel, ctx: Context) -> dict[str, str]:
    raise Exception("simple task failed")
```

```typescript
export const retries = hatchet.task({
  name: 'retries',
  retries: 3,
  fn: async (_, ctx) => {
    throw new Error('intentional failure');
  },
});
```

```go
retries := client.NewStandaloneTask("retries-task", func(ctx hatchet.Context, input RetriesInput) (*RetriesResult, error) {
	return nil, errors.New("intentional failure")
}, hatchet.WithRetries(3))
```

```ruby
SIMPLE_RETRY_WORKFLOW.task(:always_fail, retries: 3) do |input, ctx|
  raise "simple task failed"
end
```

--------------------------------

### Configure task retries and timeouts

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Examples of applying retry policies, backoff settings, and execution timeouts to tasks in various languages.

```python
@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))
```

```typescript
export const chargeOrderWithRetries = hatchet.task({
  name: 'charge-order-with-retries',
  retries: 10,
  backoff: {
    // Factor to increase the wait time between retries.
    factor: 2,
    // Maximum number of seconds to wait between retries.
    maxSeconds: 10,
  },
  executionTimeout: '30s',
  scheduleTimeout: '10m',
  fn: async (input: OrderInput): Promise => {
    const amountCents = await payments.charge(input.orderId);

    return { charged: amountCents > 0, amountCents };
  },
});
```

```go
chargeOrderWithRetries := client.NewStandaloneTask("charge-order-with-retries",
	func(ctx hatchet.Context, input OrderInput) (ChargeOutput, error) {
		amount, err := capturePayment(input.OrderID)
		if err != nil {
			return ChargeOutput{}, fmt.Errorf("charging order %q: %w", input.OrderID, err)
		}

		return ChargeOutput{Charged: true, AmountCents: amount}, nil
	},
	// retries counts retries, not attempts: maximum_attempts=11 becomes 10.
	hatchet.WithRetries(10),
	// factor, maxBackoffSeconds
	hatchet.WithRetryBackoff(2, 10),
	hatchet.WithExecutionTimeout(30*time.Second),
	hatchet.WithScheduleTimeout(10*time.Minute),
)
```

```ruby
CHARGE_ORDER_WITH_RETRIES = HATCHET.task(
  name: "charge-order-with-retries",
  # Hatchet counts retries, not attempts: this is 11 attempts in total.
  retries: 10,
  backoff_factor: 2.0,
  backoff_max_seconds: 10,
  execution_timeout: 30,
  schedule_timeout: 600
) do |input, ctx|
  charge = Payments.charge(input["order_id"], input["amount_cents"])

  { "transaction_id" => charge["transaction_id"], "attempt" => ctx.retry_count + 1 }
end
```

--------------------------------

### Install Hatchet Docs for Claude Code

Source: https://docs.hatchet.run/v1/using-coding-agents

Configures Claude Code to access Hatchet documentation via the MCP server.

```bash
hatchet docs install claude-code
```

--------------------------------

### Execute child workflows in Temporal

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Examples of spawning child workflows concurrently in Temporal across multiple languages.

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

```typescript
import { executeChild } from "@temporalio/workflow";

const results = await Promise.all(
  items.map((item) => executeChild(processItem, { args: [item] })),
);
```

```go
// Collect the child futures, then await them.
var futures []workflow.Future
for _, item := range items {
	futures = append(futures, workflow.ExecuteChildWorkflow(ctx, ProcessItem, item))
}

for _, f := range futures {
	var result ItemOutput
	if err := f.Get(ctx, &result); err != nil {
		return err
	}
}
```

```ruby
results = items.map do |item|
  Temporalio::Workflow.execute_child_workflow(
    ProcessItem, item, schedule_to_close_timeout: 300
  )
end
```

--------------------------------

### Configure Cancel In Progress Concurrency

Source: https://docs.hatchet.run/v1/concurrency

Examples of setting the CANCEL_IN_PROGRESS limit strategy across different supported languages.

```python
class WorkflowInput(BaseModel):
    group: str


concurrency_cancel_in_progress_workflow = hatchet.workflow(
    name="ConcurrencyCancelInProgress",
    concurrency=ConcurrencyExpression(
        expression="input.group",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,
    ),
    input_validator=WorkflowInput,
)
```

```typescript
export const concurrencyCancelInProgressWorkflow = hatchet.workflow({
  name: 'concurrencycancelinprogress',
  concurrency: {
    expression: 'input.group',
    maxRuns: 1,
    limitStrategy: ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,
  },
});
```

```go
var maxRuns int32 = 1
strategy := types.CancelInProgress

return client.NewStandaloneTask("cancel-in-progress",
	func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet.WithWorkflowConcurrency(types.Concurrency{
		Expression:    "input.GroupKey",
		MaxRuns:       &maxRuns,
		LimitStrategy: &strategy,
	}),
)
```

```ruby
CONCURRENCY_CANCEL_IN_PROGRESS_WORKFLOW = HATCHET.workflow(
  name: "ConcurrencyCancelInProgress",
  concurrency: Hatchet::ConcurrencyExpression.new(
    expression: "input.group",
    max_runs: 1,
    limit_strategy: :cancel_in_progress
  )
)
```

--------------------------------

### Declare Static Rate Limits

Source: https://docs.hatchet.run/v1/rate-limits

Use the Admin client to define a rate limit key, its capacity, and the duration before starting workers.

```python
RATE_LIMIT_KEY = "test-limit"

hatchet.rate_limits.put(RATE_LIMIT_KEY, 2, RateLimitDuration.SECOND)
```

```typescript
hatchet.ratelimits.upsert({
  key: 'api-service-rate-limit',
  limit: 10,
  duration: RateLimitDuration.SECOND,
});
```

```go
err = client.RateLimits().Upsert(features.CreateRatelimitOpts{
	Key:      RATE_LIMIT_KEY,
	Limit:    10,
	Duration: types.Second,
})
if err != nil {
	log.Fatalf("failed to create rate limit: %v", err)
}
```

```ruby
def main
  HATCHET.rate_limits.put(RATE_LIMIT_KEY, 2, :second)

  worker = HATCHET.worker(
    "rate-limit-worker", slots: 10, workflows: [RATE_LIMIT_WORKFLOW]
  )
  worker.start
end
```

--------------------------------

### Fetch Task Stats via cURL

Source: https://docs.hatchet.run/v1/autoscaling-workers

Example command to retrieve task statistics using an API token.

```bash
curl -H "Authorization: Bearer your-api-token-here" \
  https://cloud.onhatchet.run/api/v1/tenants/707d0855-80ab-4e1f-a156-f1c4546cbf52/task-stats
```

--------------------------------

### Execute child tasks in Hatchet

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Examples of implementing fan-out patterns using Hatchet durable tasks.

```python
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)
```

```typescript
export const packOrder = hatchet.durableTask({
  name: 'PackOrder',
  executionTimeout: '30m',
  fn: async (input: { items: string[] }): Promise<{ packed: number }> => {
    const results = await Promise.all(input.items.map((item) => processItem.run({ item })));

    return { packed: results.filter((result) => result.packed).length };
  },
});
```

```go
var wg sync.WaitGroup

outputs := make([]ItemOutput, len(items))
errs := make([]error, len(items))

wg.Add(len(items))

for i, item := range items {
	go func(i int, item Item) {
		defer wg.Done()

		result, err := processItem.Run(ctx, ItemInput{Item: item})
		if err != nil {
			errs[i] = fmt.Errorf("running item %d: %w", i, err)
			return
		}

		if err := result.Into(&outputs[i]); err != nil {
			errs[i] = fmt.Errorf("decoding result for item %d: %w", i, err)
		}
	}(i, item)
}

wg.Wait()

if err := errors.Join(errs...); err != nil {
	return ShipmentOutput{}, err
}
```

```ruby
PROCESS_ITEM = HATCHET.task(name: "process-item") do |input, _ctx|
  { "item" => input["item"], "processed" => true }
end

FAN_OUT_ORDER_ITEMS = HATCHET.durable_task(name: "FanOutOrderItems", execution_timeout: 300) do |input, _ctx|
  items = input["items"] || []

  # Ruby's trigger methods are synchronous, so fan out with the bulk API rather
  # than one call per child.
  results = PROCESS_ITEM.run_many(
    items.map { |item| PROCESS_ITEM.create_bulk_run_item(input: { "item" => item }) }
  )

  { "results" => results }
end
```

--------------------------------

### Define a batch task

Source: https://docs.hatchet.run/v1/batch-tasks

Examples of defining batch tasks across different supported languages with specific size and interval configurations.

```python
@hatchet.batch_task(
    batch_max_size=3,
    batch_max_interval=timedelta(milliseconds=200),
    input_validator=SimpleInput,
)
async def batch_simple(
    tasks: dict[BatchMemberId, SimpleInput], context: Context
) -> dict[BatchMemberId, SimpleOutput]:
    return {
        id: SimpleOutput(transformed_message=inp.message.upper())
        for id, inp in tasks.items()
    }
```

```typescript
export const batchSimple = hatchet.batchTask({
  name: 'batch-simple',
  batch: { maxSize: 3, maxInterval: 200 },
  fn: async (tasks: Record<string, SimpleInput>) => {
    const out: Record<string, SimpleOutput> = {};
    Object.entries(tasks).forEach(([id, input]) => {
      out[id] = { transformed_message: input.message.toUpperCase() };
    });
    return out;
  },
});
```

```go
batchSimple := client.NewStandaloneBatchTask("batch-simple",
	func(ctx hatchet.Context, tasks map[string]SimpleInput) (map[string]SimpleOutput, error) {
		out := make(map[string]SimpleOutput, len(tasks))
		for id, inp := range tasks {
			out[id] = SimpleOutput{TransformedMessage: strings.ToUpper(inp.Message)}
		}
		return out, nil
	},
	hatchet.BatchConfig{
		MaxSize:     3,
		MaxInterval: durationPtr(200 * time.Millisecond),
	},
)
```

```ruby
BATCH_SIMPLE = HATCHET.batch_task(
  name: 'ruby-e2e-batch-simple',
  batch: Hatchet::BatchTaskConfig.new(max_size: 3, max_interval_ms: 200)
) do |inputs, _ctx|
  inputs.transform_values { |input| { 'transformed_message' => input['message'].upcase } }
end
```

--------------------------------

### Run Celery worker with specific queue

Source: https://docs.hatchet.run/v1/from-celery-to-hatchet

Start a Celery worker process bound to a specific queue.

```bash
celery -A tasks worker -Q image-processing
```

--------------------------------

### Define Temporal Workflow for Order Processing

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Examples of Temporal workflow definitions that execute three sequential activities.

```python
@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)
        )
```

```typescript
import { proxyActivities } from "@temporalio/workflow";
// Only import the activity types
import type * as activities from "./activities";

const { validateOrder, chargeOrder, fulfillOrder } = proxyActivities<typeof activities>({
  startToCloseTimeout: "30 seconds",
});

export async function processOrder(orderId: string): Promise<boolean> {
  await validateOrder(orderId);
  await chargeOrder(orderId);
  return fulfillOrder(orderId);
}
```

```go
func ProcessOrderWorkflow(ctx workflow.Context, orderID string) (bool, error) {
	ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 30 * time.Second,
	})

	if err := workflow.ExecuteActivity(ctx, ValidateOrder, orderID).Get(ctx, nil); err != nil {
		return false, err
	}

	if err := workflow.ExecuteActivity(ctx, ChargeOrder, orderID).Get(ctx, nil); err != nil {
		return false, err
	}

	var fulfilled bool
	err := workflow.ExecuteActivity(ctx, FulfillOrder, orderID).Get(ctx, &fulfilled)

	return fulfilled, err
}
```

```ruby
require "temporalio/workflow"

class ProcessOrderWorkflow < Temporalio::Workflow::Definition
  def execute(order_id)
    Temporalio::Workflow.execute_activity(
      ValidateOrder, order_id, start_to_close_timeout: 30
    )
    Temporalio::Workflow.execute_activity(
      ChargeOrder, order_id, start_to_close_timeout: 30
    )
    Temporalio::Workflow.execute_activity(
      FulfillOrder, order_id, start_to_close_timeout: 30
    )
  end
end
```

--------------------------------

### Define and start a Hatchet worker

Source: https://docs.hatchet.run/v1/from-celery-to-hatchet

Programmatic worker definition in Hatchet, explicitly registering workflows and setting execution slots.

```python
def start_worker() -> None:
    worker = hatchet.worker("image-worker", slots=4, workflows=[process_image])
    worker.start()
```

--------------------------------

### Initialize shared application client

Source: https://docs.hatchet.run/v1/from-celery-to-hatchet

Comparison of initializing a Celery app versus a Hatchet client.

```python
from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379", backend="redis://localhost:6379")
```

```python
from hatchet_sdk import Hatchet

hatchet = Hatchet()
```

--------------------------------

### Define Hatchet Durable Task for Order Processing

Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet

Examples of Hatchet durable tasks that replicate the sequential execution logic of the original Temporal workflows.

```python
@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)
```

```typescript
export const processOrder = hatchet.durableTask({
  name: 'ProcessOrder',
  executionTimeout: '10m',
  fn: async (input: OrderInput): Promise => {
    await validateOrder.run(input);
    await chargeOrder.run(input);

    return fulfillOrder.run(input);
  },
});
```

```go
processOrder := client.NewStandaloneDurableTask("ProcessOrder",
	func(ctx hatchet.DurableContext, input OrderInput) (FulfillOutput, error) {
		if _, err := validateOrder.Run(ctx, input); err != nil {
			return FulfillOutput{}, fmt.Errorf("validating order: %w", err)
		}

		if _, err := chargeOrder.Run(ctx, input); err != nil {
			return FulfillOutput{}, fmt.Errorf("charging order: %w", err)
		}

		result, err := fulfillOrder.Run(ctx, input)
		if err != nil {
			return FulfillOutput{}, fmt.Errorf("fulfilling order: %w", err)
		}

		var fulfilled FulfillOutput
		if err := result.Into(&fulfilled); err != nil {
			return FulfillOutput{}, fmt.Errorf("decoding fulfillment: %w", err)
		}

		return fulfilled, nil
	},
)
```

--------------------------------

### Migrate from V1 Generics SDK to V1 Reflection SDK

Source: https://docs.hatchet.run/v1/migrating/migration-guide-go

Transition from the deprecated V1 Generics SDK factory-based task creation to the current Reflection SDK.

```go
package main

import (
	"log"
	"strings"

	"github.com/hatchet-dev/hatchet/pkg/client/create"
	v1 "github.com/hatchet-dev/hatchet/pkg/v1"
	"github.com/hatchet-dev/hatchet/pkg/v1/factory"
	"github.com/hatchet-dev/hatchet/pkg/v1/worker"
	"github.com/hatchet-dev/hatchet/pkg/v1/workflow"
	v0Worker "github.com/hatchet-dev/hatchet/pkg/worker"
)

func V1Old() {
	hatchet, err := v1.NewHatchetClient()
	if err != nil {
		log.Fatal(err)
	}

	simple := factory.NewTask(
		create.StandaloneTask{Name: "simple-task", OnEvents: []string{"user:create"}},
		func(ctx v0Worker.HatchetContext, input SimpleInput) (*SimpleResult, error) {
			return &SimpleResult{TransformedMessage: strings.ToLower(input.Message)}, nil
		},
		hatchet,
	)

	_, err = hatchet.Worker(worker.WorkerOpts{
		Name:      "worker",
		Workflows: []workflow.WorkflowBase{simple},
	})
	if err != nil {
		log.Fatal(err)
	}
}
```

```go
package main

import (
	"log"
	"strings"

	hatchet "github.com/hatchet-dev/hatchet/sdks/go"
)

type SimpleInput struct {
	Message string `json:"message"`
}

type SimpleResult struct {
	TransformedMessage string `json:"result"`
}

func V1() {
	client, err := hatchet.NewClient()
	if err != nil {
		log.Fatal(err)
	}

	workflow := client.NewStandaloneTask("simple-workflow", func(ctx hatchet.Context, input SimpleInput) (SimpleResult, error) {
		log.Println("executed step1")
		return SimpleResult{TransformedMessage: strings.ToLower(input.Message)}, nil
	}, hatchet.WithWorkflowEvents("user:create"))

	_, err = client.NewWorker(
		"worker",
		hatchet.WithWorkflows(workflow),
	)
	if err != nil {
		log.Fatal(err)
	}
}
```

--------------------------------

### Run Hatchet Worker

Source: https://docs.hatchet.run/v1/quickstart

Starts the worker process locally.

```sh
hatchet worker dev
```

--------------------------------

### Migrate from V0 SDK to V1 Reflection SDK

Source: https://docs.hatchet.run/v1/migrating/migration-guide-go

Compare the legacy V0 worker registration pattern with the current V1 Reflection SDK approach.

```go
package main

import (
	"log"

	"github.com/hatchet-dev/hatchet/pkg/client"
	v0Worker "github.com/hatchet-dev/hatchet/pkg/worker"
)

func V0() {
	c, err := client.New()
	if err != nil {
		log.Fatal(err)
	}

	worker, err := v0Worker.NewWorker(
		v0Worker.WithClient(c),
		v0Worker.WithName("worker"),
	)
	if err != nil {
		log.Fatal(err)
	}

	err = worker.RegisterWorkflow(
		&v0Worker.WorkflowJob{
			On:   v0Worker.Event("user:create"),
			Name: "simple-workflow",
			Steps: []*v0Worker.WorkflowStep{
				{
					Name: "step1",
					Function: func(ctx v0Worker.HatchetContext) error {
						log.Println("executed step1")
						return nil
					},
				},
			},
		},
	)
	if err != nil {
		log.Fatal(err)
	}
}
```

```go
package main

import (
	"log"
	"strings"

	hatchet "github.com/hatchet-dev/hatchet/sdks/go"
)

type SimpleInput struct {
	Message string `json:"message"`
}

type SimpleResult struct {
	TransformedMessage string `json:"result"`
}

func V1() {
	client, err := hatchet.NewClient()
	if err != nil {
		log.Fatal(err)
	}

	workflow := client.NewStandaloneTask("simple-workflow", func(ctx hatchet.Context, input SimpleInput) (SimpleResult, error) {
		log.Println("executed step1")
		return SimpleResult{TransformedMessage: strings.ToLower(input.Message)}, nil
	}, hatchet.WithWorkflowEvents("user:create"))

	_, err = client.NewWorker(
		"worker",
		hatchet.WithWorkflows(workflow),
	)
	if err != nil {
		log.Fatal(err)
	}
}
```

--------------------------------

### GET /metrics

Source: https://docs.hatchet.run/v1/worker-healthchecks

Returns metrics for the worker, intended for use by monitoring systems like Prometheus.

```APIDOC
## GET /metrics

### Description
Returns metrics for the worker, intended for use by monitoring systems like Prometheus.

### Method
GET

### Endpoint
/metrics

### Response
#### Success Response (200)
- **metrics** (text/plain) - Prometheus-formatted metrics data
```

--------------------------------

### Define a GPU-enabled Dockerfile

Source: https://docs.hatchet.run/v1/compute/gpu

A base Dockerfile configuration for installing CUDA libraries and setting up an application entrypoint.

```dockerfile
# Base image
FROM ubuntu:22.04

# Install CUDA and required libraries
RUN apt-get update && apt-get install -y --no-install-recommends \
    cuda-nvcc-12-2 \
    libcublas-12-2 \
    libcudnn8 \
    && rm -rf /var/lib/apt/lists/*

# Your application setup
WORKDIR /app
COPY . .

# Application entrypoint
CMD ["python", "worker.py"]
```

--------------------------------

### GET /api/v1/tenants/{tenantId}/prometheus-metrics

Source: https://docs.hatchet.run/v1/prometheus-metrics

Retrieves Prometheus metrics for a specific tenant in Prometheus Text Format.

```APIDOC
## GET /api/v1/tenants/{tenantId}/prometheus-metrics

### Description
Retrieves metrics for individual tenants in Prometheus Text Format, which can be scraped by services like Grafana or DataDog.

### Method
GET

### Endpoint
/api/v1/tenants/{tenantId}/prometheus-metrics

### Parameters
#### Path Parameters
- **tenantId** (string) - Required - The unique identifier of the tenant.

### Authentication
- **Authorization** (header) - Required - Bearer token authentication using a valid API token.

### Response
#### Success Response (200)
- **Content-Type** (text/plain) - Returns metrics in standard Prometheus Text Format, including HELP comments, TYPE declarations, and metric samples.
```

--------------------------------

### Instantiate workflows and tasks with the factory pattern

Source: https://docs.hatchet.run/v1/migrating/v1-sdk-improvements

Use the factory pattern to define and execute workflows and tasks with full type safety.

```go
type SimpleInput struct {
    Message string
}

type SimpleResult struct {
    TransformedMessage string
}

simple := factory.NewTask(
    create.StandaloneTask{
        Name: "simple-task",
    }, func(ctx worker.HatchetContext, input SimpleInput) (*SimpleResult, error) {
        return &SimpleResult{
            TransformedMessage: strings.ToLower(input.Message),
        }, nil
    },
    hatchet, // a Hatchet client instance
)

// somewhere else in your code
result, err := simple.Run(ctx, SimpleInput{
    Message: "Hello, World!",
})

// result is fully typed!
```

--------------------------------

### Define Workflow in Ruby

Source: https://docs.hatchet.run/v1/advanced-assignment/worker-affinity

Initialize a workflow using the Ruby SDK.

```ruby
AFFINITY_WORKER_WORKFLOW = HATCHET.workflow(name: "AffinityWorkflow")
```

--------------------------------

### Configure CUDA version compatibility

Source: https://docs.hatchet.run/v1/compute/gpu

Ensure the installed CUDA toolkit version matches the host driver requirements.

```dockerfile
# Ensure CUDA toolkit version matches driver
RUN apt-get install -y cuda-nvcc-12-2
```

--------------------------------

### Import Go OpenTelemetry package

Source: https://docs.hatchet.run/v1/opentelemetry

Import the Hatchet OpenTelemetry package for Go applications.

```go
import hatchetotel "github.com/hatchet-dev/hatchet/sdks/go/opentelemetry"
```

--------------------------------

### Initialize a Worker with Middleware in TypeScript

Source: https://docs.hatchet.run/v1/middleware

Demonstrates creating a worker using a middleware-enabled client and registering workflows.

```typescript
import { hatchetWithMiddleware } from './client';
import { taskWithMiddleware } from './workflow';

async function main() {
  const worker = await hatchetWithMiddleware.worker('task-with-middleware', {
    workflows: [taskWithMiddleware],
  });

  await worker.start();
}

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

--------------------------------

### Implement multi-stage Docker builds

Source: https://docs.hatchet.run/v1/compute/gpu

Use multi-stage builds to separate build-time dependencies from the final runtime image.

```dockerfile
# Build stage
FROM ubuntu:22.04 AS builder
RUN apt-get update && apt-get install -y cuda-nvcc-12-2

# Runtime stage
FROM ubuntu:22.04
COPY --from=builder /usr/local/cuda-12.2 /usr/local/cuda-12.2
```

--------------------------------

### Manual Slot Release Implementation

Source: https://docs.hatchet.run/v1/advanced-assignment/manual-slot-release

Examples of using the release_slot method to free worker capacity after resource-heavy operations.

```python
slot_release_workflow = hatchet.workflow(name="SlotReleaseWorkflow")


@slot_release_workflow.task()
def step1(input: EmptyModel, ctx: Context) -> dict[str, str]:
    print("RESOURCE INTENSIVE PROCESS")
    time.sleep(10)

    # 👀 Release the slot after the resource-intensive process, so that other steps can run
    ctx.release_slot()

    print("NON RESOURCE INTENSIVE PROCESS")
    return {"status": "success"}
```

```go
_ = workflow.NewTask("step1", func(ctx hatchet.Context, _ any) (*StepOutput, error) {
	fmt.Println("RESOURCE INTENSIVE PROCESS")
	time.Sleep(10 * time.Second)

	// Release the slot after the resource-intensive process,
	// so that other steps can run on this worker.
	if releaseErr := ctx.ReleaseSlot(); releaseErr != nil {
		return nil, fmt.Errorf("failed to release slot: %w", releaseErr)
	}

	fmt.Println("NON RESOURCE INTENSIVE PROCESS")

	return &StepOutput{Status: "success"}, nil
})
```

```ruby
SLOT_RELEASE_WORKFLOW = HATCHET.workflow(name: "SlotReleaseWorkflow")

SLOT_RELEASE_WORKFLOW.task(:step1) do |input, ctx|
  puts "RESOURCE INTENSIVE PROCESS"
  sleep 10

  # Release the slot after the resource-intensive process, so that other steps can run
  ctx.release_slot

  puts "NON RESOURCE INTENSIVE PROCESS"
  { "status" => "success" }
end
```
Tokens:
Raw