orbli.py

A small, friendly asyncio client for building Orbli bots (we call them Satellites). It handles the gateway connection, heartbeats, resuming after drops, rate limits and interaction replies, so your bot can be a handful of lines.

Needs Python 3.10 or newer.

pip install orbli

A /ping bot

import os
from orbli import Client, Intents

client = Client(os.environ["ORBLI_TOKEN"], intents=Intents.SERVERS)

@client.on("ready")
async def ready(data):
    await client.rest.set_commands([{"name": "ping", "description": "Check that the bot is awake"}])
    print(f"Logged in as {data['user']['username']}")

@client.on("INTERACTION_CREATE")
async def interaction(i):
    if i.command_name == "ping":
        await i.reply("Pong! 🛰️")

client.run()

Grab your token from Mission Control and put it in ORBLI_TOKEN. Never commit it.

If the connection drops, the client reconnects with backoff and resumes, so you don't miss events. If the token is wrong or you asked for an intent that isn't turned on, it stops with an OrbliGatewayError that says what to fix. Already inside an event loop? Use await client.start() instead of client.run().

Events

Every gateway event is emitted under its own name, with the raw payload as a dict. Handlers can be async or plain functions.

@client.on("MESSAGE_CREATE")
async def message(msg):
    print(msg["author"]["username"], msg["content"])

@client.on("REACTION_ADD")
async def reacted(event):
    print(event["emoji"], "on", event["messageId"])

@client.on("dispatch")
def everything(name, data):
    print("got", name)

There's also disconnect (with the close code and reason), resumed and error (for exceptions your handlers raise).

Pick intents by adding them up: Intents.SERVERS | Intents.SERVER_MESSAGES | Intents.MESSAGE_CONTENT. MEMBERS and MESSAGE_CONTENT are privileged, so turn them on for your app first. INTERACTION_CREATE always arrives.

Slash commands with options

import random

COMMANDS = [
    {
        "name": "roll",
        "description": "Roll a die",
        "options": [
            {"type": "INTEGER", "name": "sides", "description": "How many sides", "required": True, "minValue": 2, "maxValue": 100},
            {"type": "STRING", "name": "style", "description": "Dice style", "autocomplete": True},
            {"type": "USER", "name": "for", "description": "Roll for someone else"},
        ],
    }
]

@client.on("INTERACTION_CREATE")
async def interaction(i):
    if i.is_autocomplete:
        typed = str(i.focused["value"]).lower()
        return await i.autocomplete([s for s in ["Classic", "Cosmic", "Glow"] if s.lower().startswith(typed)])

    if i.command_name == "roll":
        sides = i.option("sides")
        who = i.resolved["users"].get(i.option("for"), i.user)
        await i.reply(f"{who['displayName']} rolled **{random.randint(1, sides)}**")

Register them with await client.rest.set_commands(COMMANDS). Pass server_id=... to register for one server only; those show up right away, which is handy while testing.

Right-click commands

Commands can also live in the menu you get when you right-click a person (USER) or a message (MESSAGE). They don't take options, the description is optional, and the name can have capitals and spaces (up to 32 characters).

await client.rest.set_commands([
    {"name": "roll", "description": "Roll a die"},
    {"type": "USER", "name": "Wave"},
    {"type": "MESSAGE", "name": "Report message"},
])

@client.on("INTERACTION_CREATE")
async def interaction(i):
    if i.command_type == "USER" and i.command_name == "Wave":
        return await i.reply(f"👋 {i.target_user['displayName']}!")
    if i.command_type == "MESSAGE" and i.command_name == "Report message":
        author = i.target_message["author"]["username"]
        await i.reply(f"Thanks, we'll take a look at the message from {author}.", ephemeral=True)

i.command_type is "CHAT_INPUT", "USER" or "MESSAGE", i.target_id is the id that was clicked, and i.target_user / i.target_message hand you the full object.

Translations

Give commands and options names and descriptions in other languages. Keys are language codes like es or pt-BR, and people see the version that matches their Orbli language.

