Metadata-Version: 2.5
Name: deliverd
Version: 0.12.0
Summary: Add human approval to any AI agent. Ask a person, wait for the answer, act on it.
Project-URL: Homepage, https://deliverd.dev/developers
Project-URL: Documentation, https://deliverd.dev/developers
Project-URL: Source, https://github.com/davidpreid/deliverd
Project-URL: Issues, https://github.com/davidpreid/deliverd/issues
License: MIT License
        
        Copyright (c) 2026 Deliverd
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: agent,ai,approval,deliverd,human-in-the-loop,mcp
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# deliverd

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

```bash
pip install deliverd
```

```python
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](#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.
- **Inside Microsoft's agent stack.** `deliverd.adapters` answers the tool
  approvals of [Microsoft Agent Framework](#microsoft-agent-framework) and
  [Microsoft Foundry](#microsoft-foundry) with a person's decision, without
  either becoming a dependency.

---

## Before you have an account

```python
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:

```python
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:

```python
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

```python
from deliverd import Deliverd

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

| Variable | What |
|---|---|
| `DELIVERD_API_KEY` | A `dlv_` key from Settings → API tokens, or a `dlvo_` OAuth access token |
| `DELIVERD_BASE_URL` | Override for a self-hosted deployment |
| `DELIVERD_MODE` | `development` to run the loop offline |
| `DELIVERD_DEV_OUTCOME` | `approved` · `rejected` · `timeout` · `question` |

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

| | You want | Call |
|---|---|---|
| **Approval** | May I do this? | `approve()` |
| **Confirmation** | Are you sure? — lower ceremony | `confirm()` |
| **Review** | Is what I made right? | `review()` |
| **Information** | A figure, a date, a choice | `collect()` |
| **Publishing** | People need to read this | `publish()` |

### Approve

```python
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

```python
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

```python
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

```python
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:

```python
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

```python
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

```python
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

```python
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.

## Webhook endpoints

Add, change and remove the endpoints themselves from code rather than under
Admin → Webhooks. The secret comes back from `create()` **once** — no other
call returns it — so store it where your receiver reads it.

```python
endpoint = deliverd.webhook_endpoints.create(
    url="https://hooks.example.com/deliverd",
    events=["approval.approved", "approval.rejected"],  # leave out for everything
)
save_secret(endpoint.secret)

result = deliverd.webhook_endpoints.test(endpoint.id)  # result.delivered, result.response_status
deliverd.webhook_endpoints.update(endpoint.id, enabled=False)
deliverd.webhook_endpoints.delete(endpoint.id)
```

These need a key with `webhooks:manage` held by an owner or an administrator.
An agent key is refused, reads included: an endpoint receives every event in
the organisation.

## Lists

```python
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.

```bash
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:

```python
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.

## Tasks

An agent deep in a debugging session forgets what it set out to do, because
the objective is a paragraph thousands of tokens back. A task keeps it
somewhere the agent can ask for, and a side task keeps a detour from taking
over the thread:

```python
task = deliverd.tasks.create(
    title="Ship SSO",
    objective="Customers can sign in with their own identity provider.",
    external_id="sso-rollout",  # a restart resumes this one
)

side = task.create_side_task(
    title="Token audience", goal="Find why the API refuses the token.", kind="debug"
)
side.add_finding(kind="evidence", title="aud is the SPA client id")
side.complete(
    summary="The token is issued for the wrong audience.",
    suggested_next_action="Request the token for the API audience.",
)

context = side.apply().context  # the result, carried into the parent
context.root_objective          # said first, always
context.reanchor                # one paragraph to say back to yourself
```

`task.context()` is what to read instead of the history, and it lists the
reports already published for the task. `task.graph()` is the whole tree, and
`deliverd.tasks.list(status="active")` lists objectives. A finding of kind
`decision` is a suggestion until a person confirms it;
`task.request_decision_approval(finding_id)` puts it to somebody.

A status report is written from that context and kept with the task:

```python
context = task.context()
html = render_status(context)  # your own: objective, state, decisions, what is open

if context.reports:  # newest first
    deliverd.reports.update(context.reports[0].id, content=html, change_summary="This week")
else:
    deliverd.reports.publish(title="SSO rollout: status", content=html, task_id=task.id)
```

The report is then listed on the task's page and in its context, so the next
status report is a new version of it rather than a second one. A later publish
naming another `task_id` moves it there; leaving it out keeps it where it is.
It needs `tasks:read` on the key.

## Comments

The other direction of publishing: what readers send back.

```python
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

```python
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

