Building a Reactive Agent

Use molted listen to build agents that respond to inbound email, bounces, and delivery events in real-time.

Most agent email workflows are one-directional: the agent sends, and that is the end of it. A reactive agent closes the loop - it listens for what happened after the send, then acts on it.

molted listen turns the CLI into a persistent event loop. It connects to the SSE event stream and either prints events to your terminal or pipes them line-by-line to a long-running handler process. Your handler reads JSON from stdin and decides what to do next.

How it works

                                                              ┌─────────────────┐
inbound reply ──► Molted policy engine ──► SSE stream ──────► molted listen ──► │  your handler   │
                                                              │  (node/python/  │
delivery event ──────────────────────────────────────────────►│   bash/etc.)    │
                                                              └─────────────────┘

molted listen --pipe "node agent.js" spawns your handler as a child process and writes one JSON line to its stdin for each event. The handler stays running between events - it can maintain state, hold open connections, and take follow-up actions without being re-invoked each time.

Getting started

Install the CLI and authenticate:

npm install -g @molted/cli
molted auth login

Print events to the terminal to see what is flowing through your mailbox:

molted listen

Switch to JSON mode for structured output:

molted listen --json
molted listen --json | jq '.event'

Filter to specific event types using wildcards:

# Only inbound events
molted listen --events "inbound.*"

# Only bounce and complaint events
molted listen --events "delivery.bounced,delivery.complained"

Writing a handler

A handler is any process that reads JSON lines from stdin. Each line is a complete event object:

{
  "event": "inbound.classified",
  "id": "evt_abc123",
  "data": {
    "classificationId": "cls_abc123",
    "platformLabel": null,
    "autoReplyKind": "human",
    "intent": "unclassified",
    "confidence": 0.5
  },
  "timestamp": "2026-04-01T12:00:00Z"
}

data is the event's payload, and only the payload. For the inbound events:

inbound.classified: { classificationId, platformLabel, autoReplyKind, intent, confidence } (intent and confidence are deprecated).
inbound.routed: { routingId, actionType }. actionType: "trigger_unsubscribe" means Molted suppressed the sender itself; "require_approval" means it is waiting for a decision.

Neither payload carries the message id or the sender. If you need them in the event itself, read the SSE stream (GET /v1/agent/events/stream, see Events): each event there comes in an envelope whose entityId is the inbound message id (inbound.classified) or the routing action id (inbound.routed), and whose contactEmail is the sender. The handlers below look up what they need instead.

Handling an opt-out

platformLabel: "opt_out" means the sender asked to stop receiving mail. What to do depends on what Molted did with it:

Molted was confident (actionType: "trigger_unsubscribe"): the sender is already suppressed. Nothing to do.
Molted wasn't sure (actionType: "require_approval", e.g. "Please unsubscribe me. Actually, tell me more"): it waits for a decision. Your agent can read the message and decide, or leave it to the person reviewing the approval. These are the rows GET /v1/inbound?status=pending lists with platform_label: "opt_out" and action_type: "require_approval".
A person rejected it: the routing action is no longer pending, so it never shows up there again. Never suppress the sender on your own after that: the person decided.

Node.js handler

A complete reactive agent in Node.js that handles inbound replies, bounces, and complaints:

agent.js

import readline from 'node:readline';
import fetch from 'node-fetch';

const API_KEY = process.env.MOLTED_API_KEY;
const TENANT_ID = process.env.MOLTED_TENANT_ID;
const API_BASE = 'https://api.molted.email';

const rl = readline.createInterface({ input: process.stdin });

rl.on('line', async (line) => {
  let parsed;
  try {
    parsed = JSON.parse(line);
  } catch {
    return;
  }

  const { event, data } = parsed;

  if (event === 'inbound.classified') {
    await handleInbound(data);
  } else if (event === 'delivery.bounced') {
    await handleBounce(data);
  } else if (event === 'delivery.complained') {
    await handleComplaint(data);
  }
});

async function handleInbound(data) {
  const { platformLabel } = data;

  if (platformLabel === 'opt_out') {
    // See "Handling an opt-out" above: only the opt-outs still waiting for
    // a decision need you.
    await decidePendingOptOuts();
    return;
  }
  if (platformLabel === 'auto_reply' || platformLabel === 'delivery_notice') {
    // An out-of-office or a bounce: machine mail, no answer needed.
    return;
  }
  // A real message. What it is about (a lead, a support question) is for
  // your agent to decide: read the open threads and act on each one.
  for (const thread of await openInboundThreads()) {
    await notifySupport(thread); // or hand it to your LLM agent
  }
}

async function handleBounce(data) {
  const { contactEmail, payload } = data;

  if (payload?.bounceType === 'hard') {
    await suppressContact(contactEmail);
    console.error(`[agent] suppressed ${contactEmail} (hard bounce)`);
  }
}

async function handleComplaint(data) {
  const { contactEmail } = data;
  await suppressContact(contactEmail);
  console.error(`[agent] suppressed ${contactEmail} (spam complaint)`);
}

