Code Examples

import os
import httpx

API = "https://api.actionlayer.dev/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['ACTIONLAYER_API_KEY']}"}


def needs_review_thread() -> dict | None:
    r = httpx.get(f"{API}/threads", params={"needs_review": True, "limit": 1}, headers=HEADERS)
    r.raise_for_status()
    threads = r.json()["data"]
    return threads[0] if threads else None


def read_thread(thread_id: str) -> dict:
    r = httpx.get(f"{API}/threads/{thread_id}", headers=HEADERS)
    r.raise_for_status()
    return r.json()


def submit_draft(thread_id: str, identity_id: str, based_on_message_id: str, body: str, rationale: str) -> dict:
    r = httpx.post(
        f"{API}/drafts",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={
            "thread_id": thread_id,
            "identity_id": identity_id,
            "based_on_message_id": based_on_message_id,
            "body_text": body,
            "rationale": rationale,
        },
    )
    r.raise_for_status()
    return r.json()


def send_draft(draft_id: str) -> None:
    r = httpx.post(f"{API}/drafts/{draft_id}/send", headers=HEADERS)
    if r.status_code == 409:
        # Newer inbound arrived — reject and regenerate
        httpx.post(f"{API}/drafts/{draft_id}/reject", headers=HEADERS, json={"reason": "stale"})
        raise RuntimeError("draft is stale")
    r.raise_for_status()


# Loop
thread = needs_review_thread()
if thread:
    detail = read_thread(thread["id"])
    last_inbound = next(m for m in reversed(detail["messages"]) if m["direction"] == "inbound")

    body = my_llm_call(detail)  # your agent does the writing
    draft = submit_draft(
        thread_id=thread["id"],
        identity_id="IDENTITY_ID",
        based_on_message_id=last_inbound["id"],
        body=body,
        rationale="Auto-reply from agent",
    )

    if draft.get("stale_warning"):
        print("warn: thread already has newer inbound — re-read and regenerate")
    # ... wait for human approval, then:
    # send_draft(draft["id"])

import "dotenv/config";

const API = "https://api.actionlayer.dev/v1";
const headers = {
  Authorization: `Bearer ${process.env.ACTIONLAYER_API_KEY}`,
  "Content-Type": "application/json",
};

async function listNeedsReview() {
  const r = await fetch(`${API}/threads?needs_review=true&limit=10`, { headers });
  if (!r.ok) throw new Error(await r.text());
  return (await r.json()).data as Array<{ id: string; subject: string }>;
}

async function readThread(threadId: string) {
  const r = await fetch(`${API}/threads/${threadId}`, { headers });
  if (!r.ok) throw new Error(await r.text());
  return r.json();
}

async function submitDraft(input: {
  threadId: string;
  identityId: string;
  basedOnMessageId: string;
  bodyText: string;
  rationale?: string;
}) {
  const r = await fetch(`${API}/drafts`, {
    method: "POST",
    headers,
    body: JSON.stringify({
      thread_id: input.threadId,
      identity_id: input.identityId,
      based_on_message_id: input.basedOnMessageId,
      body_text: input.bodyText,
      rationale: input.rationale,
    }),
  });
  if (!r.ok) throw new Error(await r.text());
  return r.json();
}

async function sendDraft(draftId: string) {
  const r = await fetch(`${API}/drafts/${draftId}/send`, { method: "POST", headers });
  if (r.status === 409) {
    await fetch(`${API}/drafts/${draftId}/reject`, {
      method: "POST",
      headers,
      body: JSON.stringify({ reason: "stale" }),
    });
    throw new Error("stale_draft");
  }
  if (!r.ok) throw new Error(await r.text());
}

import hmac
import hashlib
import os
from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()
SECRET = os.environ["ACTIONLAYER_WEBHOOK_SECRET"].encode()


@app.post("/actionlayer-webhook")
async def receive(request: Request, x_actionlayer_signature: str = Header(...)):
    body = await request.body()
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, x_actionlayer_signature):
        raise HTTPException(401, "bad signature")

    payload = await request.json()
    if payload["event"] == "email.received":
        thread_id = payload["thread_id"]
        message_id = payload["message_id"]
        # Hand off to your agent
        await my_agent.handle_inbound(thread_id, message_id)
    return {"ok": True}

{
  "event": "email.received",
  "thread_id": "…",
  "message_id": "…",
  "identity_id": "…",
  "from_email": "lead@example.com",
  "subject": "Pricing question",
  "received_at": "2026-04-30T14:22:11Z"
}

curl -X POST "https://api.actionlayer.dev/v1/emails/send" \
  -H "Authorization: Bearer $ACTIONLAYER_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "identity_id": "IDENTITY_ID",
    "to": ["recipient@example.com"],
    "subject": "Quick intro",
    "body_text": "Hi — wanted to introduce myself."
  }'

curl -X POST "https://api.actionlayer.dev/v1/emails/reply" \
  -H "Authorization: Bearer $ACTIONLAYER_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "thread_id": "THREAD_ID",
    "identity_id": "IDENTITY_ID",
    "body_text": "Confirmed — see you Tuesday."
  }'

# List pending
curl "https://api.actionlayer.dev/v1/inbound/pending" \
  -H "Authorization: Bearer $ACTIONLAYER_API_KEY"

# Approve sender (re-processes the held message into a real thread)
curl -X POST "https://api.actionlayer.dev/v1/inbound/pending/PENDING_ID/approve" \
  -H "Authorization: Bearer $ACTIONLAYER_API_KEY"

# Block sender (drops the held message silently and blocklists future mail)
curl -X POST "https://api.actionlayer.dev/v1/inbound/pending/PENDING_ID/block" \
  -H "Authorization: Bearer $ACTIONLAYER_API_KEY"