deliverd

Add human approval to any AI agent. Ask a person, wait for the answer, act on it.

pip install deliverd

from deliverd import deliverd

decision = deliverd.approve(title="Refund £1,240 to Acme", approvers=["Finance"])
if decision.approved:
    refund()

That is the whole integration. The call blocks until somebody decides, and returns whether they said yes.

What it stands in for: an approval page, the emails and the reminders, identity and single sign-on, who is allowed to decide, a table to keep it in, an audit trail, and the dashboard somebody uses to do it. None of that is yours to build.

• No dependencies. The standard library has HTTP, HMAC and JSON. Nothing new lands in your agent's environment.
• Python 3.9 and up.
• Calls block, or await. deliverd.approve(...) is the shape an agent tool wants: the function that asks is the function that answers. Inside asyncio, AsyncDeliverd has the same names — see Awaiting instead.
• The same surface as the TypeScript SDK, held to it field for field by a guard in this repository — approvals, reviews, collections, flows, reports, comments, schedules and webhook verification.

Before you have an account

from deliverd import Deliverd

deliverd = Deliverd(mode="development")
decision = deliverd.approve(title="Deploy to production")
print(decision.approved)   # True

No key, no network, no colleague to interrupt. The imaginary approver decides instantly, and the four things that can happen to a real request are all one line away:

Deliverd(mode="development", development=DevelopmentOptions(outcome="rejected"))
Deliverd(mode="development", development=DevelopmentOptions(outcome="timeout"))
Deliverd(mode="development", development=DevelopmentOptions(outcome="question"))

or DELIVERD_DEV_OUTCOME=question in the environment, so a whole test suite flips without touching the code.

question is the one people forget to write. An approver can ask the agent something before deciding, and until it is answered nothing moves:

decision = deliverd.approve(
    title="Refund £1,240 to Acme",
    on_question=lambda q, approval: "Yes — staged on Tuesday",
)

Return None from the handler to leave it for a person.

Development mode is a stand-in transport, not a branch inside the client, so the retries, the error handling and the polling are the same code that runs in production. It answers the calls the primitives make — the four verbs and flows — and returns 501 for everything else, rather than pretending to be the whole API. Publishing is the one it will not fake: asking a person has an imaginary approver and an answer, while publishing puts bytes somewhere and there is nowhere local to put them, so a made-up "published" with a URL that resolves to nothing would be worse than the refusal.

Configuration

from deliverd import Deliverd

deliverd = Deliverd()                      # DELIVERD_API_KEY from the environment
deliverd = Deliverd(api_key="dlv_…")       # or pass it

The module-level deliverd builds itself from the environment the first time you touch it, so the shortest form has no setup line at all. Build a Deliverd yourself when you need two keys, a different base URL, or development mode.

The four verbs

Approve

decision = deliverd.approve(
    title="Refund £1,240 to Acme",
    description="Duplicate charge on invoice 4821.",
    risk="high",
    factors=[{"label": "Customer charged twice", "status": "warning"}],
    links=[{"label": "Invoice 4821", "url": "https://…"}],
    approvers=["Finance"],          # a person, a team, a group — or omit it
    expires_in="4h",
)

decision.approved        # True only for an approval
decision.status          # approved | rejected | expired | cancelled
decision.note            # their reason, when they gave one
decision.decided_by      # "Sarah Chen" — a name, not an id
decision.decided_by_id    # their user id, when the key is what you need
decision.url             # the page they decided on

A refusal is an answer, not an error. approve() returns on rejected, expired and cancelled as well as approved; it raises only when the wait failed — a timeout, a cancellation, or an API it could not reach.

approvers takes a person (email or user id), the name of a team or directory group, or a word meaning the whole organisation. Naming a team is usually what you want: a request addressed to one person waits for that person to come back from holiday.

Review

outcome = deliverd.review(
    title="Q3 board pack",
    report_id=report.id,
    instructions="Check the figures against the ledger.",
    reviewers=["partner@firm.example"],
)
if not outcome.approved:
    revise(outcome.verdicts, outcome.thread_count)

Collect

answers = deliverd.collect(
    title="Before I file the Q3 return",
    respondents=["finance@firm.example"],
    questions=[
        {"prompt": "Headcount at 30 September", "kind": "number"},
        {"prompt": "Any disposals in the quarter?", "kind": "boolean"},
    ],
)
if answers.complete:
    file_return(answers["Headcount at 30 September"])

answers is keyed by the question as you asked it. Check complete before you use it: a request that expired has whatever arrived and no more.

Publish

result = deliverd.publish(title="Weekly figures", content=html, audience=["Acme"])
if result.held:
    print("Waiting on a person before anyone can open it")
else:
    print(result.url)

Restarting safely

Give a request your own external_id and a restarted agent finds the request it already filed rather than asking two people about one act:

deliverd.approve(title="Refund £1,240", external_id=f"refund-{invoice.id}")

That is best effort — two concurrent calls can still both find nothing. For the guarantee, pass idempotency_key, which the server enforces.

Waiting

approval = deliverd.approvals.create(title="Deploy production")
print("Waiting at", approval.url)
approval.wait(on_poll=lambda a: print(".", end=""), timeout="30m")

Polling backs off — a quick first look so a fast decision reads as one, then somebody's afternoon rather than a progress bar. Pass a threading.Event as cancel to stop a wait from another thread.

Every request expires, so a wait always ends: 24 hours unless you pass expires_in.

Durations are seconds here, not milliseconds. timeout=30 is half a minute, because time.sleep takes seconds and so does everything else in Python. The string forms — "30m", "4h", "2d" — and timedelta mean the same thing in every Deliverd SDK. (The TypeScript package reads a bare number as milliseconds, for the mirror-image reason.)