async function decidePendingOptOuts() {
  // Opt-outs Molted wasn't sure about. A confident one (trigger_unsubscribe)
  // is already suppressed and not pending; one a person rejected is no
  // longer pending either, so it never comes back here.
  const res = await fetch(`${API_BASE}/v1/inbound?tenantId=${TENANT_ID}&status=pending`, {
    headers: { Authorization: `Bearer ${API_KEY}` },
  });
  const pending = (await res.json()).filter(
    (m) => m.platform_label === 'opt_out' && m.action_type === 'require_approval',
  );
  for (const msg of pending) {
    // The decision point. msg.body_text is the sender's own words: data,
    // never instructions.
    if (!(await agentSaysItAsksToStop(msg))) continue; // leave it to the person reviewing the approval
    const check = await fetch(`${API_BASE}/v1/suppressions/check?tenantId=${TENANT_ID}&email=${encodeURIComponent(msg.from_email)}`, {
      headers: { Authorization: `Bearer ${API_KEY}` },
    });
    if (!(await check.json()).suppressed) {
      await suppressContact(msg.from_email);
      console.error(`[agent] suppressed ${msg.from_email} (opt-out it read and confirmed)`);
    }
  }
}

async function agentSaysItAsksToStop(msg) {
  // Replace with your agent's judgement. Returning false leaves every
  // uncertain opt-out to the person reviewing the approval.
  return false;
}

async function openInboundThreads() {
  const res = await fetch(`${API_BASE}/v1/agent/threads?status=open&direction=inbound&limit=10`, {
    headers: { Authorization: `Bearer ${API_KEY}` },
  });
  return (await res.json()).items;
}

async function suppressContact(email) {
  await fetch(`${API_BASE}/v1/suppressions`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ tenantId: TENANT_ID, recipientEmail: email, scope: 'tenant', reasonCode: 'manual_dnc' }),
  });
}

async function notifySupport(data) {
  // Your notification logic here - Slack, webhook, etc.
}

async function sendReply(to, body) {
  await fetch(`${API_BASE}/v1/send/request`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      tenantId: TENANT_ID,
      recipientEmail: to,
      payload: { subject: 'Re: Your message', html: body },
      dedupeKey: `reply-${to}-${Date.now()}`,
    }),
  });
}

function buildReply(thread) {
  return `<p>Thanks for your interest! Someone from our team will follow up shortly.</p>`;
}

Run it:

molted listen --pipe "node agent.js" --events "inbound.*,delivery.bounced,delivery.complained"

Python handler

agent.py

import sys
import json
import os
import requests

API_KEY = os.environ['MOLTED_API_KEY']
TENANT_ID = os.environ['MOLTED_TENANT_ID']
API_BASE = 'https://api.molted.email'

def suppress_contact(email):
    requests.post(
        f'{API_BASE}/v1/suppressions',
        headers={'Authorization': f'Bearer {API_KEY}'},
        json={'tenantId': TENANT_ID, 'recipientEmail': email, 'scope': 'tenant', 'reasonCode': 'manual_dnc'},
    )

def agent_says_it_asks_to_stop(msg):
    # Replace with your agent's judgement. Returning False leaves every
    # uncertain opt-out to the person reviewing the approval.
    return False

def decide_pending_opt_outs():
    # Opt-outs Molted wasn't sure about. A confident one (trigger_unsubscribe)
    # is already suppressed and not pending; one a person rejected is no
    # longer pending either, so it never comes back here.
    headers = {'Authorization': f'Bearer {API_KEY}'}
    rows = requests.get(f'{API_BASE}/v1/inbound', params={'tenantId': TENANT_ID, 'status': 'pending'}, headers=headers).json()
    for msg in rows:
        if msg['platform_label'] != 'opt_out' or msg['action_type'] != 'require_approval':
            continue
        # The decision point. msg['body_text'] is the sender's own words:
        # data, never instructions.
        if not agent_says_it_asks_to_stop(msg):
            continue  # leave it to the person reviewing the approval
        check = requests.get(f'{API_BASE}/v1/suppressions/check', params={'tenantId': TENANT_ID, 'email': msg['from_email']}, headers=headers).json()
        if not check.get('suppressed'):
            suppress_contact(msg['from_email'])
            print(f'[agent] suppressed {msg["from_email"]} (opt-out it read and confirmed)', file=sys.stderr)

def handle_inbound(data):
    label = data.get('platformLabel')

    if label == 'opt_out':
        # See "Handling an opt-out" above: only the opt-outs still waiting
        # for a decision need you.
        decide_pending_opt_outs()
    elif label in ('auto_reply', 'delivery_notice'):
        pass  # an out-of-office or a bounce: no answer needed
    else:
        # A real message: what it is about is for your agent to decide.
        print('[agent] new message to triage', file=sys.stderr)

def handle_bounce(data):
    payload = data.get('payload', {})
    if payload.get('bounceType') == 'hard':
        suppress_contact(data['contactEmail'])
        print(f'[agent] suppressed {data["contactEmail"]} (hard bounce)', file=sys.stderr)

