caspian-sdk

Website · PyPI · npm · SKILL.md for agents · Contributing

English · 简体中文

The largest OSS agent frameworks each built 25+ channel adapters — and still spend 8–15% of their issue trackers on channel plumbing. Caspian makes it one handler.

Caspian is an agent communication SDK. Your agent's reasoning decides what to say; Caspian is how it exists on Slack, Discord, Telegram, email, WhatsApp, X, Linear, and beyond — one channels.add() per channel, declarative rules for all of them, threading, webhook verification, and platform quirks handled.

Most agent communication work is agent-to-human, not agent-to-agent. Protocols like A2A and ACP connect agents to each other; Caspian connects your agent to the people it works for, on the channels they already use.

Version 1.0 is a full rewrite. The public surface is Caspian (not the legacy CommClient from 0.6.x). See Migrating from 0.6.x below.

Get started in 30 seconds

Building in a coding agent (Claude Code, Codex, Cursor, Kimi, …)? Paste this — it reads the live guide and does the whole integration for you:

Integrate Caspian so my agent can message people on email, Slack, Discord, Telegram, and more.
Read https://api.trycaspianai.com/SKILL.md and follow it end to end.

That's the fastest path — the guide at /SKILL.md is always current.

Or set it up by hand:

pip install caspian-sdk        # Python 3.10+
npm install caspian-sdk        # TypeScript / Node 18+ / Bun

Get an API key from dashboard.trycaspianai.com, then:

Hosted — Caspian's gateway owns inbound; your process polls for events:

from caspian import Caspian

cx = Caspian(api_key="...")                          # or CASPIAN_API_KEY in .env
cx.channels.add("telegram", bot_token="...")         # Telegram is BYO BotFather token

@cx.on_message({"overlap": "queue", "ack": "On it…"})
def handle(thread, msg, ctx):
    thread.post(f"You said: {msg.text}")

cx.run()   # polls the gateway — Ctrl+C to stop

Self-host — your process, your tokens, no gateway polling:

cx = Caspian()
cx.channels.add("telegram", via="self-host", bot_token="...",
                webhook_url="https://your.server/telegram")

@cx.on_message({"channel": "telegram"})
def handle(thread, msg, ctx):
    thread.post(f"You said: {msg.text}")

# from your HTTP route:
results = cx.handle("telegram", request_body, request_headers)

Discord and Slack can receive over a held-open socket instead of a public webhook — cx.listen("discord") (requires optional extra caspian-sdk[discord]).

TypeScript — same contract:

import { Caspian } from "caspian-sdk"

const cx = new Caspian()

await cx.channels.add("telegram", {
  via: "self-host",
  botToken: process.env.TELEGRAM_BOT_TOKEN!,
  webhookUrl: "https://your.server/telegram",
})

cx.onMessage({ channel: "telegram", overlap: "queue" }, async (thread, msg) => {
  await thread.post(`You said: ${msg.text}`)
})

// POST your webhook route → cx.webhooks.telegram(req)

Adding a channel is one more channels.add() call — handler rules stay the same.

CLI

The rewrite CLI lives in packages/cli (TypeScript + Bun). It is a thin client of the same SDK surface — catalog discovers, call invokes:

caspian init                 # mint a key → ~/.caspian/.env or project .env
caspian channels add telegram
caspian channels add telegram --via self-host --bot-token "$TG" \
  --webhook-url https://myapp.example.com/hook
caspian call post --thread telegram:123:456 --text "shipping now"
caspian threads tail telegram:123:456

See packages/cli/README.md for the full command map.

Delete your adapter layer

Without Caspian With Caspian

# slack_bolt app + socket handler
# discord.py client + intents + reconnect
# python-telegram-bot + webhook server
# smtplib/imap polling + threading logic
# 4 auth flows, 4 payload shapes,
# 4 retry/backoff paths, 4 dedup caches,
# per-channel identity bugs...
# ~1,500 lines before your agent
# says a single word

cx.channels.add("email", via="self-host", ...)
cx.channels.add("telegram", via="self-host", bot_token=TG, webhook_url=URL)
cx.channels.add("slack", via="self-host", bot_token=SLACK, ...)

@cx.on_message({"overlap": "queue"})
def handle(thread, msg, ctx):
    thread.post(agent(msg.text))

cx.run()          # hosted
# or cx.listen("slack") / cx.handle(channel, body, headers)

Using a coding agent? Point it at SKILL.md — it can do the entire integration for you.

The problem

Every agent team ends up rebuilding the same four things — and none of them make the agent smarter.

1. You own infrastructure you never wanted. Writing the Slack bot is a weekend; owning it is forever. Session/auth desync, reconnect loops, silent connection failures, payload changes on every platform version bump. The pain isn't send() — sending is a solved call. The pain is the lifecycle. The largest OSS agent frameworks each maintain 25+ channel adapters in-tree and still spend 8–15% of their issue trackers on channel plumbing. (We measured 42 open-source agent projects before writing a line of this code.)

