# Python SDK

> Send email and WhatsApp messages and manage contacts from Python with the official Nexiom Connect SDK.

`nexiom-connect` is the official SDK for Python. It works in Django, Flask, FastAPI, Celery workers, or plain scripts, has a sync and an async client, and retries safely for you.

## Install

Python 3.10 or later is required. The SDK uses httpx and anyio.

```shell
pip install nexiom-connect
```

## Initialize the client

Create one client on your server and reuse it: it keeps connections open and is safe to share between threads. It reads the API key from `NEXIOM_API_KEY`; never ship the key to the browser.

```python
from nexiom_connect import NexiomConnect

nexiom = NexiomConnect()  # Reads NEXIOM_API_KEY.
```

| Parameter | Type | Description |
| --- | --- | --- |
| `api_key` | str · optional | Your API key, starting with `nc_`. Defaults to the `NEXIOM_API_KEY` environment variable. |
| `timeout` | float · optional | Total deadline per call in seconds, including retries. Default `30`. |
| `max_retries` | int · optional | Extra attempts for reads, sends, and cancels. Default `2`. |
| `base_url` | str · optional | API root. Default `https://api-connect.nxiom.com/api`. |
| `http_client` | httpx.Client · optional | An httpx client, for proxies, instrumentation, or tests. The SDK never closes it. |

Pass options by name, for example `NexiomConnect(timeout=10)`. Close the client with `nexiom.close()` or a `with` block when your program is done with it.

## Use the async client

`AsyncNexiomConnect` has the same methods; await them. It runs on asyncio and trio, and takes an `httpx.AsyncClient` as `http_client`.

```python
from nexiom_connect import AsyncNexiomConnect

async def send_welcome(address: str) -> str:
    async with AsyncNexiomConnect() as nexiom:
        email = await nexiom.emails.send(
            from_="hello@mail.example.com",
            to=address,
            subject="Welcome",
            text="Thanks for joining us.",
        )

    return email["messageId"]
```

## Send an email

Arguments are keyword-only and snake_case. `from` is a Python keyword, so the sender is `from_`. Methods return the response data as a `dict`.

```python
email = nexiom.emails.send(
    from_="hello@mail.example.com",
    from_name="Acme",
    to="reader@example.com",
    subject="Your receipt",
    html="<p>Thanks for your order.</p>",
    text="Thanks for your order.",
    idempotency_key=f"receipt-{order.id}",
)

print(email["messageId"])
```

