commb-agent

Official Python SDK for the CommB platform — connects standalone CommB bot engines to the CommB collector dashboard via telemetry tracking and remote config sync.

Installation

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

How it works

CommB collector Dashboard (Supabase)
       ▲▼  commb-agent SDK
CommB Engine (standalone FastAPI bot)

• CommB calls get_config() to pull system prompt, knowledge docs, and catalog from CommB collector.
• CommB calls track() after every message/order to push telemetry back.
• CommB calls sync_conversation() to log 48h active session chat history to CommB collector CRM.
• FastAPI streaming endpoints use stream_chat() for SSE responses to the Telegram Mini App.

Quick Start (Sync — for scripts & workers)

from commb_agent import CommBClient

client = CommBClient(
    api_key="snx_bot_xxxx",
    host="https://commb.app",
)

Quick Start (Async — for FastAPI)

from commb_agent import AsyncCommBClient

client = AsyncCommBClient(
    api_key="snx_bot_xxxx",
    host="https://commb.app",
)

API Reference

get_config() / await client.get_config() — Pull config from CommB collector

# 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",
)

sync_conversation() / await client.sync_conversation() — 48h Chat Transcript Sync

# Sync
client.sync_conversation(
    channel="whatsapp",
    customer_id="+2348012345678",
    messages=[
        {"role": "user", "content": "How much is the blue dress?"},
        {"role": "assistant", "content": "The blue dress is ₦15,000."}
    ]
)

# Async
await client.sync_conversation(
    channel="telegram",
    customer_id="tg_123456",
    messages=[
        {"role": "user", "content": "Is shipping free?"},
        {"role": "assistant", "content": "Yes, on orders above ₦50,000."}
    ]
)

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 CommB Engine Integration

# commb_engine/main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from commb_agent import AsyncCommBClient, ChatMessage

commb = AsyncCommBClient(
    api_key=os.environ["BOT_API_KEY"],
    host=os.environ.get("COMMB_COLLECTOR_URL", "https://commb.app"),
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Pull config from CommB collector on startup
    config = await commb.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 commb.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 commb.stream_chat(message, user_id):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

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

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

Context Manager (Sync)

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

Context Manager (Async)

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

Environment Variables

BOT_API_KEY=snx_bot_xxxx              # From CommB collector Bot Settings
COMMB_COLLECTOR_URL=https://commb.app  # CommB collector host

Response Models (Pydantic v2)

All responses are typed Pydantic models:

from commb_agent import CommBConfigResponse, BotConfig, KnowledgeDoc, CatalogItem, BotInfo

License

MIT — Sannex Tech LTD