EdgeBot SDK

Минималистичный SDK для Telegram-ботов на Cloudflare Workers (Pyodide).

Установка

pip install edgebot

Требования: Python 3.10+, Cloudflare Workers с флагом python_workers.

Структура пакета

Пакет импортируется как edgebot и разделён на подмодули, каждый из которых можно использовать независимо:

• edgebot — ядро: Bot, Context, InlineKeyboard, исключения.
• edgebot.ui — UI-движок: атомарные экраны (UIComponent) и реестры клавиатур/промптов (KeyboardRegistry, PromptRegistry, Prompt).
• edgebot.storage — персистентность на Cloudflare KV: KVStore, UserRegistry.
• edgebot.utils — общие утилиты: HTTP-клиент с ретраями, структурированный логгер, таймстемпы.
• edgebot.ffi — граница Python ↔ JS: нормализация null/undefined, конвертация Python-структур в plain JS-объекты.

Наиболее используемые имена реэкспортированы из корня пакета: Bot, Context, InlineKeyboard, KVStore, UserRegistry, UIComponent, KeyboardRegistry, PromptRegistry, Prompt, sleep, to_py, to_js_object, EdgeBotError.

Возможности

Bot — маршрутизация обновлений Telegram

Декораторная регистрация хендлеров по типу события:

• Команды и текст: @bot.on_command("/start"), @bot.on_message.
• Inline-кнопки: @bot.on_callback (фолбэк после UI-компонентов).
• Медиа: @bot.on_photo, on_video, on_audio, on_voice, on_document, on_sticker, on_animation.
• Чеклисты: @bot.on_checklist, on_checklist_tasks_done.

Привязка к Workers runtime и подключение реестров через чейнинг:

• bot.env(env).ctx(ctx) — биндинги Cloudflare и ExecutionContext;
• bot.with_users(UserRegistry(env.USERS)) → ctx.users;
• bot.with_keyboards(KeyboardRegistry()) → ctx.keyboards;
• bot.with_prompts(PromptRegistry()) → ctx.prompts;
• bot.register_ui(ScreenA(), ScreenB(), ...) — UI-экраны, которые перехватывают callback по префиксу до классических on_callback.

Context — данные апдейта и ответ

В каждом хендлере доступны разобранные поля: chat_id, message_id, text, command, callback_data, from_user, медиа-объекты (photo, video, voice и т.д.), а также ссылки на ctx.env, ctx.bot, ctx.users, ctx.keyboards, ctx.prompts.

Методы отправки: send, reply, edit_message, answer_callback, send_photo, send_video, send_audio, send_voice, send_document, send_sticker, send_animation.

InlineKeyboard — builder кнопок

Fluent API: button(text, callback_data=...|url=...), row(), extend(other), build(prefix=None).

callback_data может быть относительным ("toggle") или абсолютным ("/menu:home" — слэш срезается, префикс игнорируется). При вызове build(prefix="prof") все относительные пути префиксуются: "toggle" → "prof:toggle". Это позволяет UI-движку поздно связывать клавиатуру с префиксом конкретного компонента; build() идемпотентен и не мутирует исходные кнопки.

edgebot.ui — UI-движок (экраны и реестры)

Архитектура — MVC, адаптированная под stateless-Workers:

• UIComponent — атомарный экран. Наследник задаёт уникальный __prefix__, опционально prompt/markup (ключи в реестрах) или inline-text, реализует open(ctx, **kwargs) и методы, помеченные @UIComponent.callback("pattern"). Паттерн поддерживает плейсхолдеры {name} (превращаются в kwargs метода через regex) и спец-значение "*" — fallback в рамках префикса. Методы send() / update() сами достают промпт и клавиатуру из реестров и отправляют сообщение; update() молча гасит ошибку Telegram «message is not modified».
• KeyboardRegistry — реестр фабрик клавиатур. Фабрика регистрируется через @registry.register("key") и обязана возвращать InlineKeyboard (не сырой dict — финализацию делает движок).
• PromptRegistry + Prompt — реестр текстовых вьюх. Prompt — композитный билдер (append/prepend + render(**kwargs) через str.format_map).
• Авто-фильтр kwargs. Реестры пропускают через фабрику только те ключи, что объявлены в её сигнатуре, чтобы один общий dict экрана можно было передавать и в клавиатуру, и в промпт без ловушек.
• Уникальность префиксов. Регистрация двух компонентов с одинаковым __prefix__ бросает DuplicatePrefixError.

edgebot.storage — Cloudflare KV

