Руководство по API AsyncMaxBot SDK¶
Этот раздел — ваш главный справочник по всем компонентам, классам и типам данных AsyncMaxBot SDK версии 1.4.3, основанный на актуальном коде.
ℹ️ Примечание: Эта библиотека является высокоуровневой оберткой. Для более глубокого понимания принципов работы, доступных методов и типов объектов, мы настоятельно рекомендуем ознакомиться с официальной документацией Max API.
1. Архитектура и поток данных¶
graph TD
A["Пользователь отправляет сообщение"] --> B["Max API"]
B --> C{"bot.polling"}
C --"JSON-обновление"--> D["Dispatcher"]
D --"Передает в MiddlewareManager"--> E["MiddlewareManager"]
E --"Исполняет Middleware"--> F["Парсинг в Pydantic-модели"]
F --"Создает Context"--> G["Context"]
G --"Проверяет фильтры"--> H{"Фильтры: F, command, text, ..."}
H --"ДА"--> I["Вызов вашего обработчика"]
I --"Использует Context для ответа"--> J["ctx reply"]
J --"Вызывает метод Bot"--> K["Bot send_message"]
K --"HTTP-запрос"--> B
B --> L["Пользователь получает ответ"]
H --"НЕТ"--> M["Поиск следующего обработчика"]
Поток обработки:
1. bot.polling(dispatcher=dp)
постоянно запрашивает у Max API новые обновления
2. Dispatcher
получает обновление и передает его в MiddlewareManager
3. Middleware обрабатывают входящий запрос по цепочке
4. Данные парсятся в строго типизированные Pydantic-модели
5. На основе этих моделей создается удобный объект Context
6. Dispatcher
ищет подходящий обработчик, проверяя сообщение на соответствие его фильтрам
7. Если фильтры пройдены, вызывается ваша асинхронная функция-обработчик с Context
в качестве аргумента
2. Класс Bot
¶
Основной класс для взаимодействия с Max API. Рекомендуется использовать как асинхронный контекстный менеджер.
Инициализация и запуск¶
from maxbot import Bot, Dispatcher
async with Bot(token=TOKEN) as bot:
dp = Dispatcher(bot)
# ... регистрация обработчиков ...
await bot.polling(dispatcher=dp)
Основные методы¶
Информация о боте¶
get_me()
->Dict[str, Any]
: Возвращает информацию о боте в виде словаря
Отправка сообщений¶
send_message(text: str, user_id: Optional[int] = None, chat_id: Optional[int] = None, reply_markup: Optional[Any] = None, **kwargs)
->Dict[str, Any]
: Отправляет текстовое сообщение- Обязательно: либо
user_id
, либоchat_id
reply_markup
: Поддержка inline клавиатурedit_message(message_id: str, text: str, reply_markup: Optional[Any] = None, **kwargs)
->Dict[str, Any]
: Редактирует сообщениеdelete_message(message_id: str)
->Dict[str, Any]
: Удаляет сообщение
Работа с файлами¶
upload_image(file: Union[str, BinaryIO, bytes], **kwargs)
->Dict[str, Any]
: Загружает изображениеupload_video(file: Union[str, BinaryIO, bytes], **kwargs)
->Dict[str, Any]
: Загружает видеоupload_audio(file: Union[str, BinaryIO, bytes], **kwargs)
->Dict[str, Any]
: Загружает аудиоupload_file(file: Union[str, BinaryIO, bytes], **kwargs)
->Dict[str, Any]
: Загружает файлupload_sticker(file: Union[str, BinaryIO, bytes], **kwargs)
->Dict[str, Any]
: Загружает стикерdownload_file(file_id: str, save_path: Optional[str] = None)
->Union[bytes, str]
: Скачивает файлget_file(file_id: str)
->Dict[str, Any]
: Получает информацию о файле
Отправка вложений¶
send_photo(photo: Union[str, BinaryIO, bytes, Dict], user_id: Optional[int] = None, chat_id: Optional[int] = None, caption: Optional[str] = None, **kwargs)
->Dict[str, Any]
: Отправляет фотоsend_video(video: Union[str, BinaryIO, bytes, Dict], user_id: Optional[int] = None, chat_id: Optional[int] = None, caption: Optional[str] = None, **kwargs)
->Dict[str, Any]
: Отправляет видеоsend_audio(audio: Union[str, BinaryIO, bytes, Dict], user_id: Optional[int] = None, chat_id: Optional[int] = None, caption: Optional[str] = None, **kwargs)
->Dict[str, Any]
: Отправляет аудиоsend_document(document: Union[str, BinaryIO, bytes, Dict], user_id: Optional[int] = None, chat_id: Optional[int] = None, caption: Optional[str] = None, **kwargs)
->Dict[str, Any]
: Отправляет документ
Работа с чатами¶
get_chat(chat_id: int)
->Dict[str, Any]
: Получает информацию о чатеget_chats(**kwargs)
->Dict[str, Any]
: Получает список чатовget_chat_members(chat_id: int, **kwargs)
->Dict[str, Any]
: Получает участников чатаadd_chat_members(chat_id: int, user_ids: List[int])
->Dict[str, Any]
: Добавляет участников в чатleave_chat(chat_id: int)
->Dict[str, Any]
: Покидает чат
Действия в чате¶
send_action(chat_id: int, action: str)
->Dict[str, Any]
: Отправляет действие (например, 'typing')pin_message(chat_id: int, message_id: str, **kwargs)
->Dict[str, Any]
: Закрепляет сообщениеunpin_message(chat_id: int, message_id: str)
->Dict[str, Any]
: Открепляет сообщение
Callback обработка¶
answer_callback_query(callback_id: str, **kwargs)
->Dict[str, Any]
: Отвечает на callback-запросsend_callback(callback_query_id: str, **kwargs)
->Dict[str, Any]
: Алиас для answer_callback_query
Валидация файлов¶
validate_file_size(file: Union[str, BinaryIO, bytes], max_size_mb: int = 50)
->bool
: Проверяет размер файлаvalidate_file_format(file: Union[str, BinaryIO, bytes], file_type: str)
->bool
: Проверяет формат файлаget_supported_formats(file_type: str)
->List[str]
: Получает поддерживаемые форматы
Получение обновлений¶
get_updates(offset: Optional[int] = None, limit: int = 100, timeout: int = 20)
->Dict
: Получает обновления от APIpolling(timeout: int = 1, long_polling_timeout: int = 20, dispatcher: 'Dispatcher' = None)
: Основной метод для запуска бота
3. Класс Dispatcher
¶
Диспетчер обрабатывает обновления, управляет middleware и вызывает нужные обработчики.
Инициализация¶
Регистрация обработчиков¶
Обработчики сообщений¶
@dp.message_handler(*filters)
: Декоратор для регистрации обработчика сообщений
Обработчики callback-запросов¶
@dp.callback_query_handler(*filters)
: Декоратор для регистрации обработчика callback-запросов
Обработчики событий (версия 1.4+)¶
@dp.bot_started_handler(*filters)
: Обработчик события запуска бота@dp.user_added_handler(*filters)
: Обработчик добавления пользователя в чат@dp.chat_member_updated_handler(*filters)
: Обработчик изменения статуса участника
Middleware¶
dp.include_middleware(middleware)
: Добавляет middleware в цепочку обработкиdp.include_router(router)
: Подключает роутер к диспетчеру
4. Класс Context
¶
Ключевой объект, с которым вы работаете в обработчиках. Содержит все данные об обновлении и удобные методы для ответа.
Основные атрибуты¶
Информация о пользователе и чате¶
user: Optional[User]
: Объект пользователяchat: Optional[Chat]
: Объект чатаuser_id: Optional[int]
: ID пользователяchat_id: Optional[int]
: ID чата
Информация о сообщении¶
message: Optional[Message]
: Объект сообщенияtext: Optional[str]
: Текст сообщенияmessage_id: Optional[str]
: ID сообщенияdate: int
: Временная метка сообщения
Вложения¶
attachments: Optional[List[BaseAttachment]]
: Список вложенийhas_attachments: bool
: True, если есть вложенияimages: List[BaseAttachment]
: Список изображений
Callback информация¶
is_callback: bool
: True, если это callback-запросpayload: Optional[str]
: Payload callback-запросаcallback_id: Optional[str]
: ID callback-запроса
Расширенные события (версия 1.4+)¶
bot_started: Optional[BotStarted]
: Данные события запуска ботаuser_added: Optional[UserAdded]
: Данные добавления пользователяchat_member_updated: Optional[ChatMemberUpdated]
: Данные изменения статуса участника
Основные методы¶
Отправка сообщений¶
reply(text: str, reply_markup: Optional[Any] = None, **kwargs)
->Dict[str, Any]
: Ответить в тот же чатanswer(text: str, reply_markup: Optional[Any] = None, **kwargs)
->Dict[str, Any]
: Псевдоним для reply
Редактирование и удаление¶
edit_message(text: str, reply_markup: Optional[Any] = None, **kwargs)
->Dict[str, Any]
: Редактировать сообщениеdelete_message(**kwargs)
->Dict[str, Any]
: Удалить сообщение
Callback ответы¶
answer_callback(text: Optional[str] = None, message: Optional[dict] = None, show_alert: bool = False, **kwargs)
->Dict[str, Any]
: Ответить на callback
Работа с чатом¶
get_members(**kwargs)
->Dict[str, Any]
: Получить участников чатаsend_action(action: str)
->Dict[str, Any]
: Отправить действиеpin_message(message_id: Optional[str] = None, **kwargs)
->Dict[str, Any]
: Закрепить сообщениеunpin_message(message_id: str)
->Dict[str, Any]
: Открепить сообщениеleave_chat()
->Dict[str, Any]
: Покинуть чатadd_members(user_ids: List[int])
->Dict[str, Any]
: Добавить участников
5. MagicFilter система (F)¶
MagicFilter — это мощная система фильтрации, которая позволяет создавать сложные условия в стиле, похожем на SQL или ORM.
Импорт¶
Основные возможности¶
Сравнения¶
@dp.message_handler(F.text == "привет")
@dp.message_handler(F.user_id == 123)
@dp.message_handler(F.chat_id != 456)
@dp.message_handler(F.user_id > 100)
@dp.message_handler(F.user_id <= 1000)
Строковые операции¶
@dp.message_handler(F.text.contains("hello"))
@dp.message_handler(F.text.startswith("/"))
@dp.message_handler(F.text.endswith("!"))
Проверки вхождений¶
@dp.message_handler(F.user_id.in_([1, 2, 3, 4, 5]))
@dp.message_handler(F.command.in_(["start", "help", "info"]))
Работа с вложениями¶
@dp.message_handler(F.has_attachments == True)
@dp.message_handler(F.attachment.type == "image")
@dp.message_handler(F.attachment.type.in_(["image", "video"]))
Комбинирование фильтров¶
# AND операция
@dp.message_handler(F.text.contains("admin") & F.user_id == 123)
# OR операция
@dp.message_handler(F.command == "start" | F.text.startswith("help"))
# NOT операция
@dp.message_handler(~F.text.startswith("/"))
# Сложные комбинации
@dp.message_handler(
(F.command == "start" | F.text.contains("привет")) &
F.user_id.in_([1, 2, 3])
)
Доступные атрибуты¶
F.text
- текст сообщенияF.command
- команда (без /)F.user
- объект пользователяF.user_id
- ID пользователяF.chat
- объект чатаF.chat_id
- ID чатаF.message_id
- ID сообщенияF.attachment
- вложенияF.has_attachments
- наличие вложенийF.payload
- payload callback-запроса (только для callback)
6. Inline клавиатуры и Callback¶
Создание клавиатур¶
from maxbot.max_types import InlineKeyboardMarkup, InlineKeyboardButton
# Простая клавиатура
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="Кнопка 1", payload="action1")],
[InlineKeyboardButton(text="Кнопка 2", payload="action2")]
]
)
# Клавиатура с URL
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="Открыть сайт", url="https://example.com")],
[InlineKeyboardButton(text="Действие", payload="action")]
]
)
# Многострочная клавиатура
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(text="⬅️", payload="prev"),
InlineKeyboardButton(text="➡️", payload="next")
],
[InlineKeyboardButton(text="Главное меню", payload="menu")]
]
)
Отправка с клавиатурой¶
@dp.message_handler(F.command == "menu")
async def show_menu(ctx: Context):
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="🎮 Играть", payload="play")],
[InlineKeyboardButton(text="📊 Статистика", payload="stats")],
[InlineKeyboardButton(text="ℹ️ Помощь", payload="help")]
]
)
await ctx.reply("Выберите действие:", reply_markup=keyboard)
Обработка callback¶
@dp.callback_query_handler(F.payload == "play")
async def handle_play(ctx: Context):
# Отвечаем на callback (убираем "часики")
await ctx.answer_callback("🎮 Начинаем игру!")
# Редактируем сообщение
await ctx.edit_message("Игра началась! 🎲")
@dp.callback_query_handler(F.payload == "stats")
async def handle_stats(ctx: Context):
await ctx.answer_callback("📊 Статистика загружается...")
await ctx.edit_message("Ваша статистика: 10 игр, 7 побед")
@dp.callback_query_handler(F.payload == "help")
async def handle_help(ctx: Context):
# Показываем alert
await ctx.answer_callback("Это справка по игре!", show_alert=True)
Методы Context для callback¶
ctx.answer_callback(text=None, message=None, show_alert=False, **kwargs)
: Отвечает на callbackctx.edit_message(text, reply_markup=None, **kwargs)
: Редактирует сообщениеctx.payload
: Получает payload кнопкиctx.callback_id
: ID callback-запроса
7. Middleware¶
Middleware позволяют выполнять сквозную логику для всех обновлений.
Импорт¶
from maxbot.middleware import (
LoggingMiddleware,
ThrottlingMiddleware,
ErrorHandlingMiddleware,
UserTrackingMiddleware,
MetricsMiddleware,
AntispamMiddleware,
ValidationMiddleware,
ProfilingMiddleware
)
Пример использования¶
dp.include_middleware(LoggingMiddleware())
dp.include_middleware(ThrottlingMiddleware(rate_limit=1.0)) # 1 сообщ./сек
dp.include_middleware(ErrorHandlingMiddleware())
Встроенные Middleware¶
LoggingMiddleware¶
Логирует информацию о входящих сообщениях.
ErrorHandlingMiddleware¶
Перехватывает ошибки в обработчиках, чтобы бот не падал.
async def error_handler(ctx: Context, error: Exception):
logger.error(f"Error in handler: {error}")
await ctx.reply("Произошла ошибка, но я уже сообщил о ней разработчикам.")
dp.include_middleware(ErrorHandlingMiddleware(error_handler=error_handler))
ThrottlingMiddleware¶
Ограничивает частоту сообщений от одного пользователя.
UserTrackingMiddleware¶
Отслеживает активных пользователей.
user_tracker = UserTrackingMiddleware()
dp.include_middleware(user_tracker)
# Получить количество активных пользователей
active_count = user_tracker.get_active_users_count()
MetricsMiddleware¶
Собирает метрики (время работы, кол-во сообщений/ошибок).
metrics = MetricsMiddleware()
dp.include_middleware(metrics)
# Получить метрики
stats = metrics.get_metrics()
print(f"Обработано сообщений: {stats['messages_processed']}")
AntispamMiddleware¶
Блокирует одинаковые сообщения от пользователя за короткий промежуток.
ProfilingMiddleware¶
Профилирование времени обработки.
profiler = ProfilingMiddleware()
dp.include_middleware(profiler)
# Получить среднее время обработки
avg_time = profiler.get_avg_time()
Создание кастомного Middleware¶
from maxbot.middleware import BaseMiddleware
class CustomMiddleware(BaseMiddleware):
async def __call__(self, handler, ctx):
# Логика до обработчика
print(f"Обрабатывается сообщение от {ctx.user_id}")
result = await handler(ctx)
# Логика после обработчика
print(f"Сообщение обработано")
return result
dp.include_middleware(CustomMiddleware())
8. Фильтры¶
Фильтры решают, будет ли вызван обработчик. Их можно комбинировать.
Классические фильтры¶
from maxbot.filters import command, text, regex, attachment_type, has_attachment
@dp.message_handler(command("start"))
async def start_handler(ctx: Context): ...
@dp.message_handler(text("привет", exact=False))
async def hello_handler(ctx: Context): ...
@dp.message_handler(regex(r"^\d+$"))
async def number_handler(ctx: Context): ...
@dp.message_handler(attachment_type("image"))
async def image_handler(ctx: Context): ...
@dp.message_handler(has_attachment(True))
async def attachment_handler(ctx: Context): ...
Дополнительные фильтры¶
user_filter(user_ids)
: Срабатывает, если сообщение от указанного пользователяtime_filter(start_hour, end_hour)
: Срабатывает в определенный промежуток времениcustom_filter(func)
: Ваш собственный фильтр на основе функцииlambda ctx: ...
Комбинирование фильтров¶
from maxbot.filters import and_filter, or_filter, not_filter
@dp.message_handler(and_filter(command("admin"), user_filter([123, 456])))
@dp.message_handler(or_filter(command("start"), text("помощь")))
@dp.message_handler(not_filter(text("спам")))
9. Router система (версия 1.4+)¶
Router позволяет создавать модульную архитектуру и изолировать логику обработчиков.
Создание роутера¶
from maxbot import Router
# Создаем роутер для команд
commands_router = Router()
@commands_router.message_handler(F.command == "start")
async def start_command(ctx: Context):
await ctx.reply("Привет! Это команда start")
@commands_router.message_handler(F.command == "help")
async def help_command(ctx: Context):
await ctx.reply("Это справка")
# Создаем роутер для callback
callback_router = Router()
@callback_router.callback_query_handler(F.payload == "action")
async def handle_action(ctx: Context):
await ctx.answer_callback("Действие выполнено!")
Подключение роутера к диспетчеру¶
Структура проекта с роутерами¶
my_bot/
├── main.py
├── routers/
│ ├── __init__.py
│ ├── commands.py
│ ├── callbacks.py
│ └── events.py
└── handlers/
├── __init__.py
└── common.py
10. Типы данных¶
SDK использует Pydantic для валидации. Основные модели импортируются из maxbot
.
Основные модели¶
User¶
class User(BaseModel):
user_id: int
name: str = ""
first_name: Optional[str] = None
last_name: Optional[str] = None
is_bot: bool = False
last_activity_time: Optional[int] = 0
Chat¶
class Chat(BaseModel):
chat_id: int
chat_type: str # "private" или "group"
user_id: Optional[int] = None
Message¶
MessageBody¶
class MessageBody(BaseModel):
mid: str
seq: int
text: Optional[str] = None
attachments: Optional[List[BaseAttachment]] = None
BaseAttachment¶
class BaseAttachment(BaseModel):
type: str
payload: Optional[AttachmentPayload] = None
url: Optional[str] = None
file_id: Optional[str] = None
filename: Optional[str] = None
size: Optional[int] = None
mime_type: Optional[str] = None
width: Optional[int] = None
height: Optional[int] = None
duration: Optional[int] = None
performer: Optional[str] = None
title: Optional[str] = None
thumbnail: Optional[AttachmentThumbnail] = None
latitude: Optional[float] = None
longitude: Optional[float] = None
emoji: Optional[str] = None
Клавиатуры¶
InlineKeyboardButton¶
class InlineKeyboardButton(BaseModel):
type: str = 'callback'
text: str
payload: Optional[str] = None
url: Optional[str] = None
InlineKeyboardMarkup¶
Callback¶
CallbackQuery¶
class CallbackQuery(BaseModel):
callback_id: str
user: User
payload: Optional[str] = None
message: Optional[Message] = None
Расширенные события (версия 1.4+)¶
BotStarted¶
UserAdded¶
ChatMemberUpdated¶
11. Обработка ошибок¶
Встроенная обработка¶
from maxbot.middleware import ErrorHandlingMiddleware
dp.include_middleware(ErrorHandlingMiddleware())
Кастомная обработка¶
async def error_handler(ctx: Context, error: Exception):
logger.error(f"Error in handler: {error}")
await ctx.reply("Произошла ошибка, но я уже сообщил о ней разработчикам.")
dp.include_middleware(ErrorHandlingMiddleware(error_handler=error_handler))
Try-catch в обработчиках¶
@dp.message_handler(F.command == "risky")
async def risky_handler(ctx: Context):
try:
# Рискованная операция
result = await some_risky_operation()
await ctx.reply(f"Результат: {result}")
except ValueError as e:
await ctx.reply(f"Ошибка валидации: {e}")
except Exception as e:
logger.error(f"Неожиданная ошибка: {e}")
await ctx.reply("Произошла неожиданная ошибка")
12. Логирование¶
SDK использует loguru для логирования.
from loguru import logger
# Настройка логирования
logger.add("bot.log", rotation="1 day", retention="7 days")
# В обработчиках
@dp.message_handler(F.command == "debug")
async def debug_handler(ctx: Context):
logger.info(f"User {ctx.user_id} sent command: {ctx.text}")
logger.debug(f"Full context: {ctx}")
13. Лучшие практики¶
Структура проекта¶
my_bot/
├── main.py # Точка входа
├── routers/ # Роутеры (версия 1.4+)
│ ├── __init__.py
│ ├── commands.py
│ ├── callbacks.py
│ └── events.py
├── handlers/ # Обработчики
│ ├── __init__.py
│ ├── common.py
│ └── utils.py
├── middleware/ # Middleware
│ ├── __init__.py
│ └── custom.py
├── utils/ # Утилиты
│ ├── __init__.py
│ └── helpers.py
└── config.py # Конфигурация
Организация обработчиков с роутерами¶
# routers/commands.py
from maxbot import Router, F, Context
def create_commands_router():
router = Router()
@router.message_handler(F.command == "start")
async def start_handler(ctx: Context):
await ctx.reply("Привет! Это команда start")
@router.message_handler(F.command == "help")
async def help_handler(ctx: Context):
await ctx.reply("Это справка")
return router
# main.py
from routers.commands import create_commands_router
dp = Dispatcher(bot)
commands_router = create_commands_router()
dp.include_router(commands_router)
Управление состоянием¶
# Простое состояние в словаре
user_states = {}
@dp.message_handler(F.command == "game")
async def start_game(ctx: Context):
user_states[ctx.user_id] = {"game": "started", "score": 0}
await ctx.reply("Игра началась!")
@dp.message_handler(lambda ctx: user_states.get(ctx.user_id, {}).get("game") == "started")
async def game_handler(ctx: Context):
# Обработка игровых сообщений
pass
Обработка расширенных событий¶
@dp.bot_started_handler()
async def on_bot_started(ctx: Context):
await ctx.reply(f"Бот запущен пользователем {ctx.user.name}")
@dp.user_added_handler()
async def on_user_added(ctx: Context):
await ctx.reply(f"Пользователь {ctx.user.name} добавлен в чат")
@dp.chat_member_updated_handler()
async def on_member_updated(ctx: Context):
await ctx.reply(f"Статус {ctx.user.name} изменен: {ctx.old_status} -> {ctx.new_status}")