polis-bot

Async Python client for the Polis Bot API — build bots for the Polis messaging platform.

Requires Python 3.10+. The messaging loop (receiving updates, sending messages, typing indicators, streaming) runs over a persistent bidirectional SignalR hub connection, via pysignalr. Identity, command registration, and file transfer remain plain REST calls via httpx. Models are Pydantic v2.

pysignalr's own reconnect covers a dropped websocket with unlimited retries and a capped backoff, but a failure at the SignalR negotiate step (needed on first connect, and occasionally again mid-life) only gets a bounded number of attempts before that connection attempt gives up for good — this SDK's own indefinite, 60s-capped retry (visible as hub_connect failed (...), retrying in Ns in the logs) is what actually backs "resilient to server downtime" end-to-end. See the comment above _HUB_NEGOTIATE_RETRY_COUNT in client.py for the full mechanics.

Installation

pip install polis-bot

Quick Start

from polis_bot import PolisBot, CommandContext, MessageContext

bot = PolisBot("your-bot-token")

@bot.command("/start", description="Start the bot")
async def start(ctx: CommandContext):
    await ctx.reply(f"Hello {ctx.user.display_name}! 🤖")

@bot.command("/help", description="Show help")
async def help_cmd(ctx: CommandContext):
    await ctx.reply("**Commands:**\n- /start — Greet\n- /help — This message")

@bot.on_message
async def echo(ctx: MessageContext):
    await ctx.reply(f"You said: {ctx.message.content}")

bot.run()

Commands are auto-registered with the server on startup. bot.run() opens the hub connection and dispatches server-pushed updates to the matching handler — there's no polling; updates arrive as soon as the server sends them, and any updates that arrived while disconnected are flushed automatically the moment the connection (re)opens.

Command Arguments

Handler type hints drive automatic argument parsing — like FastAPI for bots:

@bot.command("/add", description="Add two numbers")
async def add(ctx: CommandContext, a: int, b: int):
    await ctx.reply(f"{a + b}")
    # /add 3 5 → "8"

The last str parameter absorbs remaining tokens:

@bot.command("/echo", description="Echo text")
async def echo_cmd(ctx: CommandContext, text: str):
    await ctx.reply(text)
    # /echo hello world → "hello world"

Supported types: str, int, float, bool, UUID, and Pydantic BaseModel subclasses. Invalid input returns a usage hint automatically.

File Handling

Files are attachments on messages — any message (text, command, or standalone) can carry one or more files. Each attachment is an Attachment object with metadata properties and download() / download_to() methods — no need to manage URLs.

Sending files

# File-only message
await bot.send_file(user_id, "image.png", content_type="image/png")

# Text + file in one message
await bot.send_file(user_id, "report.pdf",
                    content_type="application/pdf",
                    content="Here's the report you requested")

# Via context
@bot.command("/cat", description="Send a cat picture")
async def cat(ctx: CommandContext):
    await ctx.reply_file("cat.jpg", content_type="image/jpeg")

Receiving files

File attachments are available on any message via ctx.message.attachments. Each Attachment exposes metadata and can download itself:

@bot.on_message
async def handle(ctx: MessageContext):
    if ctx.message.attachments:
        for att in ctx.message.attachments:
            print(f"{att.file_name} ({att.content_type}, {att.size_bytes} bytes)")
            # Download to memory
            data = await att.download()
            # or save to disk
            await att.download_to(f"downloads/{att.file_name}")
        await ctx.reply(f"Got {len(ctx.message.attachments)} file(s)")
    elif ctx.message.content:
        await ctx.reply(f"You said: {ctx.message.content}")

Commands can also have file attachments:

@bot.command("/analyze", description="Analyze an uploaded file")
async def analyze(ctx: CommandContext):
    if not ctx.message.attachments:
        await ctx.reply("Please attach a file to analyze")
        return
    att = ctx.message.attachments[0]
    data = await att.download()
    await ctx.reply(f"Analyzed **{att.file_name}**: {len(data)} bytes")

Lifecycle Events

from polis_bot import EventContext

@bot.on_event("BotStarted")
async def on_started(ctx: EventContext):
    await bot.send_message(ctx.user.id, "Welcome! Send /help to get started.")

@bot.on_event("BotBlocked")
async def on_blocked(ctx: EventContext):
    print(f"{ctx.user.display_name} blocked the bot")

Event types: BotStarted, BotBlocked, BotUnblocked, BotAdded, BotRemoved.

Startup Hook

@bot.on_startup
async def setup():
    print(f"Running as @{bot.me.botname} ({bot.me.display_name})")

bot.me is populated before startup hooks run and returns cached BotInfo.

Error Handling

The SDK raises typed exceptions for API errors:

from polis_bot import PolisApiError, PolisAuthError, PolisRateLimitError

try:
    await bot.send_message(user_id, "hello")
except PolisAuthError:
    print("Invalid bot token")
