Kaptanto

Every insert, update, and delete from your Postgres or MongoDB database, streamed the moment it happens, as one static Go binary.

"Kaptanto" means "who captures" in Esperanto.

./kaptanto \
  --source "postgres://localhost:5432/mydb" \
  --tables public.orders,public.payments \
  --output stdout

{"op":"insert","table":"orders","after":{"id":1234,"status":"pending","total":99.99}}
{"op":"update","table":"orders","before":{"status":"pending"},"after":{"status":"shipped"}}

Kaptanto tails your database's transaction log — Postgres WAL (logical replication) or MongoDB Change Streams — instead of polling tables. Each change is parsed into a unified ChangeEvent, durably appended to an embedded event log, and only then is the source checkpoint advanced, so a crash can never lose an event. An initial snapshot (backfill) runs concurrently with live streaming; a watermark check discards snapshot rows already superseded by a WAL event. Events are then fanned out, in per-key order, to any number of outputs.

• Zero runtime dependencies — static binary
• Durable event log — every event is written before the source checkpoint advances; a crash never loses an event
• Ten outputs — stdout (NDJSON), SSE, gRPC, Kafka, NATS, SQS, Google Pub/Sub, RabbitMQ, webhook, and vector (embeddings → pgvector / Pinecone / Qdrant)
• AI-native — MCP server for agents, fail-open ai_context enrichment, vector sink for RAG, plus TypeScript / Python / Mastra SDKs
• Consistent backfills — snapshot and live stream run concurrently; watermark coordination prevents duplicate or stale rows
• Per-key ordering — events for the same primary key always arrive in commit order, even across broker sinks
• Per-consumer cursors — each consumer tracks its own position; reconnect at any time and resume exactly where you left off
• Filtering — table, column, operation, and SQL WHERE condition filters
• High availability — leader election via Postgres advisory lock (--ha), ~5s failover; optional cluster mode (--cluster) shares consumer cursor state across nodes
• Security — TLS/mTLS and bearer-token auth on the SSE/gRPC data plane (--insecure to explicitly opt out)
• Observability — Prometheus metrics and health check on --port for SSE and queue sinks, --port + 1 for gRPC (stdout serves no HTTP endpoints)

make build   # produces ./kaptanto

Enable logical replication in postgresql.conf:

wal_level = logical

Grant replication access:

CREATE ROLE kaptanto WITH REPLICATION LOGIN PASSWORD 'secret';
GRANT SELECT ON TABLE public.orders, public.payments TO kaptanto;
-- for full before/after values on updates and deletes:
ALTER TABLE public.orders REPLICA IDENTITY FULL;

Run:

./kaptanto \
  --source "postgres://kaptanto:secret@localhost:5432/mydb" \
  --tables public.orders,public.payments \
  --output stdout

No server-side setup is required beyond a replica set (Change Streams need one, even a single-node rs0). Point --source at a mongodb:// or mongodb+srv:// URI — Kaptanto detects the source type from the DSN scheme:

./kaptanto \
  --source "mongodb://localhost:27017/mydb" \
  --tables mydb.orders,mydb.payments \
  --output stdout

If the resume token is invalidated (e.g. oplog rollover while disconnected), Kaptanto automatically falls back to a fresh snapshot.

./kaptanto --source "..." --tables public.orders --output sse --port 7654
curl -N http://localhost:7654/events?consumer=worker-1

Every broker sink routes by CDC key (partition, subject, queue, or routing key), so per-key ordering is preserved end to end; none retries internally — retry is the router's job — and every message carries an idempotency key/header for downstream dedup.

Seven connection blocks live under sinks: (kafka, nats, sqs, pubsub, rabbitmq, webhook, vector) — only the active output's block needs to be populated.

All flags are available via CLI or YAML; CLI flags always win.

source: "postgres://kaptanto:secret@localhost:5432/mydb"
output: sse
port: 7654
cors-origin: ""          # SSE Access-Control-Allow-Origin; empty = no cross-origin browser access
data-dir: /var/lib/kaptanto
retention: 24h
ha: false                # leader election (Postgres advisory lock)
node-id: ""              # unique node identity, required when ha is enabled
source-id: default       # slot name kaptanto_<id> / publication kaptanto_pub_<id>
all-tables: false        # explicit opt-in to replicate every table when 'tables:' is empty
cluster: false           # shared cursor state across nodes
cluster-dsn: ""          # Postgres DSN for the shared cursor store, required when cluster is true
cluster-peers: []        # NATS JetStream cluster peer addresses, e.g. ["node2:6222", "node3:6222"]
nats-cluster-port: 6222

