Sending & receiving

One handler receives every inbound message, whatever the channel; you call message.reply() and Caspian routes it back on the source channel, in the right thread. listen() runs that loop forever and survives errors. This page is the whole runtime: the handler, the message object, and the ways to send.

The loop

Register one on_message handler, connect whatever channels you want, and call listen(). Every inbound message (a Slack thread, an SMS, an email, a Discord DM) arrives at the same handler. You never branch on channel to reply; reply() already knows where the message came from.

from caspian_sdk import CommClient

client = CommClient()               # reads COMM_API_KEY / COMM_BASE_URL

client.connect_email()              # connect any channels you like

@client.on_message
def handle(message):
    message.reply(f"You said: {message.text}")

client.listen()                     # blocks, dispatching forever

import { CommClient } from "caspian-sdk";

const client = new CommClient();           // reads COMM_API_KEY / COMM_BASE_URL

await client.connectEmail();               // connect any channels you like

client.onMessage(async (message) => {
  await message.reply(`You said: ${message.text}`);
});

await client.listen();                     // runs forever; pass an AbortSignal to stop

You can register more than one handler; each is called for every message, in registration order. A handler that raises is logged and skipped, so one bad message never stops the loop (more on that under listen() below).

The message object

Every inbound message is the same shape on every channel. The fields are read-only; use them to route your agent's logic and to key its memory.

Replying

message.reply() sends back on whichever channel the message arrived from, threaded into the same conversation, with no channel, recipient, or thread id to pass. Give it text, or html for channels that render it (email), or both.

@client.on_message
def handle(message):
    message.reply("plain text works everywhere")

    # HTML where the channel renders it (e.g. email); text is the fallback
    message.reply(
        text="Your report is ready.",
        html="<p>Your <b>report</b> is ready.</p>",
    )

client.onMessage(async (message) => {
  await message.reply("plain text works everywhere");

  // HTML where the channel renders it (e.g. email); text is the fallback
  await message.reply(
    "Your report is ready.",
    "<p>Your <b>report</b> is ready.</p>",
  );
});

Caspian formats the reply the way each platform expects and dedupes within a run, so a handler can't accidentally reply to the same message twice. If you need dedup across restarts, persist your own cursor. See listen() below.

Typing indicators

On channels that support it (Discord, Telegram), Caspian fires a "thinking…" indicator automatically before your handler runs, so the human sees the agent is working. It's best-effort and a no-op where the platform has none. No code required.

An indicator only lasts about 5 to 10 seconds. For long work (an LLM call, a tool run), call message.typing() again to keep it alive.

@client.on_message
def handle(message):
    answer = ""
    for step in run_my_agent(message.text):   # slow, multi-step work
        message.typing()                      # keep the indicator alive
        answer = step
    message.reply(answer)

client.onMessage(async (message) => {
  let answer = "";
  for await (const step of runMyAgent(message.text)) {   // slow work
    await message.typing();                               // keep it alive
    answer = step;
  }
  await message.reply(answer);
});

Instant acknowledgement

Typing indicators only exist on some channels. On the ones without them (X, SMS, email), pass ack to listen() and Caspian sends a short reply the moment a message arrives, before your handler runs, so the human knows the agent is working. Your real answer follows when the handler finishes.

client.listen(ack="On it, one moment…")   # sent instantly; the real reply follows

await client.listen({ ack: "On it, one moment…" });   // sent instantly; real reply follows

listen()

listen() is the resilient forever-loop. It polls the ordered event stream, dispatches each inbound message to your handlers, and is built to run for the lifetime of the agent:

• Handler errors are caught. An exception in one handler is logged and skipped; the loop keeps going and the next message is dispatched.
• Failed polls retry with backoff. A network blip or gateway restart doesn't crash the loop; it retries with exponential backoff (up to max_backoff, default 30s).
• It never exits on error. In Python only KeyboardInterrupt / SIGINT stops it. In JS you pass an AbortSignal to stop it cleanly.

# Defaults: start from the newest event, poll every 1s, back off to 30s max
client.listen()

# Resume from a persisted cursor so restarts don't miss messages
client.listen(from_seq=my_saved_seq)

// Stop the loop from elsewhere with an AbortController
const controller = new AbortController();
process.on("SIGINT", () => controller.abort());

await client.listen({ signal: controller.signal });

// Or resume from a persisted cursor
await client.listen({ fromSeq: mySavedSeq });

Draining once with dispatch_pending()

When you want to run your own loop (a cron tick, a serverless invocation, a test), dispatch_pending() processes every event currently available once and returns the last seq it saw. Like listen(), it catches per-message handler errors, so it always drains the queue and advances the cursor even if a handler fails. Persist the returned cursor and pass it back next time.

cursor = load_cursor()                      # your own persistence
cursor = client.dispatch_pending(after_seq=cursor)
save_cursor(cursor)                         # resume here next tick

let cursor = loadCursor();                   // your own persistence
cursor = await client.dispatchPending(cursor);
saveCursor(cursor);                          // resume here next tick

Sending proactively

Replies are reactive: they answer an inbound message. To speak first, there are two calls, each gated by a capability the connection must have.

send_message(): into an existing conversation

Push a message into a conversation you already have an id for, without an inbound message to reply to (e.g. a follow-up, a scheduled nudge). Needs the SEND capability on the connection.

client.send_message(conversation_id, text="Your order shipped.")

await client.sendMessage(conversationId, "Your order shipped.");

initiate(): cold-start a new conversation

Reach a recipient who has never messaged you, on a connection that allows it (iMessage and SMS can cold-start; WhatsApp on Caspian's network needs an approved template; X is reactive-only and cannot). Needs the INITIATE capability.

client.initiate(connection_id, recipient="+15551234567", text="Hi, it's your agent.")

await client.initiate(connectionId, "+15551234567", "Hi, it's your agent.");

Reading the raw event stream

listen() and dispatch_pending() both sit on top of events(), which returns the underlying event records strictly ordered by seq. Read it directly when you want your own dispatch: a different queue, batching, or filtering by type.

batch = client.events(after_seq=cursor, limit=100)
for event in batch:
    cursor = event["seq"]
    if event["type"] == "message.received":
        handle_yourself(event["data"])

# Filter server-side by type
sent = client.events(type="message.sent")

const batch = await client.events({ afterSeq: cursor, limit: 100 });
for (const event of batch) {
  cursor = event.seq;
  if (event.type === "message.received") {
    handleYourself(event.data);
  }
}

// Filter server-side by type
const sent = await client.events({ type: "message.sent" });

Next steps