Approval over Slack or email

A wire transfer needs a person's yes. Post the request to a chat channel (or send a mail). The reviewer clicks a button. The call then runs once, as approved.

sequenceDiagram
  participant App
  participant Ex as Executor
  participant S as ApprovalStore
  participant W as notify worker
  participant C as Slack or email
  participant R as reviewer
  App->>Ex: run_call(call, ctx)
  Ex->>S: create (pending)
  Ex-->>App: status=pending, tool.pending event to the outbox
  W->>S: get(approval_id)
  W->>C: "Approve wire_transfer {...}?"
  R->>C: clicks Approve
  C->>App: signed callback
  App->>S: decide(approval_id, Decision)
  App->>Ex: run_call(same call, same ctx): ok

Notify

outbox: list[str] = []  # approval ids waiting for a message; a queue in real code
sent: list[str] = []  # stands in for the Slack or email API


def on_event(event: Event) -> None:
    """Event sinks must be quick and never raise: only note the id here."""
    if event.type == "tool.pending":
        outbox.append(event.data["approval_id"])


store = MemoryApprovalStore()  # a database-backed store across processes
executor = Executor(registry, approval=DeferredApproval(store), sink=CallbackEventSink(on_event))


async def send_reviews() -> None:
    """A worker: one message per pending approval, with what the reviewer must see."""
    while outbox:
        record = await store.get(outbox.pop(0))
        text = f"Approve {record.request.tool} {json.dumps(record.request.arguments)}?"
        sent.append(
            f"#approvals {record.approval_id}: {text}"
        )  # post_message(...) / send_mail(...)

An event sink runs inside the call: keep it quick and never raise. Do the slow sending in a worker. The tool.pending event carries the approval id, never the arguments; the worker reads them from the store.

The button callback

def sign(body: bytes) -> str:
    return hmac.new(SIGNING_SECRET, body, hashlib.sha256).hexdigest()


async def on_button(body: bytes, signature: str) -> str:
    """The chat app's callback: verify it, check the reviewer, record the decision."""
    if not hmac.compare_digest(sign(body), signature):
        return "403 bad signature"
    action = json.loads(body)
    reviewer = REVIEWERS.get(action["user"])
    if reviewer is None:
        return "403 not a reviewer"
    approved = action["value"] == "approve"
    decision = Decision(approved=approved, decided_by=reviewer, reason=action.get("reason", ""))
    await store.decide(action["approval_id"], decision)  # the first decision wins
    return "200 recorded"

Check	Why
Verify the chat app's signature (Slack: the signing secret; mail: a signed one-time link)	anyone can POST to your URL
Map the chat user to a known reviewer	a channel member is not a reviewer
Four-eyes: refuse when the reviewer is the invoker	keep the invoker with the approval id in your own table, and compare
Let the store pick the winner	the first decide wins; a second raises ApprovalAlreadyDecided

Pause, decide, resume

ctx = executor.context(run_id="run_payout_9")  # keep run id and call id to resume
call = ToolCall(name="wire_transfer", arguments={"account": "AC-1", "amount": 900}, id="c1")
first = await executor.run_call(call, ctx)
await send_reviews()
print(first.status, sent[0].split(": ", 1)[1])

approval_id = first.error.details["approval_id"] if first.error else ""
forged = json.dumps({"approval_id": approval_id, "user": "U_ANA", "value": "approve"})
print(await on_button(forged.encode(), "0" * 64))  # unsigned: refused
print(await on_button(forged.encode(), sign(forged.encode())))  # signed by the chat app

resumed = await executor.run_call(call, ctx)  # same run, same call id
print(resumed.status, resumed.value)

pending Approve wire_transfer {"account": "AC-1", "amount": 900.0}?
403 bad signature
200 recorded
ok sent 900.0 to AC-1

Pitfalls

Pitfall	Do instead
Resuming with a new run id or call id	it opens a new review; store both with the approval id
MemoryApprovalStore with several workers	use SQLiteApprovalStore, RedisApprovalStore, SQLAlchemyApprovalStore or DynamoDBApprovalStore (Approval stores, Database stores)
Running the tool from the callback	only record the decision; the resume runs the call through the executor
You already run the HTTP server	it has GET /approvals and POST /approvals/{id}/decide with owner checks (HTTP server)

Files

File	Does
examples/cookbook_approval_notify.py	the outbox sink, the notify worker, a signed callback, the resume
tests/docs/test_cookbook.py	runs it offline and checks the output