maxbot-chatbot-python 1.1.0

Python SDK and Chatbot Framework for MAX API

maxbot-chatbot-python

maxbot-chatbot-python — это асинхронный фреймворк для создания масштабируемых ботов для MAX BOT API на языке Python.

Построенная на основе maxbot_api_client_python, эта библиотека предоставляет чистый маршрутизатор, автоматическое получение обновлений (Long Polling) и надежный менеджер состояний (FSM) для построения многошаговых диалоговых сценариев.

Для использования библиотеки требуется получить токен бота в консоли разработчика MAX API.
Ознакомиться с инструкцией можно по ссылке.

API

Документацию по REST API MAX можно найти по ссылке dev.max.ru/docs-api. Библиотека является оберткой для REST API, поэтому документация по указанной выше ссылке также применима к используемым здесь моделям.

Документацию по MAX BOT API можно найти по ссылке green-api.com/max-bot-api/docs.

Поддержка

Руководства и новости

Установка

Убедитесь, что у вас установлен Python версии 3.12 или выше.

python --version

Установите библиотеку:

pip install maxbot-chatbot-python

---

Использование и примеры

Параметры конфигурации:

• base_url - Базовый URL-адрес серверов платформы MaxBot. Все методы API будут маршрутизироваться по этому корневому адресу. Актуальный адрес указан в официальной документации.
• token - Уникальный секретный ключ авторизации (API-ключ) вашего бота. Получить его можно в личном кабинете после регистрации или создании бота на платформе business.max.ru.
• ratelimiter - Встроенный ограничитель частоты запросов. Он контролирует количество исходящих запросов в секунду (RPS), защищая бота от блокировки со стороны сервера за превышение лимитов. Рекомендуемое значение — не менее 25.
• timeout - Максимальное время ожидания ответа от сервера (в секундах). Если сервер не ответит в течение этого времени, запрос будет завершен с ошибкой. Оптимальное значение — 30 секунд.

Инициализация бота

Использование асинхронного контекстного менеджера (async with API(...)) гарантирует безопасное закрытие сетевых соединений при остановке бота.

import asyncio

from maxbot_api_client_python import API, Config
from maxbot_chatbot_python import Bot, MapStateManager

async def main():
    cfg = Config(
        base_url="https://platform-api.max.ru/", 
        token="YOUR_BOT_TOKEN", 
        ratelimiter=25,
        timeout=35
    )

    async with API(cfg) as api_client:
        bot = Bot(api_client)
        bot.state_manager = MapStateManager(init_data={})

        polling_task = asyncio.create_task(bot.start_polling())

        try:
            await polling_task
        except asyncio.CancelledError:
            pass

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Bot stopped by user")

Маршрутизация команд, сообщений и коллбэков

Встроенный маршрутизатор (Router) позволяет легко обрабатывать конкретные команды (начинающиеся со слэша /) и нажатия на inline-кнопки (коллбэки).

@bot.router.command("/start")
async def start_command(notification):
    await notification.reply("Hello! Welcome to the MAX Bot.")

@bot.router.register("message_created")
async def ping_handler(notification):
    try:
        if notification.text() == "ping":
            await notification.reply("pong")
    except ValueError:
        pass

@bot.router.callback("accept_rules")
async def rules_callback(notification):
    await notification.reply("*Thank you for accepting the rules!*", format_type="markdown")
    await notification.answer_callback("Success!")

Управление состояниями и Сцены (FSM)

Для сложных многошаговых диалогов (например, регистрация или анкетирование) используйте Менеджер состояний (StateManager) и Сцены (Scene).

from maxbot_chatbot_python import Scene

class RegistrationScene(Scene):
    async def start(self, notification):
        try:
            text = notification.text()
        except ValueError:
            return

        if text == "/start":
            await notification.reply("Let's register! What is your *login*?", "markdown")
            return 
        
        if len(text) >= 4:
            if notification.state_manager:
                notification.state_manager.update_state_data(notification.state_id, {"login": text})
            
            await notification.reply(f"**Login** `{text}` accepted. Now enter your **password**:", "markdown")
            notification.activate_next_scene(PasswordScene())
        else:
            await notification.reply("Login must be **at least 4 characters long**.", "markdown")

class PasswordScene(Scene):
    async def start(self, notification):
        try:
            password = notification.text()
        except ValueError:
            return

        state_data = notification.state_manager.get_state_data(notification.state_id)
        login = state_data.get("login", "Unknown")

        await notification.reply(f"Success! Profile created.\nLogin: `{login}`\nPass: `{password}`", "markdown")

        notification.activate_next_scene(RegistrationScene())

@bot.router.register("message_created")
async def fsm_handler(notification):
    current_scene = notification.get_current_scene()
    if current_scene:
        await current_scene.start(notification)

Ответ с медиафайлами

Обертка Notification содержит готовые асинхронные методы для отправки файлов, геолокаций, стикеров и статусов набора текста.

@bot.router.command("/photo")
async def send_photo(notification):
    await notification.show_action("sending_photo")

    await notification.reply_with_media(
        text="Check out this image!", 
        format_type="markdown", 
        file_source="https://storage.yandexcloud.net/sw-prod-03-test/ChatBot/corgi.jpg"
    )

Эхо-бот

import asyncio

from maxbot_api_client_python import API, Config
from maxbot_chatbot_python import Bot, MapStateManager

async def main():
    cfg = Config(
        base_url="https://platform-api.max.ru/", 
        token="YOUR_BOT_TOKEN",
        ratelimiter=25
    )

    async with API(cfg) as api_client:
        bot = Bot(api_client)
        bot.state_manager = MapStateManager(init_data={})

        @bot.router.register("message_created")
        async def echo_handler(notification):
            try:
                text = notification.text()
                await notification.reply(f"**Echo:** {text}", "markdown")
            except Exception as e:
                print(f"Error handling message: {e}")

        polling_task = asyncio.create_task(bot.start_polling())

        try:
            await polling_task
        except asyncio.CancelledError:
            pass

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Bot stopped by user (KeyboardInterrupt)")