tables:
  public.orders:
    columns: [id, status, total]
    where: "status != 'archived'"
  public.payments: {}     # empty = replicate all columns, no row filter

sinks:                    # only the active sink's block needs to be populated
  kafka:
    bootstrap-servers: ["localhost:9092"]
    topic-template: "cdc.{{.Schema}}.{{.Table}}"
    sasl-mechanism: ""     # "", PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512
    tls: { ca-file: "", cert-file: "", key-file: "" }

server-tls:               # inbound TLS for the SSE/gRPC server, distinct from sink TLS above
  cert-file: ""
  key-file: ""
  client-ca-file: ""       # set to require + verify client certs (mTLS)

auth-token: ""             # bearer token for SSE/gRPC (prefer KAPTANTO_AUTH_TOKEN env var)
insecure: false            # explicit opt-out of TLS/auth — not for production

Actions turn CDC events into side effects — Slack messages, HTTP requests, email alerts, cache purges, vector upserts, or workflow triggers — without writing consumer code. Set output: none to run kaptanto as a pure action processor.

source: "postgres://localhost:5432/mydb"
output: none

actions:
  - name: order-alerts
    type: slack
    params:
      webhook-url: ${SLACK_WEBHOOK_URL}
    match:
      tables: ["public.orders"]
      operations: ["insert"]

Secret params must use ${VAR} env-var references — they are never logged or included in the OpenAPI spec.

Each action has an optional match: block that controls which events it receives. All three conditions are AND-ed; omitting any means "match all".

When a network output (SSE, gRPC, queue sinks, or output: webhook) or output: none is used, kaptanto serves a machine-readable spec at /openapi.json describing the configured actions, their parameter names, and routing rules.

Self-contained examples for workflow platforms live in examples/:

• n8n — examples/n8n-trigger/ (SSE trigger node, see also n8n-nodes-kaptanto/)
• Inngest — examples/inngest/ (Docker Compose + function handler)
• Trigger.dev — examples/trigger-dev/ (task definition + config)

Kaptanto speaks agent and RAG workflows without a sidecar: an optional MCP server, a vector sink, fail-open ai_context enrichment, serverless action types, and typed SDKs.

Enable the Model Context Protocol server (own listener; disabled by default — MCP-04):

output: none
mcp:
  enabled: true
  port: 7655
  api-keys:
    - name: claude-desktop
      key: ${MCP_API_KEY}
      tables: ["public.orders"]
      redact:
        - tables: ["public.orders"]
          columns: ["customer"]

Point Claude Desktop (or any streamable-HTTP MCP client) at it. Replace <MCP_API_KEY> with the same bearer token you export as MCP_API_KEY:

{
  "mcpServers": {
    "kaptanto": {
      "type": "http",
      "url": "http://localhost:7655",
      "headers": {
        "Authorization": "Bearer <MCP_API_KEY>"
      }
    }
  }
}

Tools: list_tables, get_table_schema, subscribe_to_changes, get_recent_events, unsubscribe, get_event_by_id. Every result is ACL-filtered and redacted (MCP-01); subscriptions are session-scoped (MCP-02); every call is audited (MCP-03). Full demo: examples/mcp-agent/.

Embed row text with a local Ollama model and upsert into pgvector — no cloud API keys:

output: vector
tables:
  public.documents: {}
sinks:
  vector:
    source:
      columns: [title, body]
    embedder:
      provider: openai
      base-url: http://localhost:11434/v1
      model: nomic-embed-text
      dimensions: 768
      # api-key intentionally empty — local Ollama needs no auth
    store:
      provider: pgvector
      dsn: ${VECTOR_DSN}
      table: kaptanto_vectors
    metadata: [category]
    batch:
      max-events: 16

Unchanged text is skipped via a SHA-256 hash cache (VEC-01); embedders must return one vector per input in order (VEC-02); vector IDs are schema.table:<canonical-key-JSON> (VEC-03). Full stack: examples/rag-pgvector/.

