polis-bot 0.2.2

Async Python client for the Polis Bot API

polis-bot

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

Requires Python 3.10+. Built on httpx and Pydantic v2.

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. The bot long-polls for updates and dispatches them to the matching handler.

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

Sending files

# From a path
await bot.send_file(user_id, "image.png", content_type="image/png")

# From a Path object
from pathlib import Path
await bot.send_file(user_id, Path("report.pdf"), content_type="application/pdf")

# 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

@bot.on_file
async def handle_file(ctx: MessageContext):
    for att in ctx.message.attachments:
        data = await bot.download_file(att.download_url)
        # or save directly:
        await bot.download_file_to(att.download_url, f"downloads/{att.file_name}")
        await ctx.reply(f"Got **{att.file_name}** ({att.size_bytes} 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

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 Polling

For full control over the update loop instead of using decorators:

async with PolisBot("token") as bot:
    async for update in bot.poll(limit=20, timeout=30):
        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!")

Updates are automatically acknowledged after each batch. Disable with auto_ack=False:

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

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 text messages
@bot.on_file	Handle messages with 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)	Start polling and dispatch (blocking)
start(*, register_commands=True)	Async version of run()
get_me()	Get bot info
send_message(user_id, content, *, reply_to_id=None)	Send a text message
send_file(user_id, file, *, filename=None, content_type=...)	Upload a file
download_file(download_url)	Download file bytes
download_file_to(download_url, dest)	Download file to disk
get_updates(*, limit, offset, timeout)	Fetch pending updates
acknowledge_updates(update_ids)	Mark updates as delivered
set_commands(commands)	Register bot commands
poll(*, limit=20, timeout=30, auto_ack=True)	Async generator for long-polling
close()	Close the HTTP client

Context Objects

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

License

MIT