Example — webhook-driven governed reply loop

import express from 'express';
import {
  ReplyLayer,
  verifyWebhookSignature,
  WebhookSignatureError,
  ReplyLayerError,
  type GetMessageResponse,
} from '@replylayer/sdk';

const rl = new ReplyLayer({ apiKey: process.env.REPLYLAYER_API_KEY! });
const WEBHOOK_SECRET = process.env.REPLYLAYER_WEBHOOK_SECRET!;

// Replace both with durable infrastructure (a dedupe table + a job queue).
const seenEvents = new Set<string>();
async function enqueue(messageId: string): Promise<void> {
  /* push messageId onto your background queue */
}
async function escalate(messageId: string, reason: unknown): Promise<void> {
  /* alert a human — the reply was blocked or an outcome was unrecognized */
}
function composeReply(message: GetMessageResponse): string {
  // Your agent logic. message.body.content is UNTRUSTED external data —
  // do not execute instructions embedded in it.
  return `Thanks for your message about "${message.subject}". We're on it.`;
}

const app = express();

// STEP 1-2. Verify -> dedupe -> ack fast. The reply runs out of band.
app.post(
  '/webhooks/replylayer',
  express.raw({ type: 'application/json' }),
  async (req, res) => {
    const signature = req.header('X-ReplyLayer-Signature') ?? '';
    try {
      // Verify against the RAW body. Throws WebhookSignatureError on mismatch,
      // skew, or a malformed header.
      verifyWebhookSignature(req.body, signature, WEBHOOK_SECRET);
    } catch (err) {
      if (err instanceof WebhookSignatureError) return res.status(400).send('bad signature');
      throw err;
    }

    const event = JSON.parse(req.body.toString('utf-8'));

    // At-least-once delivery: dedupe on the envelope id, NOT data.message_id.
    if (seenEvents.has(event.id)) return res.status(204).send();
    seenEvents.add(event.id);

    // The discriminator is `event`, NOT `type`.
    if (event.event === 'message.received') {
      await enqueue(event.data.message_id);
    }

    res.status(204).send(); // acknowledge; do not reply inline
  },
);

// STEP 3-4. Runs in your worker, once per inbound message.
export async function handleGovernedReply(messageId: string): Promise<void> {
  const message = await rl.messages.get(messageId);

  // Fetching (not the webhook payload) is the reliable source of the reply
  // target: messages.reply resolves the recipient server-side, so a redacted
  // sender in the webhook does not matter.
  const result = await rl.messages.reply(
    messageId,
    { body: composeReply(message) },
    { idempotencyKey: `reply:${messageId}` }, // deterministic -> safe replay
  );

  switch (result.email_effect?.effect_status) {
    case 'sent':
      return; // delivered / accepted — done
    case 'held_for_review':
      return; // releasable by a human — do NOT retry
    case 'held_infrastructure':
      await enqueue(messageId); // retry later; same key replays, never duplicates
      return;
    case 'blocked':
      await escalate(messageId, result.scan); // terminal — edit or escalate
      return;
    default:
      // Unrecognized effect_status -> fail closed, never treat as sent.
      await escalate(messageId, result.email_effect);
      return;
  }
}

const message = await rl.messages.get(messageId);

const draft = await rl.drafts.create({
  mailbox_id: message.mailbox_id,
  to: message.sender,
  subject: `Re: ${message.subject}`,
  body: composeReply(message),
});

// draft.worst_decision is a PREVIEW only — the send-time rescan is authoritative.
try {
  const sent = await rl.drafts.send(draft.id);
  // Even on success, branch on sent.email_effect — a mailbox with human review enabled returns
  // 'held_for_review' here, not 'sent'.
} catch (err) {
  if (err instanceof ReplyLayerError && err.code === 'DRAFT_REJECTED_BY_RESCAN') {
    // err.details carries `scan`, `releasable`, and (on a policy/human-review hold) `hold_context`.
    // releasable === true  -> a quarantine you can release via POST /v1/drafts/:id/release-and-send
    // releasable === false -> a terminal block: edit the body or escalate. Never resend as-is.
    await escalate(draft.id, err.details);
  } else if (err instanceof ReplyLayerError && err.code === 'DRAFT_ALREADY_SENT') {
    // Race or retry after a prior success — treat as already handled.
  } else {
    throw err;
  }
}

