sannex-agent

Official Python SDK for the Sannex AICB Platform — connects standalone AICB bot engines to the AgentOS dashboard via telemetry tracking and remote config sync.

Installation

pip install sannex-agent
# or
uv add sannex-agent

How it works

AgentOS Dashboard (Supabase)
       ??  sannex-agent SDK
AICB Engine (standalone FastAPI bot)

• AICB calls get_config() to pull system prompt, knowledge docs, and catalog from AgentOS.
• AICB calls track() after every message/order to push telemetry back.
• FastAPI streaming endpoints use stream_chat() for SSE responses to the Telegram Mini App.

Quick Start (Sync — for scripts & workers)

from sannex_agent import SannexClient

client = SannexClient(
    api_key="snx_bot_xxxx",
    host="https://agentos.aicb.sannex.ng",
)

Quick Start (Async — for FastAPI)

from sannex_agent import AsyncSannexClient

client = AsyncSannexClient(
    api_key="snx_bot_xxxx",
    host="https://agentos.aicb.sannex.ng",
)

API Reference

get_config() / await client.get_config() — Pull config from AgentOS

# Sync
config_resp = client.get_config()

# Async
config_resp = await client.get_config()

print(config_resp.config.system_prompt)
print(config_resp.config.model_name)        # "gemini-2.5-flash"
print(len(config_resp.knowledge_docs))      # RAG docs
print(len(config_resp.catalog_items))       # Product catalog

track() / await client.track() — Push telemetry

Non-blocking. Batches and flushes in the background. Never raises.

# Sync (thread-safe, fire-and-forget)
client.track(
    channel="telegram",
    customer_id="tg_123456",
    event="order_created",
    amount=45000.0,
    metadata={"order_id": "ORD-001"},
)

# Async
await client.track(
    channel="whatsapp",
    customer_id="+2348012345678",
    event="message_received",
)

stream_chat() — Stream AI responses (SSE)

# Sync
for chunk in client.stream_chat("What dresses do you have?", user_id="user_123"):
    print(chunk, end="", flush=True)

# Async (FastAPI SSE endpoint)
async for chunk in client.stream_chat("What dresses do you have?", user_id="user_123"):
    yield f"data: {chunk}\n\n"

ping() / await client.ping() — Health check

is_up = client.ping()          # sync
is_up = await client.ping()    # async

get_bot() / await client.get_bot() — Bot identity

bot = client.get_bot()
print(bot.name)      # "Elena Luxe Bot"
print(bot.reseller)  # "Sannex Digital Agency"

FastAPI AICB Engine Integration

# aicb_engine/main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from sannex_agent import AsyncSannexClient, ChatMessage

sannex = AsyncSannexClient(
    api_key=os.environ["BOT_API_KEY"],
    host=os.environ.get("AGENTOS_URL", "https://agentos.aicb.sannex.ng"),
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Pull config from AgentOS on startup
    config = await sannex.get_config()
    app.state.system_prompt = config.config.system_prompt
    app.state.knowledge_docs = config.knowledge_docs
    app.state.catalog_items = config.catalog_items
    yield
    await sannex.close()

app = FastAPI(lifespan=lifespan)

@app.post("/v1/chat")
async def chat(message: str, user_id: str):
    async def event_stream():
        async for chunk in sannex.stream_chat(message, user_id):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    # Track the conversation event
    await sannex.track(
        channel="telegram",
        customer_id=user_id,
        event="message_received",
    )

    return StreamingResponse(event_stream(), media_type="text/event-stream")

Context Manager (Sync)

with SannexClient(api_key="snx_bot_xxxx") as client:
    config = client.get_config()
    # ... use client
# auto-flushes and closes on exit

Context Manager (Async)

async with AsyncSannexClient(api_key="snx_bot_xxxx") as client:
    config = await client.get_config()
    # ... use client

Environment Variables

BOT_API_KEY=snx_bot_xxxx              # From AgentOS Bot Settings
AGENTOS_URL=https://agentos.aicb.sannex.ng  # AgentOS host

Response Models (Pydantic v2)

All responses are typed Pydantic models:

from sannex_agent import SannexConfigResponse, BotConfig, KnowledgeDoc, CatalogItem, BotInfo

License

MIT — Sannex Tech LTD