UniOne Python SDK

Official Python library for the UniOne transactional email API.

API reference · Docs for LLMs: llms.txt

On Django, Flask or FastAPI? Install unione-django or unione-flask or unione-fastapi instead: this SDK plus configuration, a mailer and a webhook route.

Install

pip install unione-sdk

Requirements: Python 3.8+. No third-party dependencies.

Get an API key in the control panel and keep it in the environment, never in code:

export UNIONE_API_KEY="your-api-key"

Send an email

import os

import unione
from unione.api.email import EmailSendRequest, MessageObject, MessageObjectRecipientsItem as Recipient
from unione.common import BodyObject

client = unione.client(os.environ["UNIONE_API_KEY"])

response = client.email.send(EmailSendRequest(
    message=MessageObject(
        recipients=[Recipient.from_address("Alex <alex@example.com>")],
        from_email="shop@example.com",
        from_name="Shop",
        subject="Your order",
        body=BodyObject.from_html("<p>Hello, {{to_name}}! Your order has shipped.</p>"),
    ),
))
print(response.job_id)

The sender address must be on a domain verified in your account. No domain yet? Send from the sandbox domain in three steps.

Contents

• Recipes: personalization · attachments · CC and BCC · scheduling · safe retries · errors · webhooks
• No domain yet: the test sandbox domain
• Going to production
• Reference
• Help

Recipes

Every recipe continues the example above: the client and its imports are there.

Personalize for each recipient

Up to 500 recipients per call, each gets a separate copy with its own substitutions. Metadata of a recipient comes back in its webhook events.

client.email.send(EmailSendRequest(
    message=MessageObject(
        recipients=[
            Recipient.from_address("Alex <alex@example.com>", {"order": "1001"}).with_metadata({"user_id": "42"}),
            Recipient.from_address("Maria <maria@example.com>", {"order": "1002"}).with_metadata({"user_id": "43"}),
        ],
        from_email="shop@example.com",
        subject="Order #{{order}}",
        body=BodyObject.from_html("<p>{{to_name}}, order #{{order}} has shipped.</p>"),
    ),
))

Attach a file

import base64

from unione.common import AttachmentsArrayItem

with open("invoice.pdf", "rb") as file:
    content = base64.b64encode(file.read()).decode()

client.email.send(EmailSendRequest(
    message=MessageObject(
        recipients=[Recipient.from_address("alex@example.com")],
        from_email="shop@example.com",
        subject="Invoice",
        body=BodyObject.from_html("<p>Your invoice is attached.</p>"),
        attachments=[AttachmentsArrayItem(type="application/pdf", name="invoice.pdf", content=content)],
    ),
))

CC and BCC

Every address goes to recipients (each gets a copy); the visible ones also go to the headers. audit@example.com is in neither header: a BCC. The rules and limits: CC and BCC.

from unione.common import HeadersObject

client.email.send(EmailSendRequest(
    message=MessageObject(
        recipients=[Recipient.from_address(a) for a in ("alex@example.com", "boss@example.com", "audit@example.com")],
        from_email="shop@example.com",
        subject="Weekly report",
        body=BodyObject.from_html("<p>The report is attached.</p>"),
        headers=HeadersObject(extra={"To": "alex@example.com", "CC": "boss@example.com"}),
    ),
))

Schedule

send_at is UTC, up to 24 hours ahead.

from datetime import datetime, timedelta, timezone

from unione.api.email import MessageObjectOptions

