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

gos = GenieOS()  # reads GENIEOS_API_KEY from os.environ unless passed
res = gos.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():
    gos = AsyncGenieOS()
    res = await gos.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

gos.workspace.get()

gos.keys.list()
gos.keys.get(id_)

gos.templates.list()
gos.templates.get("welcome")
gos.templates.get_schema("welcome")
gos.templates.create(name="Welcome", key="welcome")
gos.templates.compose(prompt="Warm welcome for new Glow subscribers")
gos.templates.render("welcome", variables={...})
gos.templates.send("welcome", to="…", variables={...})

gos.sequences.list()
gos.sequences.get(key_or_id)
gos.sequences.enroll(key_or_id, contact=..., variables=...)
gos.sequence_runs.get(run_id)
gos.sequence_runs.cancel(run_id)

gos.events.emit(type=..., contact=..., metadata=...)

gos.webhooks.list()
gos.webhooks.get(id_)
gos.webhooks.create(url=..., events=[...])
gos.webhooks.update(id_, ...)
gos.webhooks.delete(id_)

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

gos.pages.list()
gos.pages.get(id_or_slug)
gos.pages.compose(id_or_slug, prompt="…")
gos.pages.publish(id_or_slug)
gos.pages.unpublish(id_or_slug)

gos.messaging.kit()
gos.messaging.catalog()
gos.messaging.send(template_key="…", to="…", variables={...})
gos.messaging.list_deliveries(template_key="…")

gos.social.list_networks()
gos.social.refresh_networks()
gos.social.list(status="draft")
gos.social.get(post_id)
gos.social.create(...)
gos.social.schedule(post_id, scheduled_at="…")
gos.social.publish(post_id)
gos.social.delete(post_id)
gos.social.analytics(post_id)

gos.marketing.strategy(detail="summary")
gos.marketing.patch_strategy(...)
gos.marketing.list_icps()
gos.marketing.get_icp(icp_id)
gos.marketing.creation_defaults()
gos.marketing.set_creation_defaults(...)

gos.creations.list()
gos.creations.get(creation_id)
gos.creations.spawn(brief="…")
gos.creations.approve_strategy(creation_id)

gos.lists.list()
gos.lists.get(list_id)
gos.lists.create(name="…")
gos.lists.add_members(list_id, contact_ids=[...])

gos.approvals.list_policies()
gos.approvals.list_pending()
gos.approvals.manage_policy(surface_kind, ...)
gos.approvals.decide(request_id, decision="approve")

gos.links.list()
gos.links.get(link_id)
gos.links.utm_suggestions(field="source")
gos.links.create(
    destination_url="https://…",
    utm={"source": "newsletter", "medium": "email", "campaign": "summer"},
)
gos.links.update(link_id, destination_url="https://…/v2")
gos.links.analytics(link_id=link_id, days=30)
gos.qr.create(encodes={"kind": "shortLink", "linkId": link_id})
gos.qr.render(qr_id, format="pdf", save_to_assets=True)

gos.connectors.catalog()
gos.connectors.list()

gos.audit.list(limit=50)

Per-call options ride on a keyword:

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

Configuration

from genieos import GenieOS

gos = GenieOS(
    api_key="gos_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.

gos.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:
    gos.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["GenieOS-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 = gos.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.

Typed templates via schema + OpenAPI

There is no built-in genieos codegen CLI yet. Prefer:

gos.templates.get_schema("welcome") at runtime for the schema contract, or
generate clients from https://api.genieos.pro/v1/openapi.json.

Sync + async in the same process

Both clients can coexist:

gos = 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:

gos.close()
await amg.aclose()

Or use the context-manager forms:

with GenieOS(api_key=os.environ["GENIEOS_API_KEY"]) as gos:
    gos.events.emit(...)

async with AsyncGenieOS() as gos:
    await gos.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