• KVStore — тонкая обёртка над KV-биндингом. Методы: get, put, delete, exists, list, iter_keys, get_with_metadata. Сам решает по типу значения, что кодировать как JSON, что оставить строкой, а что отправить как Uint8Array.
• UserRegistry — реестр пользователей поверх KV. Методы: get, create, upsert, update, delete, iter_ids, items. Автоматически проставляет created_at / updated_at.

edgebot.utils

• http.fetch — обёртка над js.fetch с ретраями на 408/429/500/502/503/504, учётом Retry-After.
• http.sleep(ms) — async-задержка через JS setTimeout.
• log — структурированный JSON-логгер: debug, info, warn, error. Уровни мапятся на js.console.*, поэтому в Cloudflare Workers Logs корректно выставляется поле level.
• now() — ISO-8601 UTC-таймстемп.

edgebot.ffi

• to_js_object(value) — Python dict/list → plain JS Object/Array для API, которым нужен Object, а не Map (KV options, D1, R2, Queues, DO stub RPC).
• to_py(value) — JsProxy → Python, с нормализацией null / undefined в None. Устойчив к разной модульной структуре Pyodide в разных сборках.

Исключения

Все ошибки SDK наследуются от EdgeBotError. Дочерние ветки:

• StorageError → KVError, UsersError → UserAlreadyExists;
• UIError → KeyboardNotFound, PromptNotFound, DuplicatePrefixError.

Быстрый старт

from workers import Response, WorkerEntrypoint
from edgebot import Bot, Context, InlineKeyboard, UserRegistry


def build_bot(env) -> Bot:
    users = UserRegistry(env.USERS)
    bot = Bot(env.BOT_TOKEN, parse_mode="Markdown").with_users(users)

    @bot.on_command("/start")
    async def start(ctx: Context):
        kb = InlineKeyboard()
        kb.button("Нажми меня", callback_data="hello")
        await ctx.send("Привет!", reply_markup=kb.build())

    @bot.on_callback
    async def cb(ctx: Context):
        await ctx.answer_callback("Ок")

    @bot.on_message
    async def echo(ctx: Context):
        await ctx.reply(ctx.text)

    return bot


class Default(WorkerEntrypoint):
    def __init__(self, ctx, env):
        super().__init__(ctx, env)
        self.bot = build_bot(env)
        self.bot.env(env).ctx(ctx)
        self.env = env

    async def fetch(self, request):
        secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token")
        if not secret or secret != self.env.WEBHOOK_TOKEN:
            return Response("{}", status=401)
        update = await request.json()
        await self.bot.process_update(update)
        return Response('{"ok": true}')

Пример UI-экрана

from edgebot import (
    Bot, Context, InlineKeyboard,
    UIComponent, KeyboardRegistry, PromptRegistry, Prompt,
)

keyboards = KeyboardRegistry()
prompts = PromptRegistry()


@prompts.register("profile_main")
def profile_prompt(username: str, is_active: bool) -> Prompt:
    p = Prompt("*Профиль*: {username}")
    p.append("Статус: активен" if is_active else "Статус: выключен")
    return p


@keyboards.register("profile_main")
def profile_kb(is_active: bool) -> InlineKeyboard:
    kb = InlineKeyboard()
    kb.button("Выключить" if is_active else "Включить", callback_data="toggle")
    return kb


class ProfileScreen(UIComponent):
    __prefix__ = "prof"
    prompt = "profile_main"
    markup = "profile_main"

    async def open(self, ctx: Context) -> None:
        user = await ctx.users.get(ctx.from_user["id"])
        kw = {"username": user["username"], "is_active": user["is_active"]}
        await self.send(ctx, prompt_kwargs=kw, markup_kwargs=kw)

    @UIComponent.callback("toggle")
    async def on_toggle(self, ctx: Context) -> None:
        user_id = ctx.from_user["id"]
        user = await ctx.users.get(user_id)
        await ctx.users.update(user_id, is_active=not user["is_active"])
        user = await ctx.users.get(user_id)
        kw = {"username": user["username"], "is_active": user["is_active"]}
        await self.update(ctx, prompt_kwargs=kw, markup_kwargs=kw)
        await ctx.answer_callback()


bot = (
    Bot(token, parse_mode="Markdown")
    .with_keyboards(keyboards)
    .with_prompts(prompts)
    .register_ui(ProfileScreen())
)

Полный пример с профилем в KV, inline-кнопками, UI-экранами, медиа-эхо и GitHub Actions-деплоем — в examples/echo/.

Деплой

Бот запускается как Cloudflare Worker (флаг python_workers). См. examples/echo/DEPLOY.md и examples/echo/wrangler.toml как референс.

Лицензия

MIT