```python
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`.

## Microsoft Agent Framework

Mark the tools that need a person as Agent Framework already has you do, and
let Deliverd find that person:

```python
from agent_framework import Agent, tool
from deliverd import Deliverd
from deliverd.adapters.agent_framework import run_with_deliverd_approvals

@tool(approval_mode="always_require")
def refund(order_id: str, amount: int) -> str:
    ...

agent = Agent(client=client, tools=[refund])
session = agent.create_session()

result = await run_with_deliverd_approvals(
    agent,
    "Refund order 1042",
    session,
    Deliverd(),
    approvers=["finance"],
    when=lambda tool, args: tool == "refund" and args["amount"] > 100,
)
print(result.text)
```

Each `function_approval_request` in the run becomes a request in Deliverd,
keyed by its call id (`agent-framework:<call_id>`), so a retried round finds
the request it already made rather than asking twice. The approver decides
from the page, the email, Slack or Teams; the answers go back on the same
session, which is where Agent Framework binds an answer to the request it
issued. MCP tools work the same way: `MCPStreamableHTTPTool(...,
approval_mode="always_require")`, or `{"always_require_approval": [...]}`.

- `when(tool, args)` picks the calls that need a person. The rest are approved
  without asking. Every call when omitted.
- `actions={"refund": "finance.refund"}` puts a tool to that action's gate
  instead, so the organisation's policy decides and a person only when a rule
  says so.
- `timeout="10m"` stops waiting. The run comes back still asking; keep the
  session, and resolving the same result later finds the same requests.
- **A declined call is not run, and the model is told only that it was
  rejected.** Agent Framework's approval response has no field for a reason.
  The person's words are in `resolution.reasons` when you resolve a round
  yourself, and in Deliverd either way.
- **A call approved only with changes is declined**, with the approved values
  in its reason. Agent Framework runs an approved call exactly as it asked
  about it, so running it would spend an approval on something nobody agreed to.

One round, for a loop you own:

```python
from agent_framework import Message
from deliverd.adapters.agent_framework import resolve_approval_requests

resolution = await resolve_approval_requests(result, deliverd, approvers=["finance"], timeout="10m")
if resolution.complete:
    result = await agent.run(Message("user", resolution.responses), session=session)
```

`resolve_approval_requests_sync` is the same without an event loop. In the
harness, `deliverd_approval_rule` is an `auto_approval_rules` entry that lets a
call through without a prompt when it needs nobody — `when` does not select
it, or a gate's rule already allows it:

```python
from agent_framework import ToolApprovalMiddleware
from deliverd.adapters.agent_framework import deliverd_approval_rule

rule = deliverd_approval_rule(deliverd, actions={"refund": "finance.refund"})
agent = Agent(client=client, tools=[refund], middleware=[ToolApprovalMiddleware(auto_approval_rules=[rule])])
```

## Microsoft Foundry

An agent in Foundry Agent Service with an MCP tool that asks first —
`require_approval="always"`, or `{"always": [...]}` for the tools that matter —
stops with `mcp_approval_request` items in its response. This answers them:

```python
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential
from deliverd import Deliverd
from deliverd.adapters.foundry import run_with_deliverd_approvals

project = AIProjectClient(endpoint=PROJECT_ENDPOINT, credential=DefaultAzureCredential())
openai = project.get_openai_client()

response = run_with_deliverd_approvals(
    openai,
    Deliverd(),
    create_kwargs={
        "input": "Close the issues nobody has touched since June",
        "extra_body": {"agent_reference": {"name": agent.name, "type": "agent_reference"}},
    },
    approvers=["platform-team"],
)
print(response.output_text)
```

Each request becomes a request in Deliverd keyed by its id (`foundry:<id>`).
The answers go back as `mcp_approval_response` input items on a new response
with `previous_response_id`, carrying the rest of `create_kwargs` but not
`input` or `conversation`, until the agent asks for nothing more. A declined
answer carries the person's reason in `reason`, the optional field on OpenAI's
`McpApprovalResponse`; Foundry's MCP guide does not show it, so
`send_reason=False` leaves it off if a service refuses it. `when`, `actions`
and `timeout` work as they do for Agent Framework.

`approval_responses_for(response, deliverd, ...)` does one round and returns
the items to send. `run_with_deliverd_approvals_async` and
`approval_responses_for_async` take the async client from
`azure.ai.projects.aio`.

## 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`](https://www.npmjs.com/package/@deliverd/sdk),
  the same shapes with `await`
- **CLI** — `npm i -g deliverd`, for a build step

MIT.
