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"}'

 hashlib
 json
 sqlite3
 time
 uuid


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


 ():
     connect(path)  db:
        db.executescript()


 ():
     event.get() != :
         
    msg = event[]
    values = [delivery_id, event[], msg[],
              msg[], msg[]]
      ((x, )  x  x  values):
         ValueError()
    tenant = inbox_owners[msg[]]
     connect(path)  db:
        inserted = db.execute(
            , values[:]).rowcount
          inserted:
             
        job_id = uuid.uuid4().
        db.execute(,
          (job_id, tenant, *values[:],  + job_id))
     


 ():
    
     ( recipients   ((a, )  a  allowed
                                   a  recipients)
              (text, )   text.strip()):
         ValueError()
     connect(path)  db:
        db.execute()
        row = db.execute(,
                         (job_id, tenant)).fetchone()
          row  row[]   (, ):
             ValueError()
        snapshot = json.dumps({
            : tenant, : row[], : row[],
            : row[], : recipients, : [], : [],
            : , : text,
        }, ensure_ascii=, sort_keys=, separators=(, ))
        digest = hashlib.sha256(snapshot.encode()).hexdigest()
        db.execute(,
          (snapshot, digest, job_id))
     digest


 ():
    
      actor:
         ValueError()
     connect(path)  db:
        n = db.execute(,
          (actor, time.time() + , job_id, tenant, displayed_digest)).rowcount
         n != :
             ValueError()


 ():
     connect(path)  db:
        db.execute()
        row = db.execute(,
                         (job_id, tenant)).fetchone()
          row  row[] !=   row[] < time.time():
             ValueError()
        snapshot = row[]
         hashlib.sha256(snapshot.encode()).hexdigest() != row[]:
             ValueError()
        data = json.loads(snapshot)
          (a  allowed  a  data[]):
             ValueError()
        db.execute(, (job_id,))
    
    :
        result = send_request(data, row[])
          result.get()   result.get():
             ValueError()
         connect(path)  db:
            db.execute(,
                       (json.dumps(result), job_id))
     Exception:
         connect(path)  db:
            db.execute(, (job_id,))
        
     result