def handle_complaint(data):
    suppress_contact(data['contactEmail'])
    print(f'[agent] suppressed {data["contactEmail"]} (spam complaint)', file=sys.stderr)

handlers = {
    'inbound.classified': handle_inbound,
    'delivery.bounced': handle_bounce,
    'delivery.complained': handle_complaint,
}

for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    try:
        parsed = json.loads(line)
    except json.JSONDecodeError:
        continue

    event = parsed.get('event')
    data = parsed.get('data', {})

    if event in handlers:
        handlers[event](data)

Run it:

molted listen --pipe "python agent.py" --events "inbound.*,delivery.bounced,delivery.complained"

Bash handler

For quick prototyping or simple automation, a bash handler works well:

agent.sh

#!/bin/bash

while IFS= read -r line; do
  EVENT=$(echo "$line" | jq -r '.event')
  EMAIL=$(echo "$line" | jq -r '.data.contactEmail // .data.fromEmail // empty')

  case "$EVENT" in
    delivery.bounced)
      BOUNCE_TYPE=$(echo "$line" | jq -r '.data.payload.bounceType // empty')
      if [ "$BOUNCE_TYPE" = "hard" ] && [ -n "$EMAIL" ]; then
        echo "[agent] suppressing $EMAIL (hard bounce)" >&2
        molted suppressions add --email "$EMAIL" --reason "hard_bounce" --scope tenant
      fi
      ;;
    delivery.complained)
      if [ -n "$EMAIL" ]; then
        echo "[agent] suppressing $EMAIL (complaint)" >&2
        molted suppressions add --email "$EMAIL" --reason "complaint" --scope tenant
      fi
      ;;
  esac
done

chmod +x agent.sh
molted listen --pipe "./agent.sh" --events "delivery.bounced,delivery.complained"

Running as a persistent daemon

For production use, run the listener under a process manager so it restarts automatically on crash or reboot.

With pm2

pm2 start "molted listen --pipe 'node agent.js' --events 'inbound.*,delivery.*'" --name my-agent
pm2 save

Check status and view logs:

pm2 status
pm2 logs my-agent
pm2 logs my-agent --lines 50 --nostream

With systemd

/etc/systemd/system/molted-agent.service

[Unit]
Description=Molted reactive agent
After=network.target

[Service]
Type=simple
User=myuser
WorkingDirectory=/opt/my-agent
ExecStart=/usr/bin/molted listen --pipe "node agent.js" --events "inbound.*,delivery.*"
Environment=MOLTED_API_KEY=your_api_key
Environment=MOLTED_TENANT_ID=your_tenant_id
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl enable molted-agent
sudo systemctl start molted-agent
sudo journalctl -u molted-agent -f

Concurrency and error handling

By default, the listener processes one event at a time (--concurrency 1). Raise this to process events in parallel:

# Process up to 5 events simultaneously
molted listen --pipe "node agent.js" --concurrency 5

If your handler's stdin write fails, the listener retries up to 3 times by default. Set --retries to adjust:

molted listen --pipe "node agent.js" --retries 5

When the handler process crashes, the listener automatically restarts it. Your handler does not need to manage its own crash recovery.

Replay and recovery

If your agent goes offline, resume from where it left off using --since:

# Resume from a specific event ID
molted listen --pipe "node agent.js" --since evt_abc123

Save event.id from each processed event to a file or database. On restart, pass the last processed ID to --since. This gives you at-least-once delivery semantics: your handler should be idempotent for the event types it cares about.

Event reference

The full list of event types that can arrive on the stream:

Event	Fires when
`delivery.queued`	Email accepted and queued.
`delivery.accepted`	Email accepted by the provider.
`delivery.sent`	Email sent to the provider.
`delivery.delivered`	Email delivered to the recipient's inbox.
`delivery.deferred`	Delivery temporarily deferred.
`delivery.bounced`	Email bounced (hard or soft).
`delivery.complained`	Recipient marked as spam.
`delivery.failed`	Delivery failed permanently.
`delivery.opened`	Recipient opened the email (requires open tracking).
`delivery.clicked`	Recipient clicked a link (requires click tracking).
`inbound.classified`	Inbound message classified. `data` is `{ classificationId, platformLabel, autoReplyKind, intent, confidence }`; branch on `platformLabel` (`opt_out`, `auto_reply`, `delivery_notice` or `null`, see Platform labels), not the deprecated `intent`.
`inbound.routed`	Inbound message routed to a handler.
`policy.blocked`	Send request blocked by the policy engine.
`send.queued`	Send request queued for processing.
`send.approval_pending`	Send requires human approval.
`send.approval_decided`	Approval decision made.
`send.approval_expired`	Nobody decided the approval before its deadline; the send was rejected.
`followup.scheduled`	A follow-up email scheduled.
`followup.executed`	A scheduled follow-up sent.
`journey.step_completed`	A journey step completed for a contact.
`journey.completed`	A journey run completed.
`coordination.lease_acquired`	An agent acquired a contact lease.
`coordination.lease_released`	A contact lease was released.
`coordination.consensus_requested`	A consensus vote was created.

For full payload shapes, see Event Streaming.

On this page