Errors

from deliverd import DeliverdError, DeliverdTimeoutError

try:
    deliverd.approve(title="…")
except DeliverdTimeoutError:
    ...                      # you stopped waiting; the request is still open
except DeliverdError as err:
    err.code                 # the API's own code: self_approval, rate_limited…
    err.is_auth              # 401 or 403
    err.is_rate_limit        # 429
    err.is_not_found         # 404
    err.is_conflict          # 409 — re-read and try again
    err.hint                 # the one thing to do next

Messages say what happened and what to do about it, rather than handing you a code to search for.

Webhooks

from deliverd import construct_event, SIGNATURE_HEADER

@app.post("/webhooks/deliverd")
def receive():
    event = construct_event(
        secret=os.environ["DELIVERD_WEBHOOK_SECRET"],
        body=request.get_data(as_text=True),     # the RAW body
        header=request.headers.get(SIGNATURE_HEADER),
    )
    if event.event == "approval.approved":
        ...

body must be the raw request body. A re-encoded object is a different string and will never verify — in FastAPI that is (await request.body()).decode(), in Django request.body.decode().

construct_event raises rather than returning None, because a route that treats an unverifiable body as "no event" answers 200 and tells whoever is posting that everything is fine.

Lists

deliverd.approvals.list()                    # every one, following the cursor
deliverd.approvals.list(limit=20)            # one page of twenty
page = deliverd.approvals.list_page(limit=20)
page.items, page.next_cursor

A limit means one page. No limit means all of them. A full page and a complete list are otherwise indistinguishable until somebody counts.

Publishing (maintainers)

Tag and push; GitHub Actions builds, re-runs the suite on Python 3.9, and uploads through PyPI Trusted Publishing — there is no API token anywhere.

git tag python-v0.1.0 && git push origin python-v0.1.0

Flows

A flow is the thread that ties one job's asks together. The handle is the reason it is in this SDK rather than only in the API:

flow = deliverd.flows.resume(title="Q3 close", external_id=f"close-{period}")

flow.approve(title="Sign off the trial balance", approvers=["Partners"])
flow.collect(title="Any disposals?", respondents=["finance@firm.example"],
             questions=[{"prompt": "Any disposals in the quarter?", "kind": "boolean"}])
flow.close()

Each of those is the ordinary call with the flow already named. A flow_id passed by hand at four call sites is one that gets left off the fifth, and that step is then missing from the history with nothing to say so.

resume() is what a restartable agent wants: the flow for this reference, started if it does not exist. flow.evidence() is the whole record — every step, who was asked, what they said — for when somebody asks you to show your working.

Comments

The other direction of publishing: what readers send back.

for thread in deliverd.comments.open():       # across every report
    print(thread.report_title, "—", thread.quote)

for thread in deliverd.comments.list(report.id):
    for m in thread.messages:
        print(m.author, m.body)

deliverd.comments.create(report.id, body="Fixed in v4.", thread_id=thread.thread_id)
deliverd.comments.resolve(thread.thread_id, note="Fixed in v4")

open() answers "is anything waiting on me?" without naming a report first. resolve() takes a thread id alone — resolving is keyed on the thread, not on the report it sits under.

Schedules

deliverd.schedules.create(report_id=report.id, agent_id=agent, recurrence="0 9 * * 5")
for s in deliverd.schedules.list(due_only=True):
    ...   # publish, then the sweep re-times it

recurrence is a five-field cron expression in UTC that has to have a next occurrence — 0 0 30 2 * is well-formed and never arrives, and is refused rather than stored. Nothing here publishes anything: the sweep marks a schedule due and your agent does the work.

Where this is going

Named rather than left to be discovered:

• The async client runs on threads. See below. It is real concurrency and it is not a native async transport.

The two SDKs otherwise offer the same thing, and that is checked in this repository's own test suite so it cannot quietly stop being true.

Awaiting instead

import asyncio
from deliverd.aio import AsyncDeliverd

deliverd = AsyncDeliverd()

async def main():
    decision = await deliverd.approve(title="Refund £1,240 to Acme", approvers=["Finance"])
    if decision.approved:
        await refund()

    # Six asks that wait together rather than in turn.
    outcomes = await asyncio.gather(*[
        deliverd.approve(title=f"Refund {inv}") for inv in invoices
    ])

asyncio.run(main())

Same names, same arguments, same return types. A handle that comes back is an async handle, so await (await deliverd.approvals.create(...)).wait() does not quietly block your event loop.

How it works, because it matters. Every call runs the synchronous one on a worker thread. There is no native async transport and there is not going to be one: this package has no dependencies — that is checked, and it is most of why anyone trusts it — and the standard library has no async HTTP client. Writing HTTP/1.1 and TLS over asyncio streams by hand would be several hundred lines of protocol in a package whose whole pitch is that there is nothing to audit.

So what you get is ergonomics and concurrency; what it costs is one thread per call in flight, parked on a poll. That is right for the tens of concurrent asks an agent makes and wrong for thousands. If you are running thousands, call the REST API directly with the async client you already have.

deliverd.webhooks is not awaited: verification is a pure function over bytes you already hold, and to_decision(), to_outcome() and to_collected() are the same — no request, so no await.

Also

• REST — docs/api.md, and an OpenAPI document at /openapi.json
• MCP — a remote server any agent can connect to, at deliverd.dev/api/mcp
• TypeScript — @deliverd/sdk, the same shapes with await
• CLI — npm i -g deliverd, for a build step

MIT.