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(...)) гарантирует безопасное закрытие сетевых соединений при остановке бота.
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!")Для сложных многошаговых диалогов (например, регистрация или анкетирование) используйте Менеджер состояний (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):
if not notification.state_manager.get(notification.state_id):
notification.state_manager.create(notification.state_id)
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)")