dagu

README

Community

Dagu

Dagu is a local-first workflow engine for ops automation and AI-assisted operations. It is open source and self-hostable: a single binary with a built-in Web UI, no external database or message broker, running on Linux / Mac / Windows. Define DAGs in a declarative YAML format. It natively supports shell commands, Docker containers, Kubernetes Jobs, remote commands via SSH, external coding-agent CLIs through harness.run, and more through Dagu Actions.

Dagu turns existing scripts, runbooks, and agent-driven jobs into production workflows with scheduling, retries, approvals, and run history. It runs where your data and credentials live: on-prem, air-gapped, edge, or cloud, and scales from a single node to a distributed worker fleet.

Highlights:

• Single binary file installation.
• Declarative YAML format for defining DAGs.
• Web UI for visually managing, retrying, and monitoring pipelines.
• Use existing scripts or tools without any modifications.
• Self-contained, with no need for a DBMS.
• Built-in MCP support for AI agents to manage workflows.
• Run external coding-agent CLIs through harness.run when workflows need AI assistance.

Quick Look

For a quick look at how workflows are defined, see the examples.

Run Details	Step Logs

Try it live: Live Demo (credentials: demouser / demouser)

Why Dagu?

  Traditional Orchestrator          Dagu
  ┌────────────────────────┐        ┌──────────────────┐
  │  Web Server            │        │                  │
  │  Scheduler             │        │  dagu start-all  │
  │  Worker(s)             │        │                  │
  │  PostgreSQL            │        └──────────────────┘
  │  Redis / RabbitMQ      │         Single binary.
  │  Python Runtime        │         Self-hosted.
  └────────────────────────┘         Adds scheduling, retries, and approvals around existing automation.
    6+ services to manage

Performance

Dagu stores state in local files and reaches production throughput without external services.

• Throughput: A single machine can run thousands of workflow runs per day. Actual capacity depends on CPU, memory, disk, and workflow shape.
• Load control: Queues, concurrency limits, and resource limits control how many runs execute at once and where they run.
• Scale out: Distributed workers spread execution across machines when one node is not enough.

Real-World Use Cases

Use Case	How Dagu Helps
ETL and data operations	Turn data extraction scripts, SQL queries, dbt commands, and data-processing runbooks into observable pipelines with durable execution.
Legacy scripts and scheduled jobs	Turn complex jobs with interdependencies into maintainable DAGs with a UI, automatic logging, retries, and notifications instead of opaque cron jobs and bash scripts.
Media conversion	Run ffmpeg for video transcoding and format conversion. Thanks to Dagu's file-backed nature, workers can run heavy conversions in parallel without single machine bottlenecks or external databases.
Infrastructure and server automation	Run any command or script over SSH on remote servers, keeping logs, results, and notifications in one place.
GitHub-driven workflows	Trigger workflows from GitHub events. This is useful for running automation on private infrastructure without exposing your servers to the public internet.
Container and Kubernetes workflows	Run Docker containers and Kubernetes Jobs as steps in your workflows without building a custom control plane around containers.
Customer support automation	Run self-service support tools that non-engineering teams can use to run approved workflows for running diagnostics, querying databases, and performing common support tasks without escalating to engineering.
IoT and edge workflows	Run sensor polling, local ML inference, data preprocessing, backups, offline sync, health checks, etc. Dagu keeps these jobs close to the data source while still providing Web UI visibility.

Quick Start

Install

macOS/Linux:

curl -fsSL https://raw.githubusercontent.com/dagucloud/dagu/main/scripts/installer.sh | bash

Homebrew:

brew install dagu

Windows (PowerShell):

irm https://raw.githubusercontent.com/dagucloud/dagu/main/scripts/installer.ps1 | iex

Docker:

docker run --rm -v ~/.dagu:/var/lib/dagu -p 8080:8080 ghcr.io/dagucloud/dagu:latest dagu start-all

Kubernetes (Helm):

helm repo add dagu https://dagucloud.github.io/dagu
helm repo update
helm install dagu dagu/dagu --set persistence.storageClass=<your-rwx-storage-class>

Replace <your-rwx-storage-class> with a StorageClass that supports ReadWriteMany. See charts/dagu/README.md for chart configuration.

The script installers run a guided wizard that can add Dagu to your PATH, set it up as a background service, and create the initial admin account. Homebrew, npm, Docker, and Helm install without the wizard. See the Installation documentation for all options.

Create and run a workflow

Create hello.yaml:

steps:
  - id: hello
    run: echo "hello from Dagu"

Run the workflow with:

dagu start hello.yaml

Start the server

dagu start-all

Visit http://localhost:8080

Connect AI agents through MCP

Dagu exposes a built-in MCP server from the running HTTP server. Start Dagu, then configure MCP-capable chat or coding agents to use the Streamable HTTP endpoint:

http://localhost:8080/mcp

Use MCP when you want an AI agent to read Dagu state, preview or apply workflow changes, and start, enqueue, retry, or stop runs through dagu_read, dagu_change, and dagu_execute. See the MCP setup guide.

For authoring-only help in Claude Code, Codex, Gemini CLI, and other AI coding tools, install the Dagu workflow authoring skill:

gh skill install dagucloud/dagu dagu

How You Run Dagu?

Run Dagu on one machine, scale out with distributed workers, or use a managed Dagu instance operated by us. See the Deployment Models guide.

Local Single-Server	Self-Hosted
Managed Server	Hybrid

Model	Server	Execution	Best for
Local single-server	dagu start-all on one machine.	Same machine.	Development, single-machine scheduled workloads, edge jobs, and internal automation.
Self-hosted	Dagu server on your infrastructure.	Local execution or distributed workers on your infrastructure.	Teams that need ownership of infrastructure.
Managed Server	Full managed Dagu server in a dedicated, isolated gVisor instance on GKE.	Managed instance.	Teams that want Dagu operated for them without running the server themselves.
Hybrid	Full managed Dagu server.	Private workers in your infrastructure over mTLS.	Docker steps, private networks, specialized hardware, or data-local work.