2. Communication isn't part of your agent's decision-making. With one-off, per-channel integrations, a developer decided at build time where and how the agent talks. The agent itself can't reason "this deserves a quick Telegram ping now and an email summary afterwards" — each channel is a separate bot with separate code and a separate identity. Communication stays hardcoded plumbing instead of becoming a capability the model can actually decide with.

3. You maintain N identities for every one person. The same human DMs your agent on Instagram today and emails it tomorrow. Now your database needs its own concept of "this is one person, one relationship, one running conversation" — who said what on which channel, and what should happen next in the flow. Every team rebuilds that continuity layer from scratch, per app, and it never stops needing care.

4. A single-channel agent is a competitive disadvantage. If a competing agent is reachable on five channels and yours on one, users go where they get answered. The open-source numbers show it: the agents people actually rely on are exactly the ones deployed across dozens of human channels — and that reach is exactly where their engineering time goes.

Caspian's answer

Channels are transports, not identities. The agent is one program (cx.app.rules is inspectable data); every channel binds through the same adapter interface, and your handler code works against a normalized Thread / Message model. Messages arrive as kernel events regardless of transport, overlap policies (queue / debounce / drop / parallel) serialize concurrent chats, and thread.post() / thread.reply() always answer in the right place.

flowchart LR
    S[Slack] --> A
    D[Discord] --> A
    T[Telegram] --> A
    E[Email] --> A
    W[WhatsApp · Messenger] --> A
    X[X] --> A
    A["channel adapters
verify · normalize · thread"] --> I["one agent program"]
    I --> H["your on_message rules"]
    H -->|"thread.post()"| I

Hosted or self-host, same code. via="hosted" (default) uses the Caspian gateway at https://api.trycaspianai.com — set CASPIAN_API_KEY and optionally CASPIAN_BASE_URL. via="self-host" runs adapters in your process with your platform tokens. Switch modes without rewriting handlers.

Features

** Declarative rules, one program**

@cx.on_message({"channel": "telegram", "command": "help"}) — filters for channel, chat kind, command, overlap, and instant ack. Your bot is data: cx.app.rules is inspectable and testable offline.

** Webhook verification, always**

Slack signing secret, Meta X-Hub-Signature-256, Telegram secret header, X CRC, and signed email webhooks. Mismatches rejected.

☁️ Hosted or self-host

Gateway polling with cx.run(), or bring your own tokens and webhooks/sockets with via="self-host". Same handler rules either way.

** Offline fakes for every channel**

Adapters consume each platform's real payload shapes — 650+ tests across Python + TypeScript, zero network in CI.

⌨️ Typing, streaming, rich sends

thread.typing(), thread.stream() (post once, edit as it writes), thread.send_media(), thread.send_blocks(), reactions, pins, forwards, and cold DMs.

** Model tools from the same surface**

cx.tools(thread) exposes the Command catalog (post, react, send-photo, …) with schemas derived from the kernel — same API your handlers use.

** Per-channel packs (TypeScript)**

Import caspian-sdk/telegram, caspian-sdk/discord, caspian-sdk/slack, and the rest for parse/plan/execute without pulling the whole facade.

** Socket inbound (Discord, Slack)**

No public URL required — cx.listen("discord") or cx.listen("slack") over a held-open websocket (optional extras).

Channels

Self-host adapters ship in the SDK for the channels below. Hosted mode covers any channel the gateway supports (including Bluesky, Instagram, and channels with no local adapter).

• The fine print — read before you promise features

• X is not free: DM send/receive needs a paid X API subscription on your X developer app (the free tier is write-only and capped).
• GSM modem SMS: your own modem + SIM; carrier compliance (A2P rules) is on you.

Where to use it

If your agent needs to talk to humans, this is the layer under it:

• Customer support agents — answer on email, Slack, Instagram DM, or wherever the customer opened the thread; hand off to a human without dropping context.
• Sales & lead follow-up — first touch on the channel the lead used, follow-ups where they actually respond.
• Personal / executive assistants — one assistant identity across your email, Telegram, and Slack instead of three disconnected bots.
• Community & product bots — the same agent in your Discord, your Slack community, and members' DMs.
• OpenClaw agents — clawhub install @trycaspian/caspian (the skill) teaches your agent to wire itself up; openclaw-caspian is the native channel plugin.
• OpenCode agents — caspian-opencode-plugin bridges Caspian email / Telegram / Discord into OpenCode sessions. Details: packages/opencode.

Start from a runnable example — one folder per channel, shared handlers in app.py / app.ts.

Recipes

Same agent, three channels:

cx.channels.add("email", display_name="Acme Support")
cx.channels.add("telegram", bot_token=BOT_TOKEN)
cx.channels.add("slack", bot_token=SLACK_TOKEN, signing_secret=SLACK_SECRET)
# the @cx.on_message rules you already wrote now answer on all three
cx.run()

Filter by command and chat kind:

@cx.on_message({"channel": "telegram", "command": ["start", "help"]})
def help_menu(thread, msg, ctx):
    thread.post("Commands: /help /status /ping")

@cx.on_message({"channel": "telegram", "kind": "dm"})
def dm_only(thread, msg, ctx):
    thread.po