Metadata-Version: 2.5
Name: orbli
Version: 0.1.0
Summary: A small client for building Orbli bots (Satellites)
Project-URL: Homepage, https://orbli.dev
License: MIT
Keywords: bot,gateway,orbli,satellite,websocket
Classifier: Framework :: AsyncIO
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: websockets>=12
Description-Content-Type: text/markdown

# 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

```python
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.

```python
@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

```python
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).

```python
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.

```python
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.

```python
@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

```python
@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

```python
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

```python
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

```python
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

```python
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):

```python
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"}`.

```python
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):

```python
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:

```python
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:

```python
@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=`:

```python
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.

```python
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.

```python
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.

```python
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`).
