Python SDK

CommClient(
    api_key: str | None = None,      # else COMM_API_KEY (env or ./.env)
    base_url: str | None = None,     # else COMM_BASE_URL, default http://127.0.0.1:8000
    http: httpx.Client | None = None,
    timeout: float = 30.0,           # per-request timeout, seconds
)

client.close() -> None              # close the underlying HTTP client

from caspian_sdk import CommClient

client = CommClient()                      # reads COMM_API_KEY / COMM_BASE_URL
# or explicitly:
client = CommClient(api_key="comm_...", base_url="https://api.trycaspianai.com")

from caspian_sdk import CommError

try:
    client.connect_telegram(bot_token="bad")
except CommError as e:
    print(e.status_code)   # int, e.g. 400
    print(e.detail)        # str, gateway's message

message.reply(text: str | None = None, html: str | None = None) -> dict
# Reply on whichever channel this message arrived from (auto-threaded).

message.typing() -> None
# Show a 'thinking…' indicator on the channel (Discord/Telegram; no-op
# elsewhere). Fired automatically before your handler runs; call again
# during long work to keep it alive.

connect_email(
    customer_id: str | None = None,
    agent_id: str | None = None,
    domain: str | None = None,     # verified custom domain
    username: str | None = None,   # exact local part (custom domains only)
    **kwargs,
) -> dict
# Connect an email inbox. Returns a connection with an "address".

inbox = client.connect_email()
print(inbox["address"])            # e.g. agent-xxxx@trycaspianai.com

connect_telegram(
    bot_token: str,                # from @BotFather
    customer_id: str | None = None,
    agent_id: str | None = None,
    **kwargs,
) -> dict

connect_phone(
    customer_id: str | None = None,
    agent_id: str | None = None,
    provider=None,                 # telnyx / twilio / modem / agentphone
    **kwargs,                      # provider credentials for BYO numbers
) -> dict
# Omit provider for the deployment default.

connect_whatsapp(
    customer_id=None, agent_id=None,
    provider=None,                 # "twilio-whatsapp" or "meta-whatsapp"
    **kwargs,
) -> dict

start_whatsapp_onboarding(
    customer_id=None, agent_id=None,
    display_name=None, capabilities=None,
) -> dict
# Returns {"session", "launcher_url", "expires_in"}. Hand launcher_url
# to the WhatsApp Business account owner; poll get_connection() until active.

connect_imessage(customer_id=None, agent_id=None, **kwargs) -> dict
# iMessage line (provider: agentphone-imessage).

connect_rcs(customer_id=None, agent_id=None, **kwargs) -> dict
# RCS Business Messaging sender (provider: twilio-rcs).

install_discord(
    customer_id=None, agent_id=None,
    display_name=None,             # bot's name in that server
    **kwargs,
) -> dict
# Shared bot, no token. Returns a connection with an authorize_url.

connect_discord(
    bot_token: str | None = None,  # from discord.com/developers
    webhook_url: str | None = None,# OR a channel webhook (no bot)
    username: str | None = None,
    avatar_url: str | None = None,
    customer_id=None, agent_id=None, **kwargs,
) -> dict

install_slack(
    customer_id=None, agent_id=None,
    display_name=None, icon_url=None,  # post under your own name + icon
    **kwargs,
) -> dict
# Shared app ("Add to Slack"). Returns a connection with an authorize_url.

connect_slack(
    slack_client_id: str | None = None,
    slack_client_secret: str | None = None,
    slack_signing_secret: str | None = None,
    customer_id=None, agent_id=None, **kwargs,
) -> dict
# Bring your own app (api.slack.com/apps). Returns an authorize_url.

install_x(customer_id=None, agent_id=None, **kwargs) -> dict
# Returns a connection with an authorize_url ("Sign in with X").

connect_x(
    access_token: str,
    user_id: str,                  # numeric id (before the dash in the token)
    access_secret: str | None = None,
    username: str | None = None,
    customer_id=None, agent_id=None, **kwargs,
) -> dict
# The account must be labelled "Automated" in X settings.

connect_instagram(customer_id=None, agent_id=None, **kwargs) -> dict
# Instagram DM install (OAuth). Returns an authorize_url.

connect_facebook(customer_id=None, agent_id=None, **kwargs) -> dict
# Facebook Messenger install (OAuth). Returns an authorize_url.

update_branding(
    connection_id: str,
    display_name=None, icon_url=None,
) -> dict
# Change the name/icon the agent posts under, after connecting. No re-install.

@client.on_message                          # register a handler (repeatable)
def handle(message):
    message.reply(f"You said: {message.text}")

client.listen(ack="On it, one moment…")     # instant ack, then your real reply

on_message(handler) -> handler
# Register a message handler. The same handler answers every channel.

reply(message_id: str, text: str | None = None, html: str | None = None) -> dict
# Reply to a message by id (Message.reply wraps this).

typing(message_id: str) -> dict
# Show a 'thinking…' indicator on that message's channel. Best-effort.

listen(from_seq=None, poll_interval=1.0, max_backoff=30.0, ack=None) -> None
# Poll forever, dispatching inbound messages. Only SIGINT stops it.
# ack: instant acknowledgement reply sent before your handler runs
#   (e.g. "On it, one moment…"); the real answer follows. Best on
#   channels with no typing indicator (X, SMS, email).

dispatch_pending(after_seq: int = 0) -> int
# Drain all currently available events once; returns the last seen seq.

events(after_seq=0, limit=100, type: str | None = None) -> list[dict]
# Raw event feed. Cursor-paged by seq; filter by type.

send_message(conversation_id: str, text=None, html=None) -> dict
# Send into an existing conversation (needs Capability.SEND).

initiate(connection_id: str, recipient: str, text: str) -> dict
# Cold-start a new conversation (needs Capability.INITIATE, user account).

backfill(conversation_id: str, limit: int = 50) -> dict
# Pull history from before the connection (needs Capability.BACKFILL).

behavior_prompt() -> str
# Combined guide for every channel you've connected. Empty string if none.

channel_guide(channel: str) -> str
# Guide for a single channel, e.g. "slack", "discord".

system_prompt = "You are Acme's support agent.\n" + client.behavior_prompt()

create_customer(name: str) -> dict
# Create a customer scope. Pass its id as customer_id to connect_*().

create_agent(name: str) -> dict
# Create an agent scope. Pass its id as agent_id to connect_*().

get_connection(connection_id: str) -> dict
# Fetch one connection (poll this while an install goes active).

list_conversations(connection_id: str | None = None) -> list[dict]
# All conversations, or just one connection's.

list_messages(conversation_id: str) -> list[dict]
# Every message in a conversation.

channels() -> list[dict]
# Configured transports and their capabilities.

set_webhook(url: str, secret: str | None = None) -> dict
# Receive events by push instead of (or alongside) polling.

get_webhook() -> dict
# The currently registered webhook.

test_email(
    text="Hello from the comm test sender.",
    subject="Test email",
    connection_id: str | None = None,
) -> dict
# Inject a message into an email connection to watch your handler run.

add_domain(domain: str) -> dict
# Register a custom subdomain; returns DNS records to add at the registrar.

list_domains() -> list[dict]
get_domain(domain_id: str) -> dict
# Poll get_domain() until the domain is active, then use it in connect_email().