Pass an `idempotency_key` that identifies the email in your system. If you omit it, the SDK generates one per call, which protects its own retries but not a retry of your whole operation. See [Sending email](/docs/learn/sending#idempotency-keys).

To send a template, pass `template_id` and `template_variables` instead of `subject` and content.

## Schedule an email

Pass `scheduled_at` as a timezone-aware `datetime` or an ISO 8601 timestamp with an offset, up to 30 days ahead. A naive `datetime` is refused, since its moment is ambiguous. Keep the returned `messageId` to move or cancel the send before it goes out.

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

now = datetime.now(timezone.utc)

email = nexiom.emails.send(
    from_="hello@mail.example.com",
    to="reader@example.com",
    subject="Your trial ends soon",
    text="Your trial ends soon.",
    scheduled_at=now + timedelta(days=1),
)

# Move it one day later, or cancel it before it is sent.
nexiom.emails.reschedule(email["messageId"], scheduled_at=now + timedelta(days=2))
nexiom.emails.cancel(email["messageId"])
```

See [Reschedule](/docs/api/emails/reschedule) and [Cancel](/docs/api/emails/cancel) for the rules.

## Read email logs

`emails.list` and `emails.get` need a full-access key. `list` pages with a cursor; `get` takes a delivery ID, not a message ID.

```python
page = nexiom.emails.list(status="bounced", limit=50)

if page["items"]:
    delivery = nexiom.emails.get(page["items"][0]["id"])

    print(delivery["recipient"], delivery["delivery_reason"])

# nextCursor is None on the last page.
if page["nextCursor"] is not None:
    nexiom.emails.list(status="bounced", cursor=page["nextCursor"])
```

See [List emails](/docs/api/emails/list) and [Retrieve an email](/docs/api/emails/retrieve) for every field.

## Manage contacts

Define a contact property before you store a value for it.

```python
nexiom.contacts.properties.create(name="plan", type="string")

contact = nexiom.contacts.create(
    email="reader@example.com",
    first_name="Sam",
    properties={"plan": "pro"},
)

nexiom.contacts.update(
    contact["id"],
    email="reader@example.com",
    email_status="unsubscribed",
)

page = nexiom.contacts.list(search="example.com", limit=50)
```

`contacts.get`, `contacts.delete`, and `contacts.properties` are available too. Responses use snake_case fields such as `first_name`. On updates, `user_id=None` clears the external user ID. See the [Contacts API](/docs/api/contacts/create) for every field.

## Send a WhatsApp message

`messaging.send` sends on every messaging channel; `channel` selects the arguments. For WhatsApp, send an approved template from your connected number, named by its phone number in `from_`. Requires a key with the `whatsapp.send` permission.

```python
message = nexiom.messaging.send(
    channel="whatsapp",
    from_="+213555000111",
    to="+213555123456",
    template={
        "id": "mtp_example",
        "variables": {"body": {1: "Amina", 2: "A-1234"}},
    },
    idempotency_key="order-A-1234-shipped",
)

current = nexiom.messaging.get(message["id"])

print(current["status"])
```

See [Sending WhatsApp messages](/docs/whatsapp/sending).

## Verify webhooks

`webhooks.construct_event` checks a webhook request's signature and timestamp, and returns the event. Pass the raw body, not parsed JSON. `Webhooks()` needs no API key. In a Django view:

```python
import os

from django.http import HttpRequest, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from nexiom_connect import NexiomWebhookVerificationError, Webhooks

webhooks = Webhooks()

@csrf_exempt
@require_POST
def nexiom_webhook(request: HttpRequest) -> HttpResponse:
    try:
        event = webhooks.construct_event(
            request.body,
            request.headers,
            os.environ["NEXIOM_WEBHOOK_SECRET"],
        )
    except NexiomWebhookVerificationError as error:
        return HttpResponse(error.reason, status=400)

    if event["type"] == "email.bounced":
        print(event["data"]["recipient"])

    return HttpResponse(status=204)
```

It raises `NexiomWebhookVerificationError` when the request is not genuine. In Flask, pass `request.get_data()` and `request.headers`; in FastAPI, `await request.body()` and `request.headers`. See [Verify webhook requests](/docs/webhooks/verify-webhooks-requests) for a complete FastAPI route.

## Handle errors

API and transport failures raise `nexiom_connect.NexiomError`.

```python
from nexiom_connect import ErrorKind, NexiomError

try:
    email = nexiom.emails.send(**message, idempotency_key=idempotency_key)
except NexiomError as error:
    if error.kind is ErrorKind.API and error.status == 429:
        ...  # Rate limited: try again later with the same idempotency key.

    logger.error("%s %s %s %s", error.status, error.code, error, error.request_id)
    raise
```

| Attribute | Description |
| --- | --- |
| `kind` | `ErrorKind.API` for an error response; otherwise `NETWORK`, `TIMEOUT`, or `PROTOCOL`. |
| `status` | HTTP status, when the API responded. |
| `code` | Machine-readable error code, when present. |
| `request_id` | The request's correlation ID. Include it when contacting support. |
| `details` | Additional error details, such as validation failures. |
| `response` | Status, headers, request ID, and the send's `idempotency_key`. |

Invalid arguments — for example a missing `from_`, or an unknown argument such as `fromName` — raise `NexiomValidationError` or `TypeError` before any request is made. Errors can be pickled, so they cross process boundaries in Celery.

## Retries and timeouts

The SDK retries reads, sends and cancels on `408`, `429`, and `5xx` responses and on network failures, honouring `Retry-After`. Sends are retried with the same idempotency key, so a retry never sends twice. Other writes, such as creating a contact or rescheduling an email, are not retried automatically. If `Retry-After` asks for a longer wait than the remaining timeout, the SDK raises that error straight away. The timeout covers every attempt and reading the response.

Override per call by passing `timeout=...` or `max_retries=...` to any method.

## Types

The package ships type information and passes `mypy --strict`. Responses are typed with `TypedDict`s from `nexiom_connect.types`, such as `SendEmailResponse`, `Contact`, `Message`, and `WebhookEvent`, so your editor completes field names and your type checker catches typos.

---

Source: https://connect.nxiom.com/docs/examples/python