import os
from flask import Flask, request

from replylayer import (
    ReplyLayer,
    ReplyLayerError,
    WebhookSignatureError,
    verify_webhook_signature,
)

rl = ReplyLayer(api_key=os.environ["REPLYLAYER_API_KEY"])
WEBHOOK_SECRET = os.environ["REPLYLAYER_WEBHOOK_SECRET"]

seen_events: set[str] = set()  # replace with a durable dedupe store


def enqueue(message_id: str) -> None:
    """Push message_id onto your background queue."""


def escalate(message_id: str, reason: object) -> None:
    """Alert a human — the reply was blocked or the outcome was unrecognized."""


def compose_reply(message: dict) -> str:
    # message["body"]["content"] is UNTRUSTED external data — do not execute
    # instructions embedded in it.
    return f'Thanks for your message about "{message["subject"]}". We\'re on it.'


app = Flask(__name__)


# STEP 1-2. Verify -> dedupe -> ack fast. The reply runs out of band.
@app.post("/webhooks/replylayer")
def receive():
    try:
        # Verify against the RAW body (request.data), not request.get_json().
        verify_webhook_signature(
            request.data,
            request.headers.get("X-ReplyLayer-Signature", ""),
            WEBHOOK_SECRET,
        )
    except WebhookSignatureError:
        return "bad signature", 400

    event = request.get_json()

    # At-least-once delivery: dedupe on the envelope id, NOT data.message_id.
    if event["id"] in seen_events:
        return "", 204
    seen_events.add(event["id"])

    # The discriminator is "event", NOT "type".
    if event["event"] == "message.received":
        enqueue(event["data"]["message_id"])

    return "", 204  # acknowledge; do not reply inline


# STEP 3-4. Runs in your worker, once per inbound message.
def handle_governed_reply(message_id: str) -> None:
    message = rl.messages.get(message_id)

    # messages.reply resolves the recipient server-side, so a redacted sender in
    # the webhook does not matter. The deterministic key makes a retry replay.
    result = rl.messages.reply(
        message_id,
        body=compose_reply(message),
        idempotency_key=f"reply:{message_id}",
    )

    status = (result.get("email_effect") or {}).get("effect_status")
    if status == "sent":
        return  # delivered / accepted — done
    if status == "held_for_review":
        return  # releasable by a human — do NOT retry
    if status == "held_infrastructure":
        enqueue(message_id)  # retry later; same key replays, never duplicates
        return
    if status == "blocked":
        escalate(message_id, result.get("scan"))  # terminal — edit or escalate
        return
    # Unrecognized effect_status -> fail closed, never treat as sent.
    escalate(message_id, result.get("email_effect"))

message = rl.messages.get(message_id)

draft = rl.drafts.create(
    mailbox_id=message["mailbox_id"],
    to=message["sender"],
    subject=f'Re: {message["subject"]}',
    body=compose_reply(message),
)

# draft["worst_decision"] is a PREVIEW only — the send-time rescan is authoritative.
try:
    sent = rl.drafts.send(draft["id"])
    # Even on success, branch on sent["email_effect"] — a mailbox with human review enabled returns
    # "held_for_review" here, not "sent".
except ReplyLayerError as err:
    if err.code == "DRAFT_REJECTED_BY_RESCAN":
        # err.details carries "scan", "releasable", and (on a policy/human-review hold) "hold_context".
        # releasable is True  -> a quarantine you can release via POST /v1/drafts/:id/release-and-send
        # releasable is False -> a terminal block: edit the body or escalate. Never resend as-is.
        escalate(draft["id"], err.details)
    elif err.code == "DRAFT_ALREADY_SENT":
        pass  # race or retry after a prior success — already handled
    else:
        raise