except PolisRateLimitError as e:
    print(f"Rate limited — retry after {e.retry_after}s")
except PolisApiError as e:
    print(f"HTTP {e.status_code}: {e.message}")

Exception	HTTP Status
PolisAuthError	401
PolisForbiddenError	403
PolisNotFoundError	404
PolisRateLimitError	429
PolisApiError	Any non-2xx

For hub calls (send_message, send_typing, acknowledge_updates, streaming) there's no HTTP status to report — the server delivers a HubException message string instead. The SDK does its best to map that onto the same exception family: messages that look auth-related become PolisAuthError, everything else becomes a PolisApiError with status_code=0 (a sentinel meaning "no HTTP status — hub-originated error").

Async Context Manager

async with PolisBot("token") as bot:
    me = await bot.get_me()
    await bot.send_message(user_id, "Hello!")
# client is automatically closed

Low-Level Update Loop

For full control over the update loop instead of using decorators. This connects the hub on first use (retrying transient connection failures) and yields updates as the server pushes them:

async with PolisBot("token") as bot:
    async for update in bot.poll():
        print(update.type, update.user.display_name)

        if update.message and update.message.is_command:
            await bot.send_message(update.user.id, "Got your command!")

Each update is automatically acknowledged after it's yielded. Disable with auto_ack=False:

async for update in bot.poll(auto_ack=False):
    # manually acknowledge
    await bot.acknowledge_updates([update.update_id])

Streaming Replies

Stream a reply chunk-by-chunk — useful for relaying an LLM response as it's generated. Chunks sent via send_chunk() are relayed live but never persisted on their own; finish() is what actually saves the message:

@bot.command("/ask", description="Ask the assistant")
async def ask(ctx: CommandContext):
    stream = ctx.bot.start_stream(ctx.user.id, reply_to_id=ctx.message.id)
    async for delta in llm_response(ctx.args):
        await stream.send_chunk(delta)
    await stream.finish()

Or as an async context manager, which calls finish() automatically (with the accumulated text) on clean exit:

async with bot.start_stream(user_id) as stream:
    async for delta in llm_response(prompt):
        await stream.send_chunk(delta)

finish(content=...) can also be given the complete final text explicitly, overriding whatever was accumulated via send_chunk().

Hub Activity / Liveness Hook

Since traffic now flows over a persistent connection instead of REST long-polling, on_hub_activity gives visibility into that connection for things like driving an external watchdog. It fires on every inbound hub message — update pushes, invoke completions, and the transport's own ping/keepalive — plus "connect" / "disconnect" on connection-state changes:

@bot.on_hub_activity
def touch_watchdog(kind: str) -> None:
    watchdog.reset()

The callback may be sync or async; exceptions it raises are logged and swallowed.

API Reference

PolisBot(token, base_url="https://polis.krlv.org", *, timeout=30.0)

Properties

Property	Description
me	Cached BotInfo (available after start() / run())

Decorators

Decorator	Description
@bot.command(name, *, description=None)	Register a /command handler with auto-parsed args
@bot.on_message	Handle non-command messages (text and/or file attachments)
@bot.on_event(event_type)	Handle lifecycle events (BotStarted, etc.)
@bot.on_startup	Run a coroutine once on startup

Methods

Method	Description
run(*, register_commands=True)	Connect the hub and dispatch (blocking)
start(*, register_commands=True)	Async version of run()
get_me()	Get bot info (REST)
send_message(user_id, content, *, reply_to_id=None)	Send a text message (hub)
send_typing(user_id, duration=30)	Set/clear the typing indicator (hub)
start_stream(user_id, *, reply_to_id=None)	Begin a streamed reply — returns a MessageStream
send_file(user_id, file, *, filename=None, content_type=..., content=None)	Upload a file (optionally with text) (REST)
acknowledge_updates(update_ids)	Mark updates as delivered (hub)
set_commands(commands)	Register bot commands (REST)
poll(*, auto_ack=True)	Async generator draining hub-pushed updates
on_hub_activity(callback)	Register a liveness callback (see above)
close()	Close the hub connection and the HTTP client

MessageStream

Returned by start_stream(); also usable as an async context manager.

Property / Method	Description
stream_id	The id shared between StreamChunk calls and the final SendMessage
send_chunk(delta_text)	Relay an incremental chunk (not persisted on its own)
finish(content=None)	Persist the final message; defaults to the accumulated text

Context Objects

Class	Available On	Key Properties
CommandContext	@bot.command	user, message, args, bot, reply(), reply_file()
MessageContext	@bot.on_message	Same as CommandContext
EventContext	@bot.on_event	user, bot

Attachment

Each file attachment on ctx.message.attachments is an Attachment instance:

Property / Method	Description
file_name	Original filename
content_type	MIME type
size_bytes	File size in bytes
id	Server-assigned file ID
download()	Download and return raw bytes
download_to(dest)	Download and save to disk, returns Path

License

MIT