Task logs

Hatchet captures log output from your task functions and surfaces it in the dashboard alongside the run that produced it. You can log using Python’s standard logging module, print, or the structured ctx.log() method. All three approaches are visible in the task run detail view.

Logging from a task

Any output written to stdout — via print or the standard logging module — is captured by Hatchet and stored against the task run. No extra configuration is required.

Python
TypeScript
Go

The following example uses Python’s standard logging module inside a task. Hatchet intercepts the output and attaches it to the run.

workflow.py

import logging
import time

from hatchet_sdk import Context, EmptyModel, Hatchet

hatchet = Hatchet()
logger = logging.getLogger(__name__)

logging_workflow = hatchet.workflow(name="LoggingWorkflow")

@logging_workflow.task()
def root_logger(input: EmptyModel, ctx: Context) -> dict[str, str]:
    for i in range(12):
        logger.info(f"executed step1 - {i}")
        logger.info({"step1": "step1"})
        time.sleep(0.1)
    return {"status": "success"}

To make the root logger available to Hatchet’s internal client, pass it through ClientConfig:

client.py

import logging
from hatchet_sdk import ClientConfig, Hatchet

logging.basicConfig(level=logging.INFO)
root_logger = logging.getLogger()

hatchet = Hatchet(
    debug=True,
    config=ClientConfig(
        logger=root_logger,
    ),
)

Use console.log or any logging library inside your task function. Hatchet captures stdout.

workflow.ts

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

const loggingWorkflow = hatchet.workflow({ name: 'LoggingWorkflow' });

loggingWorkflow.task({
  name: 'root-logger',
  fn: async (input) => {
    for (let i = 0; i < 12; i++) {
      console.log(`executed step - ${i}`);
    }
    return { status: 'success' };
  },
});

Use the standard log package or any structured logger. Hatchet captures stdout output from the worker process.

worker.go

package main

import (
    "log"
    "github.com/hatchet-dev/hatchet/pkg/worker"
)

func myTask(ctx worker.HatchetContext) error {
    for i := 0; i < 12; i++ {
        log.Printf("executed step - %d", i)
    }
    return nil
}

Structured logging with `ctx.log()`

ctx.log() sends a log line directly to the Hatchet API and associates it with the current task run. Unlike stdout capture, ctx.log() lines are immediately visible in the dashboard as the task runs. You can pass a plain string or any JSON-serializable mapping.

Python
TypeScript
Go

workflow.py

import time
from hatchet_sdk import Context, EmptyModel, Hatchet

hatchet = Hatchet()
logging_workflow = hatchet.workflow(name="LoggingWorkflow")

@logging_workflow.task()
def context_logger(input: EmptyModel, ctx: Context) -> dict[str, str]:
    for i in range(12):
        ctx.log(f"executed step1 - {i}")
        ctx.log({"step1": "step1"})
        time.sleep(0.1)
    return {"status": "success"}

ctx.log() accepts:

A str — sent as-is
A dict or any JSON-serializable mapping — automatically serialized to JSON before sending

The call returns immediately and does not block the task. If the log write fails, it is silently dropped by default. Pass raise_on_error=True to surface errors:

ctx.log("important event", raise_on_error=True)

loggingWorkflow.task({
  name: 'context-logger',
  fn: async (input, ctx) => {
    for (let i = 0; i < 12; i++) {
      ctx.log(`executed step - ${i}`);
      ctx.log({ step: 'step1' });
    }
    return { status: 'success' };
  },
});

The Go SDK does not expose a ctx.Log() method directly. Use stdout logging — the worker captures and forwards it to Hatchet.

Log lines sent via ctx.log() are stored in Hatchet’s database with the task run. They are separate from your application’s logging pipeline and do not appear in external log aggregators unless you also write to stdout.

Reading logs via the API

The hatchet.logs client lets you retrieve log lines for any task run programmatically. This is useful for post-run analysis, surfacing logs in your own UI, or exporting to an external store.

Python
Go

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

hatchet = Hatchet()

# Retrieve up to 1000 log lines for a task run
logs = hatchet.logs.list(task_run_id="<task-run-id>")

for line in logs.rows:
    print(line.created_at, line.message)

Async variant:

logs = await hatchet.logs.aio_list(task_run_id="<task-run-id>")

import (
    "context"
    "github.com/google/uuid"
    "github.com/hatchet-dev/hatchet/pkg/client/rest"
)

taskRunId := uuid.MustParse("<task-run-id>")
logs, err := hatchet.Logs().List(ctx, taskRunId, &rest.V1LogLineListParams{})

Filtering logs

LogsClient.list supports filtering by time range and by the number of lines returned.

Python
Go

from datetime import datetime, timezone

logs = hatchet.logs.list(
    task_run_id="<task-run-id>",
    limit=200,
    since=datetime(2025, 1, 1, tzinfo=timezone.utc),
    until=datetime(2025, 1, 2, tzinfo=timezone.utc),
)

Parameter	Type	Default	Description
`task_run_id`	`str`	required	The ID of the task run to fetch logs for.
`limit`	`int`	`1000`	Maximum number of log lines to return.
`since`	`datetime`	`None`	Only return logs after this timestamp.
`until`	`datetime`	`None`	Only return logs before this timestamp.

import "time"

since := time.Now().Add(-1 * time.Hour)
limit := 200

logs, err := hatchet.Logs().List(ctx, taskRunId, &rest.V1LogLineListParams{
    Since: &since,
    Limit: &limit,
})

Correlating logs with task runs

Each log line returned by logs.list includes a created_at timestamp and the message text. To correlate logs with run metadata, combine the logs API with the runs API:

Python

from hatchet_sdk import Hatchet

hatchet = Hatchet()

# Get the task run to find its workflow run
task_run = hatchet.runs.get_task_run(task_run_id)

# Fetch logs for that specific task run
logs = hatchet.logs.list(task_run_id=task_run_id)

print(f"Task: {task_run.display_name}, Status: {task_run.status}")
for line in logs.rows:
    print(f"  [{line.created_at}] {line.message}")

Get Started

Running Tasks

Flow Control

Workers

Observability

Self-Hosting

Logging from a task

Structured logging with `ctx.log()`

Reading logs via the API

Filtering logs

Correlating logs with task runs

Next steps

Dashboard

Alerting

Build docs developers (and LLMs) love

Get Started

Running Tasks

Flow Control

Workers

Observability

Self-Hosting

​Logging from a task

​Structured logging with ctx.log()

​Reading logs via the API

​Filtering logs

​Correlating logs with task runs

​Next steps

Dashboard

Alerting

Build docs developers (and LLMs) love

Logging from a task

Structured logging with `ctx.log()`

Reading logs via the API

Filtering logs

Correlating logs with task runs

Next steps