Licensing

• Community self-host: No license key required. You operate the server, storage, upgrades, networking, and workers. Start with the installation guide.
• Self-host license: Adds SSO, RBAC, audit logging, and incident SaaS integration to Dagu. See self-host licensing.
• Dagu managed instance: Includes its own managed license. Private workers can run on your infrastructure.

Managed Dagu instances do not expose a Docker daemon or Docker socket. Workflows that need Docker step execution should use self-hosted Dagu or a private worker with Docker access.

Key Features

• Observability: Shared workflows and scheduling with clear visualizations, status tracking, and logs in the Web UI.
• Language-agnostic: No framework required. Define workflow steps using shell commands, Docker containers, Kubernetes Jobs, SQL queries, HTTP requests, and any other tool via official and third-party Dagu Actions.
• Reproducibility: Reproducible runs with pinned tools, plus automatic installation and caching on workers—eliminating the need to manually install dependencies on the server or workers.
• Built-in Approvals: The Human-in-the-loop steps for manual approvals, review, and intervention in any workflow.
• MCP Server: Built-in MCP server for authoring and running workflows via AI agents like Claude Code, Codex, Gemini CLI, Pi, OpenCode, and more.
• External CLI Harness: You can run coding-agent CLIs (Claude Code, Codex, Gemini CLI, Pi, OpenCode, etc.) with a built-in harness action or custom harness definition.
• Secret management: Built-in secret management with secure log masking, preventing credentials from leaking into logs or the Web UI.
• Self-host or managed: Self-hosted via a single binary that runs on Linux, macOS, and Windows. Includes an optional distributed worker mode for scaling out execution across machines.
• Permission Control: RBAC and SSO support for team environments, controlling who can view, run, and edit workflows through granular permissions and audit logging.

Architecture

Dagu can run in three configurations:

Standalone: A single dagu start-all process runs the HTTP server, scheduler, and executor. Suitable for single-machine deployments.

Coordinator/Worker: The scheduler enqueues jobs to a local file-based queue, then dispatches them to a coordinator over gRPC. Workers long-poll the coordinator for tasks, execute DAGs locally, and report status back. Workers can run on separate machines and are routed tasks based on labels.

Headless: Run without the web UI (DAGU_HEADLESS=true). Useful for CI/CD environments or when Dagu is managed through the CLI or API only.

Standalone:

  ┌─────────────────────────────────────────┐
  │  dagu start-all                         │
  │  ┌───────────┐ ┌───────────┐ ┌────────┐ │
  │  │ HTTP / UI │ │ Scheduler │ │Executor│ │
  │  └───────────┘ └───────────┘ └────────┘ │
  │  File-based storage (logs, state, queue)│
  └─────────────────────────────────────────┘

Distributed:

  ┌────────────┐                   ┌────────────┐
  │ Scheduler  │                   │ HTTP / UI  │
  │            │                   │            │
  │ ┌────────┐ │                   └─────┬──────┘
  │ │ Queue  │ │  Dispatch (gRPC)        │ Dispatch / GetWorkers
  │ │(file)  │ │─────────┐               │ (gRPC)
  │ └────────┘ │         │               │
  └────────────┘         ▼               ▼
                    ┌─────────────────────────┐
                    │      Coordinator        │
                    │  ┌───────────────────┐  │
                    │  │ Dispatch Task     │  │
                    │  │ Store (pending/   │  │
                    │  │ claimed)          │  │
                    │  └───────────────────┘  │
                    └────────▲────────────────┘
                             │
                   Worker poll / task response
                   Heartbeat / ReportStatus /
                   StreamLogs (gRPC)
                             │
               ┌─────────────┴─────────────┐
               │             │             │
          ┌────┴───┐    ┌────┴───┐    ┌────┴───┐
          │Worker 1│    │Worker 2│    │Worker N│ Sandbox execution of DAGs
          │        │    │        │    │        │
          └────────┘    └────────┘    └────────┘

Parameter Definition

Workflows can define parameters that render as typed input forms in the Web UI and can be referenced by steps.

params:
  - name: customer_id
    type: string
    description: Customer or account identifier
  - name: change_scope
    type: string
    description: What the repair is allowed to change
    enum:
      - metadata_only
      - permissions
      - full_account
    default: metadata_only
  - name: dry_run
    type: boolean
    default: true

steps:
  - id: extract
    run: >-
      ./scripts/extract.sh
      --customer "${params.customer_id}"
      --scope "${params.change_scope}"
      --dry-run="${params.dry_run}"
    retry_policy:
      limit: 3
      interval_sec: 30

Workflow Examples

Parallel executions

steps:
  - id: extract
    run: ./extract.sh

  - id: transform_a
    run: ./transform_a.sh
    depends: extract

  - id: transform_b
    run: ./transform_b.sh
    depends: extract

  - id: load
    run: ./load.sh
    depends: [transform_a, transform_b]

%%{init: {'theme': 'base', 'themeVariables': {'background': '#18181B', 'primaryTextColor': '#fff', 'lineColor': '#888'}}}%%
graph LR
    A[extract] --> B[transform_a]
    A --> C[transform_b]
    B --> D[load]
    C --> D
    style A fill:#18181B,stroke:#22C55E,stroke-width:1.6px,color:#fff
    style B fill:#18181B,stroke:#22C55E,stroke-width:1.6px,color:#fff
    style C fill:#18181B,stroke:#22C55E,stroke-width:1.6px,color:#fff
    style D fill:#18181B,stroke:#3B82F6,stroke-width:1.6px,color:#fff

External tools with pinning and caching

tools:
  - jqlang/jq@jq-1.7.1

