maxbot-chatbot-python — это асинхронный фреймворк для создания масштабируемых ботов для MAX BOT API на языке Python.
Построенная на основе maxbot_api_client_python, эта библиотека предоставляет чистый маршрутизатор, автоматическое получение обновлений (Long Polling) и надежный менеджер состояний (FSM) для построения многошаговых диалоговых сценариев.
Для использования библиотеки требуется получить токен бота в консоли разработчика MAX 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(...)) гарантирует безопасное закрытие сетевых соединений при остановке бота.
importasynciofrommaxbot_api_client_pythonimportAPI, Configfrommaxbot_chatbot_pythonimportBot, MapStateManagerasyncdefmain():
cfg=Config(
base_url="https://platform-api.max.ru/", token="YOUR_BOT_TOKEN", ratelimiter=25,
timeout=35
)
asyncwithAPI(cfg) asapi_client:
bot=Bot(api_client)
bot.state_manager=MapStateManager(init_data={})
polling_task=asyncio.create_task(bot.start_polling())
try:
awaitpolling_taskexceptasyncio.CancelledError:
passif__name__=="__main__":
try:
asyncio.run(main())
exceptKeyboardInterrupt:
print("Bot stopped by user")Встроенный маршрутизатор (Router) позволяет легко обрабатывать конкретные команды (начинающиеся со слэша /) и нажатия на inline-кнопки (коллбэки).
@bot.router.command("/start")asyncdefstart_command(notification):
awaitnotification.reply("Hello! Welcome to the MAX Bot.")
@bot.router.register("message_created")asyncdefping_handler(notification):
try:
ifnotification.text() =="ping":
awaitnotification.reply("pong")
exceptValueError:
pass@bot.router.callback("accept_rules")asyncdefrules_callback(notification):
awaitnotification.reply("*Thank you for accepting the rules!*", format_type="markdown")
awaitnotification.answer_callback("Success!")Для сложных многошаговых диалогов (например, регистрация или анкетирование) используйте Менеджер состояний (StateManager) и Сцены (Scene).
frommaxbot_chatbot_pythonimportSceneclassRegistrationScene(Scene):
asyncdefstart(self, notification):
try:
text=notification.text()
exceptValueError:
returniftext=="/start":
awaitnotification.reply("Let's register! What is your *login*?", "markdown")
returniflen(text) >=4:
ifnotification.state_manager:
notification.state_manager.update_state_data(notification.state_id, {"login": text})
awaitnotification.reply(f"**Login** `{text}` accepted. Now enter your **password**:", "markdown")
notification.activate_next_scene(PasswordScene())
else:
awaitnotification.reply("Login must be **at least 4 characters long**.", "markdown")
classPasswordScene(Scene):
asyncdefstart(self, notification):
try:
password=notification.text()
exceptValueError:
returnstate_data=notification.state_manager.get_state_data(notification.state_id)
login=state_data.get("login", "Unknown")
awaitnotification.reply(f"Success! Profile created.\nLogin: `{login}`\nPass: `{password}`", "markdown")
notification.activate_next_scene(RegistrationScene())
@bot.router.register("message_created")asyncdeffsm_handler(notification):
ifnotnotification.state_manager.get(notification.state_id):
notification.state_manager.create(notification.state_id)
current_scene=notification.get_current_scene()
ifcurrent_scene:
awaitcurrent_scene.start(notification)Обертка Notification содержит готовые асинхронные методы для отправки файлов, геолокаций, стикеров и статусов набора текста.
@bot.router.command("/photo")asyncdefsend_photo(notification):
awaitnotification.show_action("sending_photo")
awaitnotification.reply_with_media(
text="Check out this image!", format_type="markdown", file_source="https://storage.yandexcloud.net/sw-prod-03-test/ChatBot/corgi.jpg"
)importasynciofrommaxbot_api_client_pythonimportAPI, Configfrommaxbot_chatbot_pythonimportBot, MapStateManagerasyncdefmain():
cfg=Config(
base_url="https://platform-api.max.ru/", token="YOUR_BOT_TOKEN",
ratelimiter=25
)
asyncwithAPI(cfg) asapi_client:
bot=Bot(api_client)
bot.state_manager=MapStateManager(init_data={})
@bot.router.register("message_created")asyncdefecho_handler(notification):
try:
text=notification.text()
awaitnotification.reply(f"**Echo:** {text}", "markdown")
exceptExceptionase:
print(f"Error handling message: {e}")
polling_task=asyncio.create_task(bot.start_polling())
try:
awaitpolling_taskexceptasyncio.CancelledError:
passif__name__=="__main__":
try:
asyncio.run(main())
exceptKeyboardInterrupt:
print("Bot stopped by user (KeyboardInterrupt)")