GenieOS docs

Install

pip install genieos

uv add genieos

poetry add genieos

pdm add genieos

rye add genieos

Requires Python 3.9+ and pulls in httpx>=0.27 and pydantic>=2.6.

from genieos import GenieOS

mg = GenieOS()  # reads GENIEOS_API_KEY from os.environ unless passed
res = mg.templates.send(
    to="ada@example.com",
    template="welcome",
    variables={"first_name": "Ada"},
)
print(res.id)

For async code, swap the import and await everything:

import asyncio
from genieos import AsyncGenieOS

async def main():
    mg = AsyncGenieOS()
    res = await mg.templates.send(
        to="ada@example.com",
        template="welcome",
        variables={"first_name": "Ada"},
    )
    print(res.id)

asyncio.run(main())

The two clients have identical method surfaces. Pick whichever fits your runtime; you can mix them in the same process.

Quick reference

mg.workspace.get()
mg.workspace.usage()

mg.templates.list()
mg.templates.get("welcome")
mg.templates.send(to=..., template=..., variables=...)
mg.templates.send_batch(template=..., recipients=[...])
mg.templates.preview(template=..., variables=...)

mg.sequences.list()
mg.sequences.enroll("onboarding", contact=..., variables=...)
mg.sequences.transition("onboarding", contact=..., to_state=...)
mg.sequences.pause("onboarding", contact=...)
mg.sequences.resume("onboarding", contact=...)
mg.sequences.unenroll("onboarding", contact=...)

mg.events.emit(type=..., contact=..., metadata=...)
mg.events.list(type=..., since=...)

mg.webhooks.list()
mg.webhooks.create(url=..., events=[...])
mg.webhooks.update(id_, ...)
mg.webhooks.delete(id_)
mg.webhooks.rotate_secret(id_)
mg.webhooks.deliveries(id_, status="failed")
mg.webhooks.replay(id_, delivery_id)

mg.brand.list()
mg.brand.get(brand_id="default")

mg.pages.list()
mg.pages.get(id_or_slug)

mg.audit.list(since=..., until=...)

Per-call options ride on a keyword:

mg.templates.send(
    to=..., template=..., variables=...,
    idempotency_key="order:123:confirm",
    timeout=15.0,
    headers={"X-Trace": trace_id},
)

Configuration

from genieos import GenieOS

mg = GenieOS(
    api_key="mg_live_...",                         # required (or via env)
    base_url="https://api.genieos.pro",          # override for staging
    max_retries=5,
    initial_backoff_seconds=0.25,
    timeout=30.0,                                   # per-attempt
    user_agent="my-service/1.4.2",
)

Environment variables read by GenieOS():

Var	Purpose
`GENIEOS_API_KEY`	Bearer token
`GENIEOS_BASE_URL`	Override the API host
`GENIEOS_TIMEOUT`	Per-attempt timeout (seconds)
`GENIEOS_MAX_RETRIES`	Cap on retries

Idempotency

If idempotency_key isn\u2019t supplied, the SDK derives one from the request body. Pass your own for stable, domain-shaped keys. See Idempotency.

mg.templates.send(
    to=customer.email,
    template="order_confirm",
    variables={"first_name": customer.first_name},
    idempotency_key=f"order:{order.id}:confirm",
)

Errors

from genieos import (
    GenieOS,
    GenieOSError,
    GenieOSAuthError,
    GenieOSValidationError,
    GenieOSNotFoundError,
    GenieOSRateLimitError,
    GenieOSServerError,
    GenieOSNetworkError,
)

try:
    mg.templates.send(to=..., template=..., variables=...)
except GenieOSValidationError as e:
    for f in e.fields:
        print(f.path, f.code)
except GenieOSRateLimitError as e:
    sleep(e.retry_after_seconds)

Every error carries request_id, http_status, and code. See Errors.

Webhooks

import os
from flask import Flask, request, abort
from genieos import verify_webhook, GenieOSWebhookVerificationError

app = Flask(__name__)

@app.post("/genieos")
def webhook():
    try:
        event = verify_webhook(
            payload=request.get_data(),
            header=request.headers["MailGenius-Signature"],
            secret=os.environ["GENIEOS_WEBHOOK_SECRET"],
        )
    except GenieOSWebhookVerificationError:
        abort(400)
    handle(event)
    return "", 204

For FastAPI / Starlette use await request.body() and the async-friendly flow is identical.

verify_webhook accepts tolerance_seconds=300 to widen / tighten the replay window.

Typed responses

Every response body is parsed into a Pydantic model so attribute access is type-checked:

res = mg.templates.send(...)
res.id          # str
res.status      # Literal["queued", "sent", "delivered", "bounced", ...]
res.created_at  # datetime
res.template.version  # int

Models use Config.extra = "allow", so forward-compatible additions to the wire shape don\u2019t break your code — new fields ride along on the underlying dict and become typed once the SDK ships an updated model.

Codegen for typed templates

genieos codegen --out app/genieos_types.py

Generated module exports per-template *Variables Pydantic models:

from genieos import GenieOS
from app.genieos_types import WelcomeVariables

mg = GenieOS()
mg.templates.send(
    to="ada@example.com",
    template="welcome",
    variables=WelcomeVariables(first_name="Ada", plan="pro"),
)

mypy / pyright will flag missing required fields and wrong literals at type-check time — your CI sees schema contract violations before they reach production.

Sync + async in the same process

Both clients can coexist:

mg = GenieOS()                # for synchronous code paths
amg = AsyncGenieOS()          # for async tasks / FastAPI endpoints

Internally each client owns its own httpx.Client / httpx.AsyncClient and connection pool, so there\u2019s no cross-talk. Close them on shutdown:

mg.close()
await amg.aclose()

Or use the context-manager forms:

with GenieOS() as mg:
    mg.events.emit(...)

async with AsyncGenieOS() as mg:
    await mg.events.emit(...)

Async webhook verification

verify_webhook is pure CPU — no I/O — so the same function works inside async handlers without an await.

Python SDK