Python examples

Use httpx (modern async-friendly client) or requests (synchronous, ubiquitous). The examples below use httpx.

Setup

pip install httpx
export AIGEON_API_KEY=sk_...

Helper module:

# aigeon.py
import os
import httpx

BASE = "https://api.aigeon.ai"

_client = httpx.Client(
    base_url=BASE,
    headers={"X-Api-Token": os.environ["AIGEON_API_KEY"]},
    timeout=15.0,
)

def aigeon(path: str, *, method: str = "GET", json=None):
    r = _client.request(method, path, json=json)
    r.raise_for_status()
    body = r.json()
    if body["code"] != 0:
        raise RuntimeError(f"Aigeon {path}: {body['message']}")
    return body["data"]

Add a contact

aigeon(
    "/public/api/v1/contacts/add-group-members",
    method="POST",
    json={
        "group_id": "sg_abc123",
        "members": [{"email": "alice@example.com", "first_name": "Alice"}],
    },
)

Trigger a flow

aigeon(
    "/public/api/v1/flow/send-email-through-flow",
    method="POST",
    json={
        "flow_id": "flw_welcome",
        "contact_email": "alice@example.com",
        "context": {"order_id": "ord_123", "total_usd": 49.00},
    },
)

Pull campaign stats

stats = aigeon("/public/api/v1/stats/get-flow-campaign-summary?flow_id=flw_weekly")
print(stats)

Async

For high throughput, use httpx.AsyncClient:

import asyncio, httpx

async def push_batch(members):
    async with httpx.AsyncClient(headers={"X-Api-Token": os.environ["AIGEON_API_KEY"]}) as c:
        await c.post(
            "https://api.aigeon.ai/public/api/v1/contacts/add-group-members",
            json={"group_id": "sg_abc", "members": members},
            timeout=30,
        )

asyncio.run(push_batch([{"email": "alice@example.com"}]))