Documentation

• 1: Introduction
  • 1.1: What is Eventboat?
  • 1.2: Architecture
• 2: Get Started
  • 2.1: Installation
  • 2.2: Your First Pipeline
• 3: Reference
  • 3.1: CLI Reference
  • 3.2: Configuration

1 - Introduction

1.1 - What is Eventboat?

Eventboat routes events through directed acyclic graphs (DAGs) of transforms — from sources (Kafka, HTTP, cron, SQL, files) to sinks (Kafka, HTTP, files), with at-least-once delivery, dead-lettering, and replay.

It is agent-native: every capability is accessible through MCP (Model Context Protocol), CLI, and a language server — so AI agents can write, verify, deploy, and operate pipelines autonomously.

What makes it different

The one-line pitch

Eventboat lets AI agents build and run event pipelines that don’t lose messages — because machines verify every step before it goes live.

1.2 - Architecture

Three-layer model

YAML (+overlay) → Config (typed) → Static IR → Runtime Engine
                                          ↓
                              Source → Transform → Sink plugins

• Config layer: YAML parsing, strict schema validation, variable substitution
• Static IR: validated DAG + precompiled CEL programs + Starlark programs + schema
• Runtime: spool + settle + checkpoint engine, consuming only the IR

Reliability model

source → [spool: append-only durable queue] → in-memory DAG → sinks
                │                                  │
                └── checkpoint ←── settle tracker ←── terminal states

• Spool: every message hits SQLite before the DAG sees it (invariant 1)
• Settle: each message settles when all branches reach terminal state
• Checkpoint: advances only over settled prefix (invariant 2)
• Crash recovery: kill -9 → restart → replay from checkpoint, never lose (invariant 3)
• Dead letters: exhausted retries → DLQ store with query + replay CLI

Seven invariant tests

Each has a dedicated test that must pass in CI:

1. Spool before visible
2. Checkpoint advances only after settle
3. Kill -9 replay covers all unsettled
4. Dead-letter write failure blocks settle
5. required: false edges don’t block siblings
6. Redelivery keeps message ID stable
7. Cursor watermark never exceeds settled

Four machine gates

2 - Get Started

2.1 - Installation

Prerequisites

• Go 1.25+ (for building from source)
• No runtime dependencies — everything is compiled into one binary

Install

go install github.com/eventboat/eventboat/cmd/eventboat@latest

Verify:

eventboat help

You should see the help screen with 11 commands listed.

What’s in the binary

Next steps

• Write your first pipeline
• CLI reference

2.2 - Your First Pipeline

The three-section format

Every Eventboat pipeline is a single YAML file with three top-level sections: sources, transforms, and sinks — connected by from edges.

apiVersion: eventboat/v3
kind: Pipeline
metadata: { name: my-first-pipeline }

sources:
  ingest:
    cron: { expression: "*/5 * * * *" }   # every 5 minutes

transforms:
  hello:
    from: [ingest]
    script: |
      payload.greeting = "hello from eventboat"
      payload.timestamp = meta.ingest_time

sinks:
  console:
    from: [hello]
    drop: {}                              # discard (demo)

Verify (gate 1)

eventboat verify --config pipeline.yaml

Output: pipeline.yaml: 0 error(s), 0 warning(s)

Run

eventboat run --config pipeline.yaml

The pipeline starts; every 5 minutes the cron source fires, the Starlark script enriches the message, and the sink receives it.

Add branching (CEL predicates)

sinks:
  important:
    from: { hello: { when: 'payload.score > 100' } }
    http: { url: "https://api.example.com/alerts" }

  archive:
    from: [hello]                          # unconditional edge
    drop: {}

Agent mode

# Start the MCP server — AI agents connect via stdio
eventboat mcp --stdio

# Or with the admin UI
eventboat mcp --http

Next steps

• Configuration reference
• CLI reference
• Examples

3 - Reference

3.1 - CLI Reference

Overview

