This is the multi-page printable view of this section. .

Return to the regular view of this page.

Reference

CLI commands, configuration sections, plugin catalog, and extension ladder.

1: CLI Reference
2: Configuration

1 - CLI Reference

All 11 Eventboat CLI commands with flags and examples.

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.

Flag	Default	Description
`--config`	(required)	Pipeline YAML file
`--strict`	false	Upgrade warnings to errors

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.

Flag	Default	Description
`--config`	—	Single pipeline file
`--config-dir`	—	Directory of pipeline files (daemon mode)
`--runtime`	`./eventboat.yaml`	Runtime config (telemetry endpoints)
`--data-dir`	`data`	SQLite storage directory
`--ephemeral`	false	In-memory store (nothing persists)

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.

Flag	Description
`--stdio`	Speak MCP over stdin/stdout (for agent hosts)
`--http`	Serve MCP over HTTP with Admin REST + SSE + UI
`--config-dir`	Deploy pipelines at startup

2 - Configuration

Pipeline YAML sections, edge attributes, variable substitution, and the Runtime config.

Top-level sections

Section	Required	Purpose
`apiVersion` / `kind` / `metadata`	✅	Resource identity (K8s convention)
`sources`	✅ (≥1)	Topology: where events come from
`transforms`	optional	Topology: what happens to events
`sinks`	✅ (≥1)	Topology: where events go
`run`	job pipelines only	Job scheduling (mode/schedule/overlap/catchup_window)
`parameters`	job optional	Typed job parameters with defaults
`constants`	optional	Read-only values visible to scripts and predicates
`hooks`	optional	Lifecycle hooks (failure/success → inline sink)
`limits`	optional	Per-pipeline resource limits
`edge_defaults`	optional	Default edge attributes
`codecs`	optional	Named codec declarations
`dlq`	optional	Dead-letter policy

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:

Attribute	Type	Description
`when`	string or object	CEL predicate (or `{lang: cesql, expr: ...}`)
`delivery`	object	`{retries, backoff, timeout_ms}`
`required`	bool	`false` = best-effort (failure doesn’t block siblings)
`buffer`	object	`{max_events, strategy}`

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

Name	Key config
`kafka`	brokers, topics, group_id
`http_server`	listen, max_body_bytes
`cron`	expression (5-field)
`file`	path (tail)
`sql`	driver (mysql/postgres/sqlite), query, cursor, pagination

Sinks

Name	Key config
`kafka`	brokers, topic
`http`	url, timeout_ms
`file`	path (JSON lines)
`drop`	(none — discards)

Codecs

Name	Key config
`json`	(none)
`raw`	(none)
`csv`	columns or header
`avro`	schema (inline or file)
`protobuf`	descriptor_set (file path)

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 }