COMMANDS = [
    {
        "name": "roll",
        "description": "Roll a die",
        "nameLocalizations": {"es": "tirar", "pt-BR": "rolar"},
        "descriptionLocalizations": {"es": "Tira un dado", "pt-BR": "Role um dado"},
        "options": [
            {"type": "INTEGER", "name": "sides", "description": "How many sides", "nameLocalizations": {"es": "caras"}},
        ],
    }
]

@client.on("INTERACTION_CREATE")
async def interaction(i):
    await i.reply("¡Hola!" if (i.locale or "").startswith("es") else "Hi!")

Your code still sees the English names (i.command_name == "roll", i.option("sides")). i.locale is the person's language, like "es-ES", or None if we don't know it.

User-installed apps

People can add your app to their own account, not just to a server. Your global commands then work for them everywhere: in servers your bot isn't in, in DMs and in group DMs.

@client.on("INTERACTION_CREATE")
async def interaction(i):
    if i.install_type == "user":
        print(f"{i.user['username']} used {i.command_name} from their own account")

i.install_type is "server" or "user". When it's "user" and your bot isn't in that channel, Orbli shows your replies only to the person who used the command, even if you didn't ask for ephemeral. Server-only commands (registered with a server id) stay inside that server.

Buttons, menus and pop-up forms

@client.on("INTERACTION_CREATE")
async def interaction(i):
    if i.command_name == "vote":
        return await i.reply(
            "Pick a planet",
            components=[
                {
                    "type": "ROW",
                    "components": [
                        {"type": "BUTTON", "style": "PRIMARY", "label": "Mars", "customId": "vote:mars"},
                        {"type": "BUTTON", "style": "SECONDARY", "label": "Venus", "customId": "vote:venus"},
                        {"type": "BUTTON", "style": "LINK", "label": "Learn more", "url": "https://orbli.app"},
                    ],
                }
            ],
        )

    if i.custom_id and i.custom_id.startswith("vote:"):
        return await i.update(f"You picked {i.custom_id[5:]}!")

    if i.command_name == "feedback":
        return await i.show_modal({
            "customId": "feedback",
            "title": "Send feedback",
            "fields": [{"customId": "text", "label": "What's on your mind?", "style": "PARAGRAPH"}],
        })

    if i.is_modal_submit and i.custom_id == "feedback":
        await i.reply(f"Thanks! You said: {i.field('text')}", ephemeral=True)

Select menus work the same way, and i.values holds what was picked.

Taking a while? await i.defer() (or i.defer_update() for buttons), do the work, then await i.edit_original(...). Send more messages with i.follow_up(...). Interactions have to be answered within 3 seconds.

Embeds

from orbli import iso_timestamp

await client.rest.send_message(channel_id, embeds=[
    {
        "title": "Launch report",
        "description": "All systems go.",
        "color": 0x7C5CFF,
        "fields": [
            {"name": "Crew", "value": "Lotl, Fen", "inline": True},
            {"name": "Fuel", "value": "98%", "inline": True},
        ],
        "footer": {"text": "Mission Control"},
        "timestamp": iso_timestamp(),
    }
])

Up to 5 embeds per message, and up to 6000 characters of text across them. iso_timestamp() gives you the ...Z format Orbli expects (optionally from a datetime). reply, update, edit_original and follow_up take embeds= too.

Presence

client = Client(token, intents=Intents.SERVERS, presence={"status": "online", "activity": {"type": "PLAYING", "name": "with orbits"}})

await client.set_presence("idle", {"type": "LISTENING", "name": "the stars"})

Status is online, idle or dnd. Activities are PLAYING, LISTENING or STREAMING (streaming can have an https url). Orbli allows 5 presence changes a minute; the client holds back extra ones and sends the latest when it can.

REST on its own

from orbli import Rest

async with Rest(os.environ["ORBLI_TOKEN"]) as rest:
    servers = await rest.get_servers()
    msg = await rest.send_message(channel_id, "Hello!")
    await rest.add_reaction(channel_id, msg["id"], "🚀")
    await rest.edit_message(channel_id, msg["id"], "Hello again!")