Optional HTTP enricher runs before EventLog.Append. Failures fail open — the unenriched event is still durable (AIC-01). Response bodies must be JSON objects ≤ 16 KiB (AIC-02):

enrichment:
  url: http://localhost:8080/enrich
  tables: ["public.orders"]       # explicit opt-in; empty = disabled
  operations: [insert, update]    # default when omitted
  timeout: 150ms
  auth-token: ${ENRICH_TOKEN}     # optional Bearer

Documented ai_context shape (opaque to Kaptanto):

{
  "intent": "refund_request",
  "entities": [{"type": "order_id", "value": "1234", "field": "id"}],
  "suggested_actions": ["notify_support"],
  "embedding": {"model": "nomic-embed-text", "vector": [0.1, 0.2]},
  "custom": {}
}

ai_context is omitted from JSON when empty and never participates in idempotency_key.

Invoke AWS Lambda Function URLs, Cloudflare Workers, and Vercel endpoints from CDC events via the lambda, cloudflare-worker, and vercel action types. See docs/serverless.md and the landing Serverless Actions page. Runnable demo: examples/lambda/.

Demos: examples/langchain/, examples/mastra/.

By default, the SSE/gRPC data plane requires both a bearer token (--auth-token or KAPTANTO_AUTH_TOKEN) and, if --tls-cert/--tls-key are set, TLS — with optional mutual TLS via --tls-client-ca. --insecure disables all of this for local development and logs a loud warning on startup; it is not meant for production. Outbound sink connections (Kafka, NATS, SQS, Pub/Sub, RabbitMQ, webhook, vector) have their own independent TLS/mTLS or secret settings under each sink's block. MCP API keys and enrichment auth tokens must use ${VAR} references.

{
  "id": "<ulid>",
  "idempotency_key": "<source>:<schema>.<table>:<pk>:<op>:<lsn>",
  "operation": "insert | update | delete | read | control",
  "table": "orders",
  "key": { "id": 1234 },
  "before": { "status": "pending" },
  "after":  { "status": "shipped" },
  "metadata": { "lsn": "0/1A2B3C4", "checkpoint": "...", "snapshot": false },
  "ai_context": { "intent": "refund_request" }
}

read events are emitted during the initial snapshot; control events mark lifecycle transitions (slot created, backfill complete). idempotency_key is deterministic across restarts — use it for exactly-once processing downstream. Optional ai_context carries opaque AI enrichment metadata when present (omitted when empty; never part of the idempotency key).

./data/
├── events/        # Badger event log
├── checkpoint.db  # Source position checkpoint (SQLite, or PostgreSQL in HA mode)
├── cursors.db     # Per-consumer cursor positions
└── backfill.db    # Snapshot progress and watermark state

Kaptanto is safe to restart — it resumes from the last checkpoint, and each consumer resumes from its last cursor.

Metrics and health are served on --port + 1 (default :7655):

curl http://localhost:7655/healthz   # 200 OK when healthy
curl http://localhost:7655/metrics   # Prometheus text format

make build            # CGO_ENABLED=0 static binary (default, cross-platform)
make lint             # golangci-lint over ./internal/... ./cmd/...
make test             # all tests, CGO_ENABLED=0
make test-race        # race detector (requires CGO)
make cover            # coverage run; fails below the configured threshold
make verify-no-cgo    # cross-compile linux/amd64 + darwin/arm64 to confirm no CGO leakage
make build-rust       # optional Rust-accelerated binary (requires Rust 1.77+, cargo, cbindgen)
make clean            # remove binary

Run a single test:

go test ./internal/router -run TestPerKeyOrdering -v

Slower, env-gated suites (skipped by default):

POSTGRES_TEST_DSN=... MONGO_TEST_URI=... make test-integration   # live Postgres + MongoDB
POSTGRES_TEST_DSN=... make test-e2e                              # black-box binary tests, -tags e2e
make mutation                                                    # gremlins over router/eventlog/parser/backfill

See CLAUDE.md for the full architecture/package reference and the critical invariants (durability, per-key ordering, watermark consistency, etc.) that these tests enforce.

Apache 2.0