eventboat [--json] verify --config <pipeline.yaml> [--strict]
eventboat [--json] test <testfile-or-dir> [...]
eventboat run --config <pipeline.yaml> [--data-dir DIR] [--ephemeral]
eventboat run --config-dir <dir> [--runtime runtime.yaml]
eventboat [--json] trigger --config <job.yaml> [--parameters '{"from":"..."}']
eventboat [--json] jobs list --config <job.yaml> [--limit N]
eventboat [--json] jobs show <run-id> --config <job.yaml>
eventboat [--json] explain --config <pipeline.yaml> [--message f.json] [--topology]
eventboat [--json] replay --config <pipeline.yaml> (--dlq | --spool --from N | --job <run-id>) [--dry-run]
eventboat repl [--message sample.json] [--cel 'expr' | --script f.star]
eventboat lsp
eventboat [--json] plugin catalog
eventboat [--json] plugin schema <name>
eventboat mcp (--stdio | --http) [--config-dir <dir>] [--data-dir DIR]

verify

Statically validate a pipeline. Checks schema, topology invariants (no cycles, no orphans, source has no in-edges, sink has no out-edges, at least one source→sink path), CEL predicate compilation, Starlark script compilation, job configuration, and semantic lint.

test

Run contract test suites against the real in-process engine. Test files declare injection points, expected captures, and DLQ assertions.

suite: my-test
pipeline: ../pipeline.yaml
cases:
  - name: vip-order-routed-correctly
    inject: { at: ingest, messages: [fixtures/order.json] }
    expect:
      capture: { at: out }
      messages:
        - payload.total: 12000    # subset match

run

Execute a pipeline. Job pipelines (with run.mode: job) run under the jobs manager with scheduling and catchup. --config-dir starts a multi-pipeline daemon with the admin surface.

trigger

Manually fire a job pipeline once, optionally with parameters (backfill).

eventboat trigger --config sync.yaml --parameters '{"from":"2026-08-01","to":"2026-09-01"}'

explain

Deterministic walkthrough of a pipeline. With --message, performs real CEL evaluation and Starlark dry-run on the sample. With --topology, renders the DAG (mermaid + ASCII).

replay

Re-inject dead letters (--dlq), a spool window (--spool --from N), or one job run’s dead letters (--job <run-id>) into a live pipeline.

repl

Evaluate CEL predicates and Starlark scripts against one sample message without running a pipeline.

eventboat repl --message sample.json --cel 'payload.score > 100'
eventboat repl --message sample.json --script transform.star

mcp

Start the MCP server for AI agents.

3.2 - Configuration

Top-level sections

Three-section topology

Nodes are organized by section; edges declared on the downstream side via from:

sources:
  ingest:
    decoder: json
    kafka: { brokers: ["${KAFKA_BROKERS}"], topics: [orders] }

transforms:
  enrich:
    from: [ingest]                         # unconditional edge
    script: |
      payload.total = payload.price * payload.qty

sinks:
  eu-out:
    from: { enrich: { when: 'meta.region == "eu"' } }  # conditional edge
    kafka: { topic: orders-eu }

Edge attributes

Attributes on from elements:

Variable substitution

• ${VAR} — environment variable (unset = error)
• ${?VAR} — optional (unset = omit key)
• ${constants.name} — pipeline constant
• Applies to all string values

Built-in plugins

Sources

Sinks

Codecs

Job pipeline example

apiVersion: eventboat/v3
kind: Pipeline
metadata: { name: nightly-sync }

run:
  mode: job
  schedule: "0 1 * * *"
  overlap: skip
  catchup_window: 2h
  skip_if_successful: true
  retention: { history: 90d }

parameters:
  from: { type: string, default: cursor }
  to:   { type: string, default: now }

sources:
  pull:
    sql:
      driver: mysql
      query: |
        SELECT * FROM orders
        WHERE updated_at >= :from AND updated_at < :to
      args: { from: "${parameters.from}", to: "${parameters.to}" }
      cursor: { column: updated_at }
      pagination: { key: [updated_at, id], page_size: 5000 }

transforms:
  enrich:
    from: [pull]
    script: |
      payload.source_system = constants.source_system

sinks:
  out:
    from: [enrich]
    kafka: { brokers: ["${KAFKA_BROKERS}"], topic: orders-sync }