# Python SDK

`avelto` on PyPI is the official Python SDK. Python 3.9 or newer, sync and asyncio clients with the same surface, typed responses, one dependency (`httpx`).

**pip**

```bash
pip install avelto
```

## Send

```python
import os
from avelto import Avelto

avelto = Avelto(os.environ["AVELTO_API_KEY"])

email = avelto.emails.send(
    from_="Acme <hello@mail.acme.com>",
    to="jane@example.com",
    subject="Your receipt",
    html="<p>Thanks for your order.</p>",
)
print(email["id"])
```

`from_` is the sender, because `from` is a Python keyword. `to`, `cc` and `bcc` take one address or a list. Addresses can be `a@b.com` or `Name <a@b.com>`. Provide `html`, `text` or both. You can also pass `reply_to`, `headers`, `tags` (up to 10) and `attachments` (dicts with `filename` and base64 `content` or an http(s) `url` we fetch, up to 10 files and 7 MB in total). Set `unsubscribe_url` on bulk mail to add the one-click `List-Unsubscribe` headers. To send a stored template instead of a body, pass `template_id` or `template_slug` with `variables` and leave `subject`, `html` and `text` unset (see [Templates](#templates)).

Every method returns the API's JSON as a dict, exactly as the [reference](/docs/api) documents it, so `email["id"]` rather than `email.id`. The dict shapes are `TypedDict`s in `avelto.types`, so an editor knows the keys.

### asyncio

`AsyncAvelto` has the same options and methods, each one awaited. Use it as an async context manager, or call `aclose()`, to release the connection pool.

```python
import os
from avelto import AsyncAvelto

async def welcome(address: str) -> str:
    async with AsyncAvelto(os.environ["AVELTO_API_KEY"]) as avelto:
        email = await avelto.emails.send(
            from_="Acme <hello@mail.acme.com>",
            to=address,
            subject="Welcome to Acme",
            text="Thanks for signing up.",
        )
    return email["id"]
```

### Options

```python
avelto = Avelto(
    api_key,
    base_url="https://api.avelto.dev",  # also read from AVELTO_BASE_URL
    timeout=30.0,                                     # per request, seconds
    max_attempts=3, base_delay=0.3, max_delay=5.0,    # the retry defaults
)
```

The client holds a connection pool. Use it as a context manager, or call `close()`, when you are done with it; one client per process is the normal shape.

## Idempotency

Every `emails.send` call carries an `Idempotency-Key` header. By default it is a fresh random UUID per call, which is what makes the SDK's own retries (below) safe: a retried request can only ever create the one email. Pass your own key, such as an order id, to make retries at your level safe too. A replay returns the id of the email that was already created, with status `200`, and does not send again.

```python
email = avelto.emails.send(
    from_="billing@mail.acme.com",
    to="jane@example.com",
    subject="Receipt #1042",
    text="Thanks for your order.",
    idempotency_key="receipt-1042",  # a retry with the same key returns the same id
)
```

## Retries

The SDK makes up to three attempts at a failed request (the call plus two retries) with exponential backoff and jitter (300 ms, then 600 ms, capped at 5 s, each varied by up to half). On `429` it waits for the `Retry-After` the server sends instead.

What is retried depends on whether the request could have been processed:

| Failure | `GET`, `DELETE`, `emails.send`, `emails.send_batch`, `emails.validate` | Every other `POST` and `PATCH` (create domain, webhook, suppression, template; update template; cancel; retry delivery) |
| --- | --- | --- |
| `429`, `502`, `503` (the server did not process the request) | retried | retried |
| Network error, timeout, `504` (unknown whether it was processed) | retried | not retried |

`emails.send` and `emails.send_batch` sit in the first column because they carry an idempotency key, and `emails.validate` because it changes nothing. The others are not repeated after a network error, because a second `domains.create` could succeed twice, and a repeated `templates.update` would save a duplicate version. Tune it with `max_attempts`, `base_delay` and `max_delay`, or set `max_attempts=1` to turn it off.

```python
# Two retries with the default backoff:
avelto = Avelto(api_key, max_attempts=3, base_delay=0.3, max_delay=5.0)

# No retries: every failure raises at once.
avelto = Avelto(api_key, max_attempts=1)
```

This is also what keeps deploys invisible to you: while an API instance restarts, a request that reaches it is refused before it is read, and the retry lands on the other instance.

## Scheduled send and cancel

`scheduled_at` is an ISO 8601 timestamp, in the future and at most 30 days ahead. A scheduled email can be cancelled until it is sent.

```python
from datetime import datetime, timedelta, timezone

email = avelto.emails.send(
    from_="hello@mail.acme.com",
    to="jane@example.com",
    subject="Your trial ends tomorrow",
    text="...",
    scheduled_at=(datetime.now(timezone.utc) + timedelta(days=1)).isoformat(),
)

cancelled = avelto.emails.cancel(email["id"])  # cancelled["status"] == "cancelled"
```

Cancelling an email that is no longer `scheduled` raises an `AveltoError` with code `not_scheduled`.

## Validate an address

`emails.validate(email)` asks whether an address is worth a send before you spend one on it. Nothing is sent or stored. See [validate an address](/docs/send-email#validate-an-address) for what each result means.

```python
check = avelto.emails.validate("jane@gmial.com")
check["result"]      # "deliverable" | "risky" | "undeliverable"
check["reason"]      # None | "invalid_syntax" | "suppressed" | "no_mail_server" | "disposable" | "possible_typo" | "role_address"
check["suggestion"]  # "jane@gmail.com"
```

## Batch send

`emails.send_batch` sends up to 100 messages in one call. Each message is a dict with the same keys as `send` (`"from"`, or `"from_"` if you prefer to avoid the keyword in a dict literal too). Each is validated, limited and queued exactly as a single send is, and each is accepted or refused on its own, so read `results` rather than assuming the call either worked or did not. `results[i]` lines up with `messages[i]`. The call carries an `Idempotency-Key` like `emails.send`, so a retry after a timeout returns the same ids instead of sending twice.

```python
batch = avelto.emails.send_batch([
    {"from": "billing@mail.acme.com", "to": "jane@example.com", "subject": "Receipt #1042", "text": "Thanks for your order."},
    {"from": "billing@mail.acme.com", "to": "sam@example.com", "subject": "Receipt #1043", "text": "Thanks for your order."},
], idempotency_key="receipts-2026-09-24")

for r in batch["results"]:
    if r["ok"]:
        print(r["index"], r["id"])
    else:
        print(r["index"], r["error"]["code"], r["error"]["message"])
```

## Fetch and list

```python
email = avelto.emails.get(email_id)
print(email["status"])                       # "queued" | "scheduled" | "sent" | "delivered" | "bounced" | "complained" | "failed" | "cancelled"
print([e["type"] for e in email["events"]])  # ["email.queued", "email.sent", "email.delivered"]
```

Lists are newest first and cursor-paginated. Filter by `status`, `tag`, `mode` and `q`, a substring search over recipient, subject and message id (or an exact email id). `mode`: a live key defaults to live and may ask for test; a test key only ever sees test emails.

```python
cursor = None
while True:
    page = avelto.emails.list(status="bounced", tag="onboarding", limit=100, cursor=cursor)
    for e in page["data"]:
        print(e["id"], e["to"], e["subject"])
    cursor = page["next_cursor"]
    if not cursor:
        break
```

## Domains

Add a domain, publish the DNS records it returns, then poll `get` until it is verified. `domains.get` re-checks verification on every call until the domain is verified.

```python
import time

domain = avelto.domains.create("mail.acme.com")
for r in domain["dns_records"]:
    print(r["type"], r["name"], r["value"], f"({r['purpose']})", sep="\t")

# Publish the records, then poll. GET re-checks DNS on every call.
status = domain["status"]
while status == "pending":
    time.sleep(30)
    status = avelto.domains.get(domain["id"])["status"]
print(status)  # "verified" or "failed"
```

```python
domains = avelto.domains.list()["data"]
avelto.domains.delete(domain["id"])
```

## Templates

A template is a stored subject and body with `{{variables}}` in them. Create one, then send it by passing `template_id` or `template_slug` and `variables` instead of a subject and body. A variable the template uses and you do not supply is an error, not an empty string. Every `update` saves a version; `versions` lists them newest first, and `restore` writes an old one forward as a new version, so the history stays append-only. See [Templates](/docs/templates).

```python
tpl = avelto.templates.create(name="Welcome", subject="Hi {{name}}", html="<p>Welcome, {{name}}.</p>")

avelto.emails.send(
    from_="hello@mail.acme.com",
    to="jane@example.com",
    template_slug=tpl["slug"],
    variables={"name": "Jane"},
)

avelto.templates.update(tpl["id"], subject="Hello {{name}}")  # saves a new version
avelto.templates.update(tpl["id"], text=None)                 # None clears a body part; leaving it out keeps it
versions = avelto.templates.versions(tpl["id"])["data"]        # newest first
avelto.templates.restore(tpl["id"], versions[-1]["version"])
```

## Webhooks

Create an endpoint. The signing `secret` is returned once, on creation. Store it.

```python
endpoint = avelto.webhooks.create(
    "https://acme.com/hooks/avelto",
    events=["email.delivered", "email.bounced", "email.complained"],  # optional; defaults to all eight event types
)
print(endpoint["secret"])
```

Every delivery is a JSON POST with an `Avelto-Signature` header of the form `t=<unix seconds>,v1=<hex>`. Verify it against the raw body with `verify_webhook_signature` before you trust the payload, and use the event `id` to de-duplicate. Signatures older than five minutes are rejected; pass `tolerance_seconds=` to change that. In Django the raw body is `request.body`; in Flask it is `request.get_data()`.

```python
import json, os
from flask import Flask, request, abort
from avelto import verify_webhook_signature

app = Flask(__name__)

@app.post("/hooks/avelto")
def avelto_hook():
    body = request.get_data()  # the raw bytes, before any parsing
    if not verify_webhook_signature(os.environ["AVELTO_WEBHOOK_SECRET"], body, request.headers.get("Avelto-Signature")):
        abort(400)
    event = json.loads(body)
    if event["type"] == "email.bounced":
        print("bounced", event["data"]["email_id"], event["data"]["details"])
    return "", 200
```

### Deliveries

```python
endpoints = avelto.webhooks.list()["data"]
endpoint = avelto.webhooks.get(endpoint_id)

page = avelto.webhooks.list_deliveries(endpoint_id, limit=50)
failed = next((d for d in page["data"] if d["status"] == "failed"), None)
if failed:
    avelto.webhooks.retry_delivery(endpoint_id, failed["id"])

avelto.webhooks.delete(endpoint_id)
```

## Suppressions

Addresses that hard-bounce or complain are suppressed automatically and sends to them are rejected. You can manage the list yourself.

```python
page = avelto.suppressions.list(limit=100)
for s in page["data"]:
    print(s["email_address"], s["reason"])  # "bounce" | "complaint" | "manual"

avelto.suppressions.create("unsubscribed@example.com")
avelto.suppressions.delete("unsubscribed@example.com")
```

## Account

`account.get()` returns the account as an integrator sees it: the key's mode and scopes, the sandbox domain with every recipient a sandbox send may reach, and the domains, webhook endpoints and templates that exist. Any scope can read it. Read it first when wiring up a new project; it says what is set up and what is not.

```python
summary = avelto.account.get()
print(summary["key"]["mode"], summary["key"]["scopes"])       # "test" ["emails:send", ...]
print(summary["sandbox"]["domain"], summary["sandbox"]["recipients"])
print(summary["domains"], summary["webhooks"], summary["templates"])
```

## Errors

Every failed request raises `AveltoError` with the HTTP `status`, the API `code`, the `message` and optional `details`, plus `request_id` (the `x-request-id` the API answered with, to quote when you write in) and, on a `429`, `retry_after_seconds` from the `Retry-After` header. A request that gets no response at all (DNS, TLS, timeout) has status `0` and code `network_error`, with the underlying exception as `__cause__`; `is_network_error` is true for exactly those.

```python
from avelto import AveltoError

try:
    avelto.emails.send(from_="hello@mail.acme.com", to="jane@example.com", subject="Hi", text="Hi")
except AveltoError as err:
    if err.code == "domain_not_verified":             # 403: verify mail.acme.com first
        ...
    elif err.code == "recipient_suppressed":          # 422: the address is on your suppression list
        ...
    elif err.code == "validation_error":              # 400: err.details lists the failing fields
        ...
    elif err.is_network_error:                        # status 0: no response (DNS, TLS, timeout)
        ...
    else:
        print(err.status, err.code, err.message, err.request_id)
```

The codes and their meanings are listed in the [API reference](/docs/api#errors).

## Test mode

Keys starting with `av_test_` never send real email. They run the full pipeline and emit the same events and webhooks, so you can build against them without touching an inbox. The simulator recipients `delivered@sandbox.avelto.dev`, `bounced@sandbox.avelto.dev` and `complained@sandbox.avelto.dev` work in both modes. See [Test mode and sandbox](/docs/test-mode).

## Types

Every response shape is a `TypedDict` in `avelto.types`:

```python
from avelto.types import Email, EmailSummary, Domain, WebhookEndpoint, WebhookPayload, Suppression, ValidateEmailResponse
```

---

Rendered page: https://avelto.dev/docs/sdk-python