steps:
  - id: inspect
    run: jq --version

  - id: summarize
    action: python-script@v1
    with:
      input:
        rows: [42, 8]
      script: |
        return {"total": sum(input["rows"])}

Dagu installs declared portable CLIs before the DAG run, exposes them on PATH for host command steps, and caches them on each worker. Tool provisioning uses aqua as the default provider. See the Tools documentation and Dagu Actions for more details.

Third-party Dagu Actions

params:
  - BUILD_ID

steps:
  - id: notify
    action: acme/dagu-action-notify@v1.2.0
    with:
      text: "Build ${params.BUILD_ID} finished"

  - id: audit
    depends: notify
    run: echo "Notification result: ${notify.outputs.messageId}"

A third-party Dagu Action package contains a DAG, manifest, schemas, and helper files behind an action: reference. See the Dagu Actions and Third-Party Actions documentation for details.

Docker step

steps:
  - name: build
    container:
      image: node:20-alpine
    run: npm run build

Kubernetes Pod execution

steps:
  - name: batch-job
    action: kubernetes.run
    with:
      namespace: production
      image: my-registry/batch-processor:latest
      resources:
        requests:
          cpu: "2"
          memory: "4Gi"
      command: ./process.sh

SSH remote execution

steps:
  - name: deploy
    action: ssh.run
    with:
      host: prod-server.example.com
      user: deploy
      key: ~/.ssh/id_rsa
      command: cd /var/www && git pull && systemctl restart app

Sub-DAG composition

steps:
  - name: extract
    action: dag.run
    with:
      dag: etl/extract
      params:
        SOURCE: s3://bucket/data.csv

  - name: transform
    action: dag.run
    with:
      dag: etl/transform
      params:
        INPUT: ${extract.outputs.result}
    depends: extract

  - name: load
    action: dag.run
    with:
      dag: etl/load
      params:
        DATA: ${transform.outputs.result}
    depends: transform

---

# You can include multiple DAGs in the same YAML file, or reference DAGs defined in separate files.
name: etl/extract

params:
  - SOURCE

tools:
  - aws/aws-cli@2.11.14

steps:
  - name: download
    run: aws s3 cp ${params.SOURCE} data.csv
    outputs:
      result: data.csv

Retry and error handling

steps:
  - name: flaky-api-call
    run: curl -f https://api.example.com/data
    retry_policy:
      limit: 3
      interval_sec: 10
    continue_on:
      failure: true

Scheduling with overlap control and catch-up

schedule:
  - "0 */6 * * *"          # Every 6 hours
overlap_policy: skip       # Skip if previous run is still active
catchup_window: "5h"       # Catch up missed runs when scheduler is down for up to 5 hours
  
timeout_sec: 3600
handler_on:
  failure:
    run: notify-team.sh
  exit:
    run: cleanup.sh

External coding-agent CLI harness step with manual approval

steps:
  - id: review
    action: harness.run
    with:
      provider: codex
      prompt: Review the README.md file and return concise Markdown findings.
    stdout:
      artifact: review.md

  - id: approval
    action: noop
    depends: review
    approval:
      prompt: Review the review.md artifact. Approve to post an issue with the findings, or reject to skip.

  - id: read_review
    action: artifact.read
    depends: approval
    with:
      path: review.md

  - id: post_issue
    run: gh issue create --title "Review Findings" --body-file "${read_review.stdout}"
    depends: read_review

LLM-driven workflow where the model chooses each step

type: controller

env:
  - ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}

llm:
  provider: anthropic
  model: claude-sonnet-4-20250514
  system: |
    If the tests fail, run the fixer and test again. Never ask for sign-off on
    a red build.

steps:
  - name: run_tests
    description: Run the test suite.
    run: |
      test -f /tmp/dagu-flaky-cleared || { echo "FAIL: flaky_test timed out"; exit 1; }
      echo "all green"

  - name: fix_flaky
    description: Quarantine the known flaky test.
    run: |
      touch /tmp/dagu-flaky-cleared
      echo "quarantined flaky_test"

  - id: sign_off
    name: sign_off
    action: human.task
    with:
      prompt: Tests are green. Ship it?
      form:
        type: object
        properties:
          approved: { type: boolean }
        required: [approved]

tasks:
  - name: tests_green
    description: Finished when run_tests succeeded on its most recent run.
  - name: approved
    description: Finished when a person answered sign_off with approved=true.

With type: controller the steps stop being a plan and become a catalog of actions; tasks declares the goals and ends the run once all are settled. The model decides what runs next, so a failing test can be fixed and retried without that path being wired in advance. When it opens the human task the run releases its worker slot and resumes on the same run once someone answers.

For more examples, see the Examples documentation.

Built-in Actions

Dagu includes built-in actions that run within the Dagu process or on the selected worker. Local shell commands use the run: field; structured work uses action:.

