AgentMail Webhook 教程：验签、事件去重与回复审批工作流

python3 -m venv .venv
source .venv/bin/activate
python -m pip install Flask==3.1.3 svix==2.8.0
export WORKFLOW_DB=workflow.sqlite3
export INBOX_OWNERS_JSON='{"[email protected]":"tenant-a"}'

"""Single-host tutorial core. No network calls or public approval endpoint."""
import hashlib
import json
import sqlite3
import time
import uuid


def connect(path):
    db = sqlite3.connect(path, timeout=10)
    db.row_factory = sqlite3.Row
    return db


def init(path):
    with connect(path) as db:
        db.executescript('''
        CREATE TABLE IF NOT EXISTS events(
          delivery_id TEXT PRIMARY KEY, event_id TEXT UNIQUE NOT NULL);
        CREATE TABLE IF NOT EXISTS jobs(
          id TEXT PRIMARY KEY, tenant TEXT NOT NULL, inbox TEXT NOT NULL,
          message TEXT NOT NULL, thread TEXT NOT NULL, state TEXT NOT NULL,
          snapshot TEXT, digest TEXT, approved_by TEXT, expires REAL,
          send_key TEXT NOT NULL, receipt TEXT,
          UNIQUE(tenant,inbox,message));
        ''')


def ingest(path, delivery_id, event, inbox_owners):
    if event.get('event_type') != 'message.received':
        return 'ignored'
    msg = event['message']
    values = [delivery_id, event['event_id'], msg['inbox_id'],
              msg['message_id'], msg['thread_id']]
    if not all(isinstance(x, str) and x for x in values):
        raise ValueError('invalid identifiers')
    tenant = inbox_owners[msg['inbox_id']]
    with connect(path) as db:
        inserted = db.execute(
            'INSERT OR IGNORE INTO events VALUES (?,?)', values[:2]).rowcount
        if not inserted:
            return 'duplicate'
        job_id = uuid.uuid4().hex
        db.execute('''INSERT OR IGNORE INTO jobs
          (id,tenant,inbox,message,thread,state,send_key)
          VALUES (?,?,?,?,?,'queued',?)''',
          (job_id, tenant, *values[2:], 'reply-' + job_id))
    return 'stored'


def stage(path, job_id, tenant, recipients, text, allowed):
    # Addresses here are explicit addr-spec strings selected by trusted app code.
    if (not recipients or not all(isinstance(a, str) and a in allowed
                                  for a in recipients)
            or not isinstance(text, str) or not text.strip()):
        raise ValueError('recipient or body rejected')
    with connect(path) as db:
        db.execute('BEGIN IMMEDIATE')
        row = db.execute('SELECT * FROM jobs WHERE id=? AND tenant=?',
                         (job_id, tenant)).fetchone()
        if not row or row['state'] not in ('queued', 'pending'):
            raise ValueError('job not editable')
        snapshot = json.dumps({
            'tenant': tenant, 'inbox': row['inbox'], 'message': row['message'],
            'thread': row['thread'], 'to': recipients, 'cc': [], 'bcc': [],
            'reply_all': False, 'text': text,
        }, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
        digest = hashlib.sha256(snapshot.encode()).hexdigest()
        db.execute('''UPDATE jobs SET state='pending',snapshot=?,digest=?,
          approved_by=NULL,expires=NULL WHERE id=?''',
          (snapshot, digest, job_id))
    return digest


def approve(path, job_id, tenant, displayed_digest, actor):
    # actor must come from your authenticated internal operator session.
    if not actor:
        raise ValueError('operator required')
    with connect(path) as db:
        n = db.execute('''UPDATE jobs SET state='approved',approved_by=?,expires=?
          WHERE id=? AND tenant=? AND state='pending' AND digest=?''',
          (actor, time.time() + 3600, job_id, tenant, displayed_digest)).rowcount
        if n != 1:
            raise ValueError('stale approval')


def dispatch(path, job_id, tenant, allowed, send_request):
    with connect(path) as db:
        db.execute('BEGIN IMMEDIATE')
        row = db.execute('SELECT * FROM jobs WHERE id=? AND tenant=?',
                         (job_id, tenant)).fetchone()
        if not row or row['state'] != 'approved' or row['expires'] < time.time():
            raise ValueError('not approved or expired')
        snapshot = row['snapshot']
        if hashlib.sha256(snapshot.encode()).hexdigest() != row['digest']:
            raise ValueError('snapshot changed')
        data = json.loads(snapshot)
        if not all(a in allowed for a in data['to']):
            raise ValueError('allowlist changed')
        db.execute("UPDATE jobs SET state='sending' WHERE id=?", (job_id,))
    # Claim commits before the external side effect. Other workers cannot claim it.
    try:
        result = send_request(data, row['send_key'])
        if not result.get('message_id') or not result.get('thread_id'):
            raise ValueError('ambiguous response')
        with connect(path) as db:
            db.execute("UPDATE jobs SET state='sent',receipt=? WHERE id=?",
                       (json.dumps(result), job_id))
    except Exception:
        with connect(path) as db:
            db.execute("UPDATE jobs SET state='unknown' WHERE id=?", (job_id,))
        raise
    return result

import json
import os
import sqlite3
from flask import Flask, request
from svix.webhooks import Webhook, WebhookVerificationError
from workflow import ingest, init

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 2 * 1024 * 1024
DB = os.environ.get('WORKFLOW_DB', 'workflow.sqlite3')
OWNERS = json.loads(os.environ['INBOX_OWNERS_JSON'])
verifier = Webhook(os.environ['AGENTMAIL_WEBHOOK_SECRET'])
init(DB)


@app.post('/webhooks/agentmail')
def receive():
    try:
        raw = request.get_data()
        verifier.verify(raw, dict(request.headers))
    except WebhookVerificationError:
        return '', 400
    try:
        event = json.loads(raw)
        if not isinstance(event, dict):
            return '', 400
        ingest(DB, request.headers['svix-id'], event, OWNERS)
    except (KeyError, ValueError, TypeError):
        return '', 400
    except sqlite3.Error:
        return '', 503
    return '', 204


if __name__ == '__main__':
    app.run(host='127.0.0.1', port=3000)

python server.py

from workflow import stage, approve

allowed = {"[email protected]"}
displayed_digest = stage(
    "workflow.sqlite3", job_id, "tenant-a",
    ["[email protected]"], "已收到资料，我们会核实后继续回复。", allowed,
)
#在内部审核界面展示完整快照，操作者确认后才执行下一行。
approve("workflow.sqlite3", job_id, "tenant-a", displayed_digest, operator_id)

import json
import os
from urllib.parse import quote
from urllib.request import Request, urlopen


def agentmail_reply(snapshot, send_key):
    # Call only from dispatch after a trusted operator has approved this snapshot.
    inbox = quote(snapshot['inbox'], safe='')
    message = quote(snapshot['message'], safe='')
    url = f'https://api.agentmail.to/v0/inboxes/{inbox}/messages/{message}/reply'
    payload = {k: snapshot[k] for k in ['to', 'cc', 'bcc', 'reply_all', 'text']}
    req = Request(url, data=json.dumps(payload).encode(), method='POST', headers={
        'Authorization': 'Bearer ' + os.environ['AGENTMAIL_SEND_KEY'],
        'Content-Type': 'application/json', 'Idempotency-Key': send_key,
    })
    with urlopen(req, timeout=20) as response:
        return json.load(response)