# UTC, "YYYY-MM-DD hh:mm:ss"
in_one_hour = (datetime.now(timezone.utc) + timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")

client.email.send(EmailSendRequest(
    message=MessageObject(
        recipients=[Recipient.from_address("alex@example.com")],
        from_email="shop@example.com",
        subject="Reminder",
        body=BodyObject.from_text("See you tomorrow!"),
        options=MessageObjectOptions(send_at=in_one_hour),
    ),
))

Retry safely

A repeat with the same idempotence_key within a minute is not sent twice: retry network errors, timeouts, 429 and 5xx with it.

client.email.send(EmailSendRequest(
    message=MessageObject(
        recipients=[Recipient.from_address("alex@example.com")],
        from_email="shop@example.com",
        subject="Reset your password",
        body=BodyObject.from_html("<p>Use this link to reset your password.</p>"),
        idempotence_key="password-reset-8f3a",
    ),
))

Handle errors

Recipients the API drops (unsubscribed, invalid, suppressed) are not an error: they are in failed_emails of the response, the rest still get the message.

from unione import ApiError

try:
    client.email.send(request)
except ApiError as e:
    # HTTP status (0: network error or timeout), API error code, message
    print(e.http_status, e.error_code, e.message)

Receive delivery events

Register your HTTPS endpoint once (webhook/set). Then check the signature of every notification (auth: MD5 of the raw body with the auth value replaced by your API key; a wrong one gets 401) and process the events.

import hashlib
import os

from unione.api.webhook import WebhookApi


def handle_webhook(raw_body: str) -> int:
    """raw_body: the body of the POST request as received; returns the HTTP status to answer."""
    callback = WebhookApi.parse_callback(raw_body)
    # auth: MD5 of the raw body with the auth value replaced by your API key
    signed = raw_body.replace(callback.auth or "", os.environ["UNIONE_API_KEY"])
    if hashlib.md5(signed.encode()).hexdigest() != callback.auth:
        return 401

    for user in callback.events_by_user or []:
        for event in user.events or []:
            # event.event_data: email, status (delivered, opened, hard_bounced, spam, ...), job_id, metadata
            pass
    return 200

Answer 200 within 3 seconds and process the events in the background (a queue): otherwise UniOne repeats the request every 10 minutes for 24 hours (retry_count grows), and a handler that failed every request for at least 10 events of a day is turned inactive. The handler must take at least 5 parallel requests of up to 100 events each; deduplicate, a repeat delivers the same events again.

No domain yet: the test sandbox domain

Until your domain is verified, send from the sandbox domain: a test domain every account has, already verified, with DKIM and SPF set. It delivers only to addresses you confirm (the confirmation link comes by email, valid 48 hours), up to 100 emails a day. The sandbox domain is deleted 31 days after it was created, but you can get it again: the next domain/get-sandbox-domain call creates a new one (with a new name, so read it from the call, do not hardcode it).

from unione.api.confirmed_email import ConfirmedEmailRequest

# 1. Your sandbox domain: created on the first call, the same one afterwards
domain = client.domain.get_sandbox_domain().domain

# 2. Confirm your inbox: UniOne sends a link, click it within 48 hours
client.confirmed_email.create(ConfirmedEmailRequest(email="you@example.com"))

# 3. Send from any address on the sandbox domain to the confirmed one
client.email.send(EmailSendRequest(
    message=MessageObject(
        recipients=[Recipient.from_address("you@example.com")],
        from_email=f"noreply@{domain}",
        subject="Hello from UniOne",
        body=BodyObject.from_html("<p>It works!</p>"),
    ),
))

BASE=https://api.unione.io/en/transactional/api/v1

curl -s -X POST $BASE/domain/get-sandbox-domain.json \
  -H "Content-Type: application/json" -H "X-API-KEY: $UNIONE_API_KEY" -d '{}'

curl -s -X POST $BASE/confirmed-email/create.json \
  -H "Content-Type: application/json" -H "X-API-KEY: $UNIONE_API_KEY" \
  -d '{"email": "you@example.com"}'

curl -s -X POST $BASE/email/send.json \
  -H "Content-Type: application/json" -H "X-API-KEY: $UNIONE_API_KEY" \
  -d '{
        "message": {
          "recipients": [
            {
              "email": "you@example.com"
            }
          ],
          "from_email": "noreply@<sandbox-domain>",
          "subject": "Hello from UniOne",
          "body": {
            "html": "<p>It works!</p>"
          }
        }
      }'

Going to production

1. Verify your sending domain. domain/get-dns-records returns the verification record, the DKIM key and the DMARC target; add the records at your DNS provider, then domain/validate-verification-record and domain/validate-dkim.

2. Make sending reliable.

Reference

Sections

API host

The second argument selects the API host (datacenter) or a full base URL.

import unione
from unione.client import HOST_US1

client = unione.client("YOUR_API_KEY", "us1.unione.io")
client = unione.client("YOUR_API_KEY", HOST_US1)

Custom HTTP client / headers

headers adds HTTP headers to every request. transport replaces the HTTP layer: any object with send(method, url, headers, body, timeout) -> (status, body) (body: bytes; or such a function). Wrap the default UrllibTransport for retries or logging, or call requests / httpx in your own one; an exception of the transport becomes ApiError.

import time

import unione
from unione.client import UrllibTransport


class RetryTransport:
    """Logs every request and retries network failures and HTTP 5xx."""

    def __init__(self, inner=None, attempts=3):
        self.inner = inner or UrllibTransport()
        self.attempts = attempts

    def send(self, method, url, headers, body, timeout):
        for attempt in range(1, self.attempts + 1):
            print(method, url)
            try:
                status, raw = self.inner.send(method, url, headers, body, timeout)
            except unione.ApiError:
                if attempt == self.attempts:
                    raise
            else:
                if status < 500 or attempt == self.attempts:
                    return status, raw
            time.sleep(attempt)


client = unione.client("YOUR_API_KEY", transport=RetryTransport(), headers={"X-Request-Id": "42"})

Models with free-form keys (HeadersObject) have an extra dict: its keys go next to the named fields.

from unione.common import HeadersObject

headers = HeadersObject(extra={
    "X-MyHeader": "some data",
    "CC": "copy@example.com",
    "List-Unsubscribe": "<mailto:unsubscribe@example.com>",
})

Error handling

Every method raises ApiError when the API returns an error; it carries the API error code and the HTTP status.

Help

• Integration guides
• AI coding agents: llms.txt, the MCP server, skills: npx skills add https://github.com/unione-repo/unione-skills.git
• Support · Issues
• Contributing: AGENTS.md — the SDK is generated from the OpenAPI spec.