Also there: me, get_server, get_channels, get_roles, get_members, get_messages, delete_message, remove_reaction, get_commands, set_commands, plus get, post, patch, put and delete for anything else. When Orbli says you're going too fast (429), it waits for retry-after and tries again. Other failures raise OrbliApiError with .status and the server's .message.

Moderation and roles

await rest.add_role(server_id, user_id, role_id)
await rest.remove_role(server_id, user_id, role_id)
await rest.set_nickname(server_id, user_id, "Captain Lotl")  # None clears it
await rest.timeout(server_id, user_id, 60, reason="Cooling off")  # None lifts it
await rest.kick(server_id, user_id, reason="Spam")
await rest.ban(server_id, user_id, reason="Raid", delete_hours=24)
await rest.unban(server_id, user_id)

Each one needs the matching permission on your bot's roles: Manage Roles, Manage Nicknames, Timeout Members, Kick Members or Ban Members. Your bot can only act on people (and hand out roles) below its own highest role, and never on the server owner. Managed roles and the default role can't be added or removed. Timeouts last 1, 5, 10, 60, 1440 (a day) or 10080 (a week) minutes, and timeout returns timeoutUntil. delete_hours on a ban can be 0, 1, 24 or 168 and removes that much of their recent messages.

Files, pins, threads and DMs

Upload a file first, then send its id with your message (up to 4 per message):

from pathlib import Path

file = await rest.upload_file(Path("chart.png").read_bytes(), "chart.png", content_type="image/png")
await rest.send_message(channel_id, "This week's numbers", attachment_ids=[file["id"]])

upload_file returns {"id", "url", "filename", "contentType", "size"}.

await rest.pin_message(channel_id, message_id)
pins = await rest.get_pins(channel_id)

thread = await rest.create_thread(channel_id, "Launch day", message_id=message_id)
await rest.join_thread(thread["id"])
await rest.add_thread_member(thread["id"], user_id)
await rest.edit_thread(thread["id"], archived=True, locked=True)
old = await rest.get_threads(channel_id, archived=True)

dm = await rest.open_dm(user_id)
await rest.send_message(dm["channelId"], "Your report was received 👍")

Also there: unpin_message, leave_thread and remove_thread_member.

Webhook mode (no gateway)

Prefer not to keep a connection open, or running serverless? Set an Interactions Endpoint URL in Mission Control and Orbli will POST interactions to it instead. You answer in the HTTP response.

client.webhook(secret) gives you a handler that checks the signature, answers the verification ping and runs the same INTERACTION_CREATE handlers. Call it with the request headers and the raw body, and it returns (status, body).

With FastAPI (or any ASGI app):

import os
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from orbli import Client

client = Client(os.environ["ORBLI_TOKEN"])
handle = client.webhook(os.environ["ORBLI_INTERACTIONS_SECRET"])
app = FastAPI()

@client.on("INTERACTION_CREATE")
async def interaction(i):
    if i.command_name == "ping":
        await i.reply("Pong! 🛰️")

@app.post("/orbli")
async def orbli(request: Request):
    status, body = await handle(request.headers, await request.body())
    return JSONResponse(body, status_code=status)

With Flask:

import asyncio
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.post("/orbli")
def orbli():
    status, body = asyncio.run(handle(request.headers, request.get_data()))
    return jsonify(body), status

Flask runs each request in a fresh event loop that closes when the response is sent, so stick to direct replies there. For defer() followed by edit_original() or follow_up(), use an ASGI server like the FastAPI example; those calls use your bot token over REST.

Your secret starts with ois_. Want to check signatures yourself? verify_webhook(secret, timestamp, raw_body, signature) returns True or False.

Scheduled triggers

Set up a schedule in Mission Control ("every morning at 9", "every hour") and Orbli will nudge your bot when it's time. Over the gateway it arrives as a SCHEDULE_RUN event:

@client.on("SCHEDULE_RUN")
async def scheduled(run):
    print(f"{run['schedule']['name']} for {run['scheduledFor']}")
    await client.rest.send_message(channel_id, "Good morning, crew! ☀️")