Action	Purpose
run: field	Local shell commands and scripts (bash, sh, PowerShell, custom shells)
exec	Direct process execution without shell parsing
noop	Output-only or approval-only placeholder step
log.write	Write structured log messages
docker.run / container.run	Run containers with registry auth, volume mounts, and resource limits
kubernetes.run / k8s.run	Execute Kubernetes Jobs with namespace, image, and resource settings
ssh.run	Remote command execution over SSH
sftp.upload / sftp.download	File transfer over SFTP
http.request	HTTP requests with headers, auth, and request bodies
chat.completion	Run an LLM chat completion step
harness.run	Run external coding-agent CLIs such as Claude Code, Codex, Copilot, OpenCode, and Pi
postgres.query / postgres.import	PostgreSQL queries and imports
sqlite.query / sqlite.import	SQLite queries and imports
redis.<operation>	Redis commands, pipelines, and Lua scripts
s3.upload / s3.download / s3.list / s3.delete	Upload, download, list, and delete S3 objects
file.stat / file.read / file.write / file.copy / file.move / file.delete / file.mkdir / file.list	Local file operations without shell commands
artifact.write / artifact.read / artifact.list	Write, read, and list DAG-run artifacts
state.get / state.set / state.delete / state.list / state.diff	Persistent JSON state across DAG runs
data.convert / data.pick	Convert and select structured data
jq.filter	JSON transformation using jq expressions
archive.create / archive.extract / archive.list	Create, extract, and list zip/tar archives
wait.duration / wait.until / wait.file / wait.http	Wait for time, file state, or HTTP readiness
human.task	Wait for acknowledgement or typed operator input before downstream steps continue
mail.send	Send email via SMTP
template.render	Text generation with template rendering
router.route	Conditional step routing based on values and patterns
dag.run	Invoke another DAG as a sub-workflow with params and dependencies
dag.enqueue	Queue another DAG asynchronously and continue after enqueue
git.checkout	Clone or update Git repositories
outputs.write	Publish DAG or Dagu Action outputs for callers

Custom Actions

Custom Actions are inline reusable wrappers defined with the top-level actions field. They expand to built-in actions during DAG load, so you can wrap a common shell, HTTP, SQL, or other pattern behind a typed interface with validated input.

actions:
  webhook.send:
    input_schema:
      type: object
      additionalProperties: false
      required: [url, text]
      properties:
        url:
          type: string
        text:
          type: string
    template:
      action: http.request
      with:
        method: POST
        url: {{ .input.url }}
        headers:
          Content-Type: application/json
        body: |
          {"text": {{ json .input.text }}}

steps:
  - action: webhook.send
    with:
      url: https://hooks.example.com/ops
      text: deploy complete

See Custom Actions and the YAML Specification for the exact actions, action, and run field behavior.

Official Dagu Actions

Dagu Actions are official action packages maintained in the dagucloud GitHub organization. They use the same action package runtime as third-party action packages, but callers use the short form action: name@version.

Dagu Action	Purpose
node-script@v1	Run small JavaScript transforms or glue code with action-owned Node.js
python-script@v1	Run small Python transforms or glue code with action-owned Python and optional requirements
dbt@v1	Run dbt Core commands with action-owned Python and adapter requirements
duckdb@v1	Run DuckDB SQL through the DuckDB CLI without adding DuckDB to the core binary
ffmpeg@v1	Run FFmpeg conversion, transcoding, probing, and stream-processing tasks
github-cli@v1	Run GitHub issue, pull request, release, repository, and API automation through gh
rclone@v1	Run portable copy, sync, check, list, and storage-management workflows through rclone

Versions are required. Pin production workflows to a version tag or commit SHA. See Official Dagu Actions for the current Dagu Action list and exact input/output contracts.

For non-official packages, use Third-Party Actions such as action: owner/repo@version. They contain a dagu-action.yaml manifest and a DAG entrypoint, run as sub-DAGs, and are transferred to distributed workers as workspace bundles after the reference is resolved. See the documentation for package layout and reference formats.

Security and Access Control

Authentication

Dagu supports three top-level authentication modes, configured via DAGU_AUTH_MODE:

• none — No authentication
• basic — HTTP Basic authentication
• builtin — JWT-based authentication with user management, API keys, per-DAG webhook tokens, and optional OIDC/SSO integration

Role-Based Access Control

When using builtin auth, five roles control access:

Role	Capabilities
admin	Full access including user management
manager	Create, edit, delete, run, stop DAGs; view audit logs
developer	Create, edit, delete, run, stop DAGs
operator	Run and stop DAGs only (no editing)
viewer	Read-only access

API keys can be created with independent role assignments. Audit logging tracks all actions.

TLS and Secrets

• TLS for the HTTP server (DAGU_CERT_FILE, DAGU_KEY_FILE)
• Mutual TLS for gRPC coordinator/worker communication (DAGU_PEER_CERT_FILE, DAGU_PEER_KEY_FILE, DAGU_PEER_CLIENT_CA_FILE)
• Secret management with environment variables, files, Kubernetes Secrets, HashiCorp Vault, and cloud-provider secret stores

Production Hardening

For self-hosted production deployments, treat network exposure and execution boundaries as the primary controls:

• Prefer auth.mode: builtin for any shared or network-exposed instance. Use basic only for simple private setups, and avoid none outside isolated local development.
• Keep metrics: private unless the metrics endpoint is reachable only on a trusted private network.
• Bind Dagu to loopback or a private interface when possible. If you must use 0.0.0.0, place it behind a trusted reverse proxy, TLS, and network-level access controls.
• Leave terminal.enabled: false unless the instance is admin-only and tightly scoped.
• In distributed deployments, set peer.insecure=false and configure peer TLS when coordinator and workers communicate across host or network boundaries.
• Treat Docker socket mounts, root containers, and host-level executors as privileged access to the underlying machine.

See Server Configuration, Docker deployment, and Distributed execution for operator-focused guidance.

Observability

Prometheus Metrics

Dagu exposes Prometheus-compatible metrics:

• dagu_info — Build information (version, Go version)
• dagu_uptime_seconds — Server uptime
• dagu_dag_runs_total — Total DAG runs by status
• dagu_dag_runs_total_by_dag — Per-DAG run counts
• dagu_dag_run_duration_seconds — Histogram of run durations
• dagu_dag_runs_currently_running — Active DAG runs
• dagu_dag_runs_queued_total — Queued runs
• dagu_workers_registered — Registered distributed workers
• dagu_worker_info — Worker heartbeat labels as key/value metadata
• dagu_worker_heartbeat_timestamp_seconds — Last worker heartbeat timestamp
• dagu_worker_health_status — Worker health by heartbeat freshness
• dagu_worker_pollers — Worker poller capacity by state
• dagu_worker_running_tasks — Running tasks per worker
• dagu_worker_oldest_running_task_age_seconds — Age of the oldest running task per worker

