Any CLI Agent

A zero-dependency Node script for agents with a plain command line — claude -p, codex exec, aider, or your own script. It polls EventPort, pipes each event as JSON to your command's stdin, and dead-letters failures locally. This is the fallback when no first-class adapter exists.

Prerequisites

• Node >= 18
• An EventPort subscription — you need its gateway URL and consumer token (starts with egc_, from the subscription's Agent tab)

Installation

1. Create the destination directory

mkdir -p ~/.local/bin

2. Save the file below

Save to ~/.local/bin/event-agent.mjs:

#!/usr/bin/env node
/**
 * event-agent — generic eventport consumer for any agent CLI.
 *
 * The gateway DELETES events as they are read (consume-on-read is the
 * default subscription mode), so delivery is at-most-once and the server
 * never redelivers. This wrapper compensates with a local dead-letter file:
 *
 *   GET /events  → rows deleted on read → pipe event JSON to COMMAND stdin
 *   exit 0    → done
 *   exit ≠ 0  → event appended to the dead-letter file (EG_DLQ), so nothing
 *               is silently lost; replay later with --replay
 *
 * Usage (daemon mode, poll every 60s):
 *   EG_URL=https://gw.eventport.dev EG_TOKEN=egc_xxx \
 *     node event-agent.mjs -- claude -p --output-format text
 *
 * One-shot mode (drive it from cron / launchd / Task Scheduler):
 *   node event-agent.mjs --once -- claude -p
 *
 * Replay failed events through the same command:
 *   node event-agent.mjs --once --replay -- claude -p
 *
 * Environment:
 *   EG_URL       gateway base URL (required unless --replay)
 *   EG_TOKEN     consumer token, egc_* (required unless --replay)
 *   EG_INTERVAL  poll interval seconds, default 60 (consumer limit: 60 req/min)
 *   EG_DLQ       dead-letter file, default ./eventport.dead.ndjson
 *
 * The child command receives on stdin:
 *   { "messageId": "...", "payload": { ...raw upstream webhook body... },
 *     "timestamp": 1721203200000 }
 *
 * Node >= 18, zero dependencies.
 */
import { spawn } from 'node:child_process';
import { appendFileSync, readFileSync, writeFileSync, existsSync } from 'node:fs';

// ------------------------------------------------------------------ args

const argv = process.argv.slice(2);
const onceIdx = argv.indexOf('--once');
const ONCE = onceIdx !== -1;
if (ONCE) argv.splice(onceIdx, 1);
const replayIdx = argv.indexOf('--replay');
const REPLAY = replayIdx !== -1;
if (REPLAY) argv.splice(replayIdx, 1);
const sep = argv.indexOf('--');
const COMMAND = sep === -1 ? [] : argv.slice(sep + 1);

const URL = process.env.EG_URL?.replace(/\/$/, '');
const TOKEN = process.env.EG_TOKEN;
const INTERVAL = (Number(process.env.EG_INTERVAL ?? 60) || 60) * 1000;
const DLQ = process.env.EG_DLQ ?? 'eventport.dead.ndjson';

if (((!URL || !TOKEN) && !REPLAY) || COMMAND.length === 0) {
  console.error('usage: EG_URL=... EG_TOKEN=egc_... node event-agent.mjs [--once] [--replay] -- <command> [args...]');
  process.exit(2);
}

const log = (...a) => console.log(new Date().toISOString(), ...a);

// ---------------------------------------------------------------- gateway

async function pollEvents() {
  const res = await fetch(`${URL}/events`, {
    headers: { Authorization: `Bearer ${TOKEN}` },
  });
  if (!res.ok) throw new Error(`GET /events -> ${res.status}`);
  const body = await res.json();
  // consume-on-read: the rows below are already deleted server-side
  return body.events ?? [];
}

// ------------------------------------------------------------ dead-letter

function deadLetter(event, reason) {
  appendFileSync(DLQ, `${JSON.stringify({ ...event, reason, failedAt: Date.now() })}\n`);
  log(`event ${event.messageId} failed (${reason}) — appended to ${DLQ}`);
}

function readDeadLetters() {
  if (!existsSync(DLQ)) return [];
  return readFileSync(DLQ, 'utf8')
    .split('\n')
    .filter(Boolean)
    .map((line) => {
      try {
        return JSON.parse(line);
      } catch {
        return null;
      }
    })
    .filter(Boolean);
}

// ---------------------------------------------------------------- command

/** Run COMMAND with the event JSON on stdin. Resolves true on exit 0. */
function runCommand(event) {
  return new Promise((resolve) => {
    const child = spawn(COMMAND[0], COMMAND.slice(1), { stdio: ['pipe', 'inherit', 'inherit'] });
    child.on('error', (err) => {
      console.error(`spawn ${COMMAND[0]} failed:`, err.message);
      resolve(false);
    });
    child.on('close', (code) => resolve(code === 0));
    child.stdin.end(JSON.stringify(event));
  });
}

// ------------------------------------------------------------------- loop

/** Poll cycle. Events are consumed on read; failures go to the DLQ file. */
async function pollCycle() {
  const events = await pollEvents();
  let ok = 0;
  for (const event of events) {
    log(`event ${event.messageId} -> ${COMMAND.join(' ')}`);
    if (await runCommand(event)) ok += 1;
    else deadLetter(event, 'command-exit-nonzero');
  }
  log(`cycle done: ${events.length} event(s), ${ok} ok, ${events.length - ok} dead-lettered`);
}

/** Replay dead-lettered events through the same command; successes are removed. */
async function replayCycle() {
  const entries = readDeadLetters();
  if (entries.length === 0) {
    log('replay: dead-letter file is empty');
    return;
  }
  const failed = [];
  for (const entry of entries) {
    const event = { messageId: entry.messageId, payload: entry.payload, timestamp: entry.timestamp };
    log(`replay ${event.messageId} -> ${COMMAND.join(' ')}`);
    if (await runCommand(event)) log(`replay ${event.messageId}: ok`);
    else failed.push(entry);
  }
  writeFileSync(DLQ, failed.map((e) => JSON.stringify(e)).join('\n') + (failed.length ? '\n' : ''));
  log(`replay done: ${entries.length} total, ${entries.length - failed.length} recovered, ${failed.length} still failing`);
}

async function main() {
  for (;;) {
    try {
      if (REPLAY) await replayCycle();
      else await pollCycle();
    } catch (err) {
      console.error('poll error:', err.message ?? err);
    }
    if (ONCE) process.exit(0);
    await new Promise((r) => setTimeout(r, INTERVAL));
  }
}

main();

3. Make it executable (optional)

chmod +x ~/.local/bin/event-agent.mjs

Run

# daemon mode, polls every 60s
EG_URL=https://gw.eventport.dev \
EG_TOKEN=egc_xxx \
node ~/.local/bin/event-agent.mjs -- claude -p --output-format text

# one-shot mode — let cron / launchd / Task Scheduler drive it instead
node ~/.local/bin/event-agent.mjs --once -- claude -p

Your command receives one event per invocation, as JSON on stdin:

{ "messageId": "…", "payload": { …upstream webhook body… }, "timestamp": 1721203200000 }

When the queue is empty the script just logs cycle done: 0 event(s) and waits for the next interval.

Configuration

| Env | Meaning | Default | |-----|---------|---------| | EG_URL | gateway base URL | required | | EG_TOKEN | consumer token (egc_*) | required | | EG_INTERVAL | poll interval in seconds | 60 | | EG_DLQ | dead-letter file for failed runs | ./eventport.dead.ndjson |

Delivery semantics

The gateway deletes events on read (consume-on-read): once GET /events returns an event it is removed server-side, and delivery is at-most-once with no redelivery.

• Exit 0 → done, the event was already consumed by the poll.
• Non-zero exit → the event is appended to the local dead-letter file (EG_DLQ, default ./eventport.dead.ndjson) — nothing is silently lost. Replay failed events with:

node ~/.local/bin/event-agent.mjs --once --replay -- claude -p

• Make handlers idempotent by messageId — dead-letter replays can present the same event twice.
• One instance tracks one subscription (the one that issued EG_TOKEN). To consume another subscription, run another instance with its token.