In webhook mode the run is a signed POST to your interactions endpoint, and client.webhook(secret) sends it to the same SCHEDULE_RUN handlers. Using WebhookHandler directly? Pass on_schedule=:

handle = WebhookHandler(secret, on_interaction, on_schedule=post_daily_digest)

The run looks like {"type": "SCHEDULE", "id": ..., "schedule": {"id": ..., "name": ...}, "scheduledFor": ...}. The handler answers {"type": 1} as soon as your code finishes, or after timeout (2.5 seconds) if it's still busy, and your code keeps running in the background. If it raises before then, the run shows up as failed.

Storage and timers

Every app gets a small key-value store, so a serverless bot can remember things between requests. Keys are up to 200 letters, numbers, _, ., : or -. Values are anything JSON.

await rest.kv_set("prefix:server1", "!")
await rest.kv_set("cooldown:user1", True, ttl=60)  # gone after 60 seconds
entry = await rest.kv_get("prefix:server1")  # {"key", "value", "expiresAt", "updatedAt"} or None
await rest.kv_increment("xp:user1", 10)
await rest.kv_delete("prefix:server1")

page = await rest.kv_list(prefix="xp:", limit=50)  # page["items"], page["next"]

Timers are one-off reminders: Orbli calls your bot back once, seconds from now or at a time, with whatever payload you gave it.

from datetime import datetime, timezone

await rest.create_timer(seconds=3600, name="remind", payload={"userId": user_id, "channelId": channel_id})
await rest.create_timer(at=datetime(2026, 12, 31, 23, 59, tzinfo=timezone.utc), name="new-year")

@client.on("TIMER_FIRE")
async def timer(fire):
    if fire["timer"]["name"] == "remind":
        await client.rest.send_message(fire["payload"]["channelId"], "Time's up! ⏰")

It arrives like a scheduled run: over the gateway as TIMER_FIRE, or in webhook mode as {"type": "TIMER", "id": ..., "timer": {"id": ..., "name": ...}, "payload": ..., "scheduledFor": ...}, which client.webhook(secret) sends to the same handlers (with WebhookHandler, pass on_timer=). Use get_timers, get_timer and cancel_timer to look at or call off the ones still waiting.

Log in with Orbli

Let people sign in to your website with their Orbli account. Add your redirect URL in Mission Control first.

import secrets
from orbli import OAuth

oauth = OAuth(
    os.environ["ORBLI_APP_ID"],
    client_secret=os.environ["ORBLI_CLIENT_SECRET"],
    redirect_uri="https://example.com/callback",
)

# 1. Send them to Orbli
verifier, challenge = OAuth.create_pkce()
state = secrets.token_urlsafe(16)
# keep verifier and state in their session, then redirect to:
url = oauth.authorize_url(scope=["identify", "email", "servers"], state=state, code_challenge=challenge)

# 2. They come back to /callback?code=...&state=...
tokens = await oauth.exchange_code(code, code_verifier=verifier)
me = await oauth.get_user(tokens["access_token"])
servers = await oauth.get_servers(tokens["access_token"])

# 3. Later
fresh = await oauth.refresh(tokens["refresh_token"])
await oauth.revoke(fresh["access_token"])

Scopes are identify (who they are), email (adds email and emailVerified to get_user) and servers (the servers they're in, with owner and permissions). Always check state matches before exchanging the code.

Running somewhere that can't keep a secret, like a desktop or mobile app? Leave out client_secret and use PKCE: that's what the verifier is for. Token responses are {"access_token", "token_type", "expires_in", "refresh_token", "scope"}, and errors raise OrbliApiError whose .code is the OAuth error (like invalid_grant). OAuth works as an async with block too. For local testing, pass base_url= (where the authorize page lives) and api_url= (defaults to base_url + "/api").

Constants

Intents, GatewayOps, CloseCodes and CallbackTypes are exported. To point at a local Orbli server, pass api_url= and gateway_url= to Client (or api_url= to Rest).