Structured Logging

JSON or text format logging (DAGU_LOG_FORMAT). Logs are stored per-run with separate stdout/stderr capture per step.

Notifications

• Email notifications on DAG success, failure, or wait status via SMTP
• Per-DAG webhook endpoints with token authentication

Artifacts

Dagu runs can write arbitrary files under ${context.paths.artifacts_dir} in value-resolved fields, with DAG_RUN_ARTIFACTS_DIR also exposed to step processes. Dagu stores those files per run as Artifacts. In the Web UI, operators can browse the file tree, preview Markdown, text, and image files inline, and download any artifact when they need the raw file.

This is useful for generated reports, screenshots, charts, exported JSON or CSV files, and other outputs that do not fit simple key/value outputs.

See the Artifacts documentation and the Web UI guide for the full artifact browser workflow and screenshots.

Scheduling and Reliability

• Cron scheduling with timezone support and multiple schedule entries per DAG
• Overlap policies: skip (default — skip if previous run is still active), all (queue all), latest (keep only the most recent)
• Catch-up scheduling: Automatically runs missed intervals when the scheduler was down
• Zombie detection: Identifies and handles stalled DAG runs (configurable interval, default 45s)
• Retry policies: Per-step retry with configurable limits, intervals, and exit code filtering
• Human tasks: Pause root DAG runs for acknowledgement or schema-validated operator input, locally or on distributed workers, then expose form values to downstream steps
• Lifecycle hooks: onInit, onSuccess, onFailure, onAbort, onExit, onWait
• Preconditions: Gate DAG or step execution on shell command results
• High availability: Scheduler lock with stale detection for failover

Distributed Execution

The coordinator/worker architecture distributes DAG execution across multiple machines:

• Coordinator: gRPC server that manages task distribution, worker registry, and health monitoring
• Workers: Connect to the coordinator, pull tasks from the queue, execute DAGs locally, report results
• Worker labels: Route DAGs to specific workers based on labels (e.g., gpu=true, region=us-east-1)
• Health checks: HTTP health endpoints on coordinator and workers for load balancer integration
• Queue system: File-based persistent queue with configurable concurrency limits

# Start coordinator
dagu coordinator

# Start workers (on separate machines)
DAGU_WORKER_LABELS=gpu=true,memory=64G dagu worker

See the distributed execution documentation for setup details.

CLI Reference

Command	Description
dagu start <dag>	Execute a DAG
dagu start-all	Start HTTP server + scheduler + coordinator
dagu server	Start HTTP server only
dagu scheduler	Start scheduler only
dagu coordinator	Start coordinator (distributed mode)
dagu worker	Start worker (distributed mode)
dagu stop <dag>	Stop a running DAG
dagu restart <dag>	Restart a DAG
dagu retry --run-id=<run-id> <dag>	Retry a failed run
dagu human-task complete --run-id=<run-id> --step=<id> <dag>	Complete a waiting human task
dagu dry <dag>	Dry run — show what would execute
dagu status <dag>	Show DAG run status
dagu history <dag>	Show execution history
dagu validate <dag>	Validate DAG YAML
dagu enqueue <dag>	Add DAG to the execution queue
dagu dequeue <queue-name> [--dag-run=<dag>:<run-id>]	Remove a DAG-run from the queue
dagu cleanup <dag>	Clean up old run data
dagu version	Show version

Environment Variables

Precedence: Command-line flags > Environment variables > Configuration file (~/.config/dagu/config.yaml)

Embedded Go API (Experimental)

Go applications can import Dagu and start DAG runs from the host process:

import "github.com/dagucloud/dagu"

engine, err := dagu.New(ctx, dagu.Options{
	HomeDir: "/var/lib/myapp/dagu",
})
if err != nil {
	return err
}
defer engine.Close(context.Background())

run, err := engine.RunYAML(ctx, []byte(`
params:
  - MESSAGE
steps:
  - name: hello
    run: echo "${params.MESSAGE}"
`), dagu.WithParams(map[string]string{
	"MESSAGE": "hello from the host app",
}))
if err != nil {
	return err
}

status, err := run.Wait(ctx)
if err != nil {
	return err
}
fmt.Println(status.Status)

The embedded API is experimental and may change. See the embedded API documentation and examples/embedded.

Server

