Python + FastAPI

pip install aelio-sdk fastapi uvicorn websockets

import os
import threading

import uvicorn
from fastapi import FastAPI

from aelio_sdk import Aelio

app = FastAPI()
aelio = Aelio(
    secret=os.getenv("AELIO_SDK_SECRET", "change-me-in-production"),
    url=os.getenv("AELIO_SERVER_URL", "ws://127.0.0.1:3010"),
)


@aelio.expose(
    "getOrderStatus",
    description="Get the status of a customer order",
    params={"orderId": "string"},
    safety="read",
)
async def get_order_status(args, ctx):
    order_id = args.get("orderId")
    customer_id = ctx.get("customerId")
    # Query your database here
    return {
        "orderId": order_id,
        "customerId": customer_id,
        "status": "shipped",
        "tracking": "1Z999AA10123456784",
    }


@aelio.expose(
    "cancelOrder",
    description="Cancel a pending customer order",
    params={"orderId": "string"},
    safety="write",
)
async def cancel_order(args, ctx):
    order_id = args.get("orderId")
    # Cancel in your database
    return {
        "orderId": order_id,
        "customerId": ctx.get("customerId"),
        "status": "cancelled",
    }


@aelio.expose(
    "listOrders",
    description="List customer orders, optionally filtered by status",
    params={"status": "string?", "limit": "integer?"},
    safety="read",
)
async def list_orders(args, ctx):
    orders = [
        {"orderId": "A123", "status": "shipped"},
        {"orderId": "B456", "status": "pending"},
    ]
    status_filter = args.get("status")
    if status_filter:
        orders = [o for o in orders if o["status"] == status_filter]
    limit = args.get("limit", len(orders))
    return {"customerId": ctx.get("customerId"), "orders": orders[:limit]}


@app.get("/health")
async def health():
    return {"ok": True}

def _run_sdk():
    aelio.run()


if __name__ == "__main__":
    sdk_thread = threading.Thread(target=_run_sdk, daemon=True)
    sdk_thread.start()
    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")))

AELIO_SDK_SECRET=change-me-in-production \
AELIO_SERVER_URL=ws://127.0.0.1:3010 \
python main.py

aelio.state("active", description="Fully onboarded customer with full tool access.")
aelio.state("onboarding", description="New user. Setup only.", blocked_tools=["cancelOrder"])

# Push state from API route
@app.post("/api/onboarding/complete")
async def complete_onboarding(user_id: str):
    await aelio.set_customer_state(user_id, "active", "Onboarding completed")
    return {"status": "ok"}

@aelio.on_send
async def deliver_message(msg):
    channel = msg.get("channel")
    to = msg.get("to")
    content = msg.get("content")
    await my_provider.send(to=to, body=content)


@app.post("/whatsapp-webhook")
async def whatsapp_webhook(request: Request):
    body = await request.json()
    from_number = body.get("from")
    text = body.get("text")
    aelio.ingest(channel="whatsapp", from_=from_number, text=text)
    return {"status": "ok"}

import asyncio

@app.on_event("startup")
async def startup():
    asyncio.create_task(aelio.listen())

<script
  src="http://localhost:3010/widget.js"
  data-server-url="http://localhost:3010"
  data-customer-id="user_abc123"
  data-email="user@example.com"
></script>