Variable	Default	Description
DAGU_HOST	127.0.0.1	Bind address
DAGU_PORT	8080	HTTP port
DAGU_BASE_PATH	—	Base path for reverse proxy
DAGU_HEADLESS	false	Run without web UI
DAGU_TZ	—	Timezone (e.g., Asia/Tokyo)
DAGU_LOG_FORMAT	text	text or json
DAGU_CERT_FILE	—	TLS certificate
DAGU_KEY_FILE	—	TLS private key
DAGU_CORS_ALLOWED_ORIGINS	—	Comma-separated list of allowed CORS origins (e.g. https://app.example.com). When unset, cross-origin browser access is disabled. Exact origins enable credentials. An explicit * allows every origin without credentials and emits a security warning.

Paths

Variable	Default	Description
DAGU_HOME	—	Overrides all path defaults
DAGU_DAGS_DIR	~/.config/dagu/dags	DAG definitions directory
DAGU_LOG_DIR	~/.local/share/dagu/logs	Log files
DAGU_DATA_DIR	~/.local/share/dagu/data	Application state
DAGU_TOOLS_DIR	{DAGU_DATA_DIR}/tools	Managed DAG tool cache
DAGU_DAG_STATE_DIR	{DAGU_DATA_DIR}/dag-state	Persistent DAG state files

Authentication

Variable	Default	Description
DAGU_AUTH_MODE	builtin	none, basic, or builtin
DAGU_AUTH_BASIC_USERNAME	—	Basic auth username
DAGU_AUTH_BASIC_PASSWORD	—	Basic auth password
DAGU_AUTH_TOKEN_SECRET	(auto)	JWT signing secret
DAGU_AUTH_TOKEN_TTL	24h	JWT token lifetime (maximum: 8760h / 365 days)

OIDC variables: DAGU_AUTH_OIDC_CLIENT_ID, DAGU_AUTH_OIDC_CLIENT_SECRET, DAGU_AUTH_OIDC_ISSUER, DAGU_AUTH_OIDC_SCOPES, DAGU_AUTH_OIDC_WHITELIST, DAGU_AUTH_OIDC_AUTO_SIGNUP, DAGU_AUTH_OIDC_DEFAULT_ROLE, DAGU_AUTH_OIDC_ALLOWED_DOMAINS.

Scheduler

Variable	Default	Description
DAGU_SCHEDULER_PORT	8090	Health check port
DAGU_SCHEDULER_ZOMBIE_DETECTION_INTERVAL	45s	Zombie run detection interval (0 to disable)
DAGU_SCHEDULER_LOCK_STALE_THRESHOLD	30s	HA lock stale threshold
DAGU_QUEUE_ENABLED	true	Enable queue system

Coordinator / Worker

Variable	Default	Description
DAGU_COORDINATOR_HOST	127.0.0.1	Coordinator bind address
DAGU_COORDINATOR_PORT	50055	Coordinator gRPC port
DAGU_COORDINATOR_HEALTH_PORT	8091	Coordinator health check port
DAGU_WORKER_ID	—	Worker instance ID
DAGU_WORKER_MAX_ACTIVE_RUNS	100	Max concurrent runs per worker
DAGU_WORKER_HEALTH_PORT	8092	Worker health check port
DAGU_WORKER_LABELS	—	Worker labels (key=value,key=value)

Peer TLS (gRPC)

Variable	Default	Description
DAGU_PEER_CERT_FILE	—	Peer TLS certificate
DAGU_PEER_KEY_FILE	—	Peer TLS private key
DAGU_PEER_CLIENT_CA_FILE	—	CA for client verification
DAGU_PEER_INSECURE	true	Use h2c instead of TLS

Git Sync

Variable	Default	Description
DAGU_GITSYNC_ENABLED	false	Enable Git sync
DAGU_GITSYNC_REPOSITORY	—	Repository URL
DAGU_GITSYNC_BRANCH	main	Branch to sync
DAGU_GITSYNC_AUTH_TYPE	token	token or ssh
DAGU_GITSYNC_AUTOSYNC_ENABLED	false	Enable periodic auto-pull
DAGU_GITSYNC_AUTOSYNC_INTERVAL	300	Sync interval in seconds

Community

• Discord — Questions and discussion
• GitHub Issues — Bug reports and feature requests
• Bluesky

Development

Prerequisites: Go 1.26+, Node.js, pnpm

git clone https://github.com/dagucloud/dagu.git && cd dagu
make build    # Build frontend + Go binary
make test     # Run tests with race detection
make lint     # Run golangci-lint

See CONTRIBUTING.md for development workflow and code standards.

Acknowledgements

Premium Sponsors

Supporters

Contributing

We welcome contributions of all kinds. See our Contribution Guide for details.

License

GNU GPLv3 - See LICENSE. See LICENSING.md for embedded API and commercial embedding notes.

Documentation

Overview

Package dagu provides an experimental embedded engine API for running Dagu DAGs from Go applications.

The embedding API is experimental and may change before it is declared stable. It currently supports local file-backed execution and distributed execution against existing Dagu coordinators.

Example

package main

import (
	"context"
	"log"

	"github.com/dagucloud/dagu"
)

func main() {
	ctx := context.Background()

	engine, err := dagu.New(ctx, dagu.Options{
		HomeDir: "/var/lib/myapp/dagu",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := engine.Close(context.Background()); err != nil {
			log.Fatal(err)
		}
	}()

	run, err := engine.RunYAML(ctx, []byte(`
name: embedded
steps:
  - name: hello
    command: echo "$MESSAGE"
`), dagu.WithParams(map[string]string{"MESSAGE": "hello"}))
	if err != nil {
		log.Fatal(err)
	}

	status, err := run.Wait(ctx)
	if err != nil {
		log.Fatal(err)
	}
	_ = status
}

Example (Distributed)

package main

import (
	"context"
	"log"

	"github.com/dagucloud/dagu"
)

func main() {
	ctx := context.Background()

	engine, err := dagu.New(ctx, dagu.Options{
		HomeDir:     "/var/lib/myapp/dagu-worker",
		DefaultMode: dagu.ExecutionModeDistributed,
		Distributed: &dagu.DistributedOptions{
			Coordinators: []string{"127.0.0.1:50055"},
			TLS:          dagu.TLSOptions{Insecure: true},
			WorkerSelector: map[string]string{
				"pool": "default",
			},
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := engine.Close(context.Background()); err != nil {
			log.Fatal(err)
		}
	}()

	worker, err := engine.NewWorker(dagu.WorkerOptions{
		Labels: map[string]string{"pool": "default"},
	})
	if err != nil {
		log.Fatal(err)
	}

	workerCtx, stopWorker := context.WithCancel(ctx)
	defer stopWorker()
	go func() {
		if err := worker.Start(workerCtx); err != nil {
			log.Print(err)
		}
	}()
	if err := worker.WaitReady(ctx); err != nil {
		log.Fatal(err)
	}

	run, err := engine.RunFile(ctx, "daily-report.yaml")
	if err != nil {
		log.Fatal(err)
	}
	status, err := run.Wait(ctx)
	if err != nil {
		log.Fatal(err)
	}
	_ = status
}

Index

• func RegisterExecutor(name string, factory ExecutorFactory, opts ...ExecutorOption)
• func UnregisterExecutor(name string)
• type DistributedOptions
• type Engine
  • func New(ctx context.Context, opts Options) (*Engine, error)
  • func (e *Engine) Close(ctx context.Context) error
  • func (e *Engine) NewWorker(opts WorkerOptions) (*Worker, error)
  • func (e *Engine) Outputs(ctx context.Context, ref RunRef) (map[string]string, error)
  • func (e *Engine) RunFile(ctx context.Context, path string, opts ...RunOption) (*Run, error)
  • func (e *Engine) RunYAML(ctx context.Context, yaml []byte, opts ...RunOption) (*Run, error)
  • func (e *Engine) Status(ctx context.Context, ref RunRef) (*Status, error)
  • func (e *Engine) Stop(ctx context.Context, ref RunRef) error
• type ExecutionMode
• type Executor
• type ExecutorCapabilities
• type ExecutorFactory
• type ExecutorOption
  • func WithExecutorCapabilities(caps ExecutorCapabilities) ExecutorOption
  • func WithStepValidator(validator StepValidator) ExecutorOption
• type Options
• type Run
  • func (r *Run) ID() string
  • func (r *Run) Name() string
  • func (r *Run) Outputs(ctx context.Context) (map[string]string, error)
  • func (r *Run) Ref() RunRef
  • func (r *Run) Status(ctx context.Context) (*Status, error)
  • func (r *Run) Stop(ctx context.Context) error
  • func (r *Run) Wait(ctx context.Context) (*Status, error)
• type RunOption
  • func WithDefaultWorkingDir(dir string) RunOption
  • func WithDryRun(enabled bool) RunOption
  • func WithLabels(labels ...string) RunOption
  • func WithMode(mode ExecutionMode) RunOption
  • func WithName(name string) RunOption
  • func WithParams(params map[string]string) RunOption
  • func WithParamsList(params []string) RunOption
  • func WithRunID(id string) RunOption
  • func WithTags(tags ...string) RunOption
  • func WithWorkerSelector(selector map[string]string) RunOption
• type RunRef
• type Status
• type Step
• type StepValidator
• type TLSOptions
• type Worker
  • func (w *Worker) Start(ctx context.Context) error
  • func (w *Worker) Stop(ctx context.Context) error
  • func (w *Worker) WaitReady(ctx context.Context) error
• type WorkerOptions

Examples

• Package
• Package (Distributed)

Functions

func RegisterExecutor

func RegisterExecutor(name string, factory ExecutorFactory, opts ...ExecutorOption)

RegisterExecutor registers a custom executor type before engine or runtime use. It panics when name is empty, invalid, or factory is nil. Registration mutates global process state and must be completed before concurrent DAG execution.

func UnregisterExecutor

func UnregisterExecutor(name string)

UnregisterExecutor removes a custom executor type registered by RegisterExecutor. It is intended for tests and should not run concurrently with engine use.

Types

type DistributedOptions

type DistributedOptions struct {
	// Coordinators are coordinator gRPC addresses.
	Coordinators []string
	// TLS configures coordinator client TLS.
	TLS TLSOptions
	// WorkerSelector constrains distributed runs to matching workers.
	WorkerSelector map[string]string
	// PollInterval controls distributed run status polling.
	PollInterval time.Duration
	// MaxStatusErrors is the number of consecutive status failures before Wait fails.
	MaxStatusErrors int
}

DistributedOptions configures distributed execution.

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

Engine is an embedded Dagu engine backed by the configured file stores.

func New

func New(ctx context.Context, opts Options) (*Engine, error)

New creates an embedded Dagu engine.

func (*Engine) Close

func (e *Engine) Close(ctx context.Context) error

Close releases engine resources.

func (*Engine) NewWorker

func (e *Engine) NewWorker(opts WorkerOptions) (*Worker, error)

NewWorker creates an embedded distributed worker.

func (*Engine) Outputs

func (e *Engine) Outputs(ctx context.Context, ref RunRef) (map[string]string, error)

Outputs reads the collected step outputs for a local DAG run.

func (*Engine) RunFile

func (e *Engine) RunFile(ctx context.Context, path string, opts ...RunOption) (*Run, error)

RunFile loads a DAG definition from a file and starts it asynchronously.

func (*Engine) RunYAML

func (e *Engine) RunYAML(ctx context.Context, yaml []byte, opts ...RunOption) (*Run, error)

RunYAML loads a DAG definition from YAML bytes and starts it asynchronously.

func (*Engine) Status

func (e *Engine) Status(ctx context.Context, ref RunRef) (*Status, error)

Status reads the latest status for a local DAG run.

func (*Engine) Stop

func (e *Engine) Stop(ctx context.Context, ref RunRef) error

Stop requests cancellation for a local DAG run.

type ExecutionMode

type ExecutionMode string

ExecutionMode controls how a DAG run is dispatched.

const (
	// ExecutionModeLocal runs the DAG in the current process.
	ExecutionModeLocal ExecutionMode = "local"
	// ExecutionModeDistributed dispatches the DAG to configured coordinators.
	ExecutionModeDistributed ExecutionMode = "distributed"
)

type Executor

type Executor = runtimeexec.Executor

Executor is implemented by custom step executors.

type ExecutorCapabilities

type ExecutorCapabilities = core.ExecutorCapabilities

ExecutorCapabilities declares which step fields a custom executor supports.

type ExecutorFactory

type ExecutorFactory func(context.Context, Step) (Executor, error)

ExecutorFactory creates an Executor for a loaded step.

type ExecutorOption

type ExecutorOption func(*executorRegistration)

ExecutorOption customizes custom executor registration.

func WithExecutorCapabilities

func WithExecutorCapabilities(caps ExecutorCapabilities) ExecutorOption

WithExecutorCapabilities registers supported step fields for the custom executor.

func WithStepValidator

func WithStepValidator(validator StepValidator) ExecutorOption

WithStepValidator registers a validation function for the custom executor.

type Options

type Options struct {
	// HomeDir is the Dagu application home used for default config and data paths.
	HomeDir string
	// ConfigFile loads Dagu configuration from an explicit config file.
	ConfigFile string
	// DAGsDir overrides the directory used to resolve named DAGs and sub-DAGs.
	DAGsDir string
	// DataDir overrides the file-backed state directory.
	DataDir string
	// LogDir overrides the run log directory.
	LogDir string
	// ArtifactDir overrides the artifact directory.
	ArtifactDir string
	// BaseConfig points at a base configuration file applied during DAG loading.
	BaseConfig string
	// Logger receives embedded engine logs. A quiet logger is used when nil.
	Logger *slog.Logger

	// DefaultMode is used when a run does not set WithMode.
	DefaultMode ExecutionMode
	// Distributed configures dispatch and worker clients.
	Distributed *DistributedOptions
}

Options configures an embedded Dagu engine.

type Run

type Run struct {
	// contains filtered or unexported fields
}

Run is a handle for an asynchronous DAG run.

func (*Run) ID

func (r *Run) ID() string

ID returns the DAG run ID.

func (*Run) Name

func (r *Run) Name() string

Name returns the DAG name.

func (*Run) Outputs

func (r *Run) Outputs(ctx context.Context) (map[string]string, error)

Outputs reads the collected step outputs for this run.

func (*Run) Ref

func (r *Run) Ref() RunRef

Ref returns the run reference.

func (*Run) Status

func (r *Run) Status(ctx context.Context) (*Status, error)

Status returns the current run status.

func (*Run) Stop

func (r *Run) Stop(ctx context.Context) error

Stop requests cancellation for this run.

func (*Run) Wait

func (r *Run) Wait(ctx context.Context) (*Status, error)

Wait blocks until the DAG run reaches a terminal state or ctx is canceled.

type RunOption

type RunOption func(*runOptions)

RunOption customizes a single DAG run.

func WithDefaultWorkingDir

func WithDefaultWorkingDir(dir string) RunOption

WithDefaultWorkingDir sets the default working directory while loading a DAG.

func WithDryRun

func WithDryRun(enabled bool) RunOption

WithDryRun enables or disables dry-run mode.

func WithLabels

func WithLabels(labels ...string) RunOption

WithLabels adds labels to one run.

func WithMode

func WithMode(mode ExecutionMode) RunOption

WithMode overrides the engine default execution mode.

func WithName

func WithName(name string) RunOption

WithName overrides the loaded DAG name.

func WithParams

func WithParams(params map[string]string) RunOption

WithParams sets DAG parameters from a key-value map.

func WithParamsList

func WithParamsList(params []string) RunOption

WithParamsList sets DAG parameters from Dagu-style KEY=VALUE entries.

func WithRunID

func WithRunID(id string) RunOption

WithRunID sets an explicit DAG run ID.

func WithTags

func WithTags(tags ...string) RunOption

WithTags adds labels to one run.

Deprecated: use WithLabels.

func WithWorkerSelector

func WithWorkerSelector(selector map[string]string) RunOption

WithWorkerSelector sets the distributed worker selector for one run.

type RunRef

type RunRef struct {
	Name string
	ID   string
}

RunRef identifies a DAG run.

type Status

type Status struct {
	Name        string
	RunID       string
	AttemptID   string
	Status      string
	StartedAt   time.Time
	FinishedAt  time.Time
	Error       string
	LogFile     string
	ArchiveDir  string
	WorkerID    string
	TriggerType string
}

Status is a stable snapshot of a DAG run.

type Step

type Step = core.Step

Step is the public alias for a Dagu step passed to custom executors.

type StepValidator

type StepValidator = core.StepValidator

StepValidator validates custom executor step configuration during DAG loading.

type TLSOptions

type TLSOptions struct {
	// Insecure explicitly allows plaintext coordinator connections.
	Insecure bool
	// CertFile is the client certificate file for TLS connections.
	CertFile string
	// KeyFile is the client private key file for TLS connections.
	KeyFile string
	// ClientCAFile is the CA file used to verify coordinator certificates.
	ClientCAFile string
	// SkipTLSVerify skips coordinator certificate verification.
	SkipTLSVerify bool
}

TLSOptions configures TLS for coordinator and worker peer clients.

type Worker

type Worker struct {
	// contains filtered or unexported fields
}

Worker is a distributed worker connected to configured coordinators.

func (*Worker) Start

func (w *Worker) Start(ctx context.Context) error

Start registers and starts the worker. It blocks until ctx is canceled or the worker exits with an error.

func (*Worker) Stop

func (w *Worker) Stop(ctx context.Context) error

Stop stops the worker.

func (*Worker) WaitReady

func (w *Worker) WaitReady(ctx context.Context) error

WaitReady blocks until the worker has registered with a coordinator.

type WorkerOptions

type WorkerOptions struct {
	// ID is the worker identifier. A host and process based ID is generated when empty.
	ID string
	// MaxActiveRuns limits concurrent DAG runs. A default is used when zero or negative.
	MaxActiveRuns int
	// Labels are advertised to coordinators and matched by worker selectors.
	Labels map[string]string
	// Coordinators overrides DistributedOptions.Coordinators when non-empty.
	// If empty, the worker falls back to the engine-level DistributedOptions.
	// The resolved coordinator list must contain at least one non-empty address.
	Coordinators []string
	// TLS overrides DistributedOptions.TLS when non-zero. If zero, the worker
	// falls back to the engine-level DistributedOptions TLS settings.
	TLS TLSOptions
	// HealthPort starts the worker health endpoint on the given port. Zero disables it.
	HealthPort int
}

WorkerOptions configures an embedded distributed worker.