Примеры

Здесь собраны практические примеры использования MaxAPI для различных задач.

Эхо-бот

Простейший пример бота, который повторяет все текстовые сообщения:

import asyncio
import logging

from maxapi import Bot, Dispatcher, F
from maxapi.types import MessageCreated

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


@dp.message_created(F.message.body.text)
async def echo(event: MessageCreated):
    await event.message.answer(f"Повторяю за вами: {event.message.body.text}")


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Обработка событий

Пример обработки различных событий, доступных в MaxAPI:

import asyncio
import logging

from maxapi import Bot, Dispatcher
from maxapi.types import (
    BotStarted,
    Command,
    MessageCreated,
    CallbackButton,
    MessageCallback,
    BotAdded,
    ChatTitleChanged,
    MessageEdited,
    MessageRemoved,
    UserAdded,
    UserRemoved,
    BotStopped,
    DialogCleared,
    DialogMuted,
    DialogUnmuted,
    ChatButton,  # deprecated: 0.9.14
    MessageChatCreated,  # deprecated: 0.9.14
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


@dp.message_created(Command("start"))
async def hello(event: MessageCreated):
    builder = InlineKeyboardBuilder()

    builder.row(
        CallbackButton(
            text="Кнопка 1",
            payload="btn_1",
        ),
        CallbackButton(
            text="Кнопка 2",
            payload="btn_2",
        ),
    )
    builder.add(
        ChatButton(  # deprecated: 0.9.14
            text="Создать чат",
            chat_title="Тест чат",
        )
    )

    await event.message.answer(
        text="Привет!",
        attachments=[
            builder.as_markup(),
        ],
    )


@dp.bot_added()
async def bot_added(event: BotAdded):
    chat = await event.fetch_chat()
    if chat is None:
        logging.info(
            "Не удалось получить chat, возможно отключен auto_requests!"
        )
        return

    await bot.send_message(
        chat_id=event.chat_id, text=f"Привет чат {chat.title}!"
    )


@dp.message_removed()
async def message_removed(event: MessageRemoved):
    await bot.send_message(
        chat_id=event.chat_id,
        text="Я всё видел!",
    )


@dp.bot_started()
async def bot_started(event: BotStarted):
    await bot.send_message(
        chat_id=event.chat_id,
        text="Привет! Отправь мне /start",
    )


@dp.chat_title_changed()
async def chat_title_changed(event: ChatTitleChanged):
    await bot.send_message(
        chat_id=event.chat_id,
        text=f'Крутое новое название "{event.title}"!',
    )


@dp.message_callback()
async def message_callback(event: MessageCallback):
    await event.answer(
        new_text=f"Вы нажали на кнопку {event.callback.payload}!"
    )


@dp.message_edited()
async def message_edited(event: MessageEdited):
    await event.message.answer(text="Вы отредактировали сообщение!")


@dp.user_removed()
async def user_removed(event: UserRemoved):
    from_user = await event.fetch_from_user()
    if from_user is None:
        return await bot.send_message(
            chat_id=event.chat_id,
            text=f"Неизвестный кикнул {event.user.first_name} 😢",
        )

    await bot.send_message(
        chat_id=event.chat_id,
        text=f"{from_user.first_name} кикнул {event.user.first_name} 😢",
    )


@dp.user_added()
async def user_added(event: UserAdded):
    chat = await event.fetch_chat()
    if chat is None:
        return await bot.send_message(
            chat_id=event.chat_id,
            text=f"Чат приветствует вас, {event.user.first_name}!",
        )

    await bot.send_message(
        chat_id=event.chat_id,
        text=f'Чат "{chat.title}" приветствует вас, {event.user.first_name}!',
    )


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Если бот создан с auto_requests=False, то event.chat и event.from_user могут быть не загружены автоматически. Чтобы безопасно получить их независимо от режима, используйте явный fetch:

chat = await event.fetch_chat()
from_user = await event.fetch_from_user()

После этого работайте уже с локальными переменными:

if chat is not None:
    print(chat.title)

if from_user is not None:
    print(from_user.first_name)

MagicFilter

Использование MagicFilter для гибкой фильтрации сообщений:

import asyncio
import logging

from maxapi import Bot, Dispatcher, F
from maxapi.types import MessageCreated

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


@dp.message_created(F.message.body.text == "привет")
async def on_hello(event: MessageCreated):
    await event.message.answer("Привет!")


@dp.message_created(F.message.body.text.lower().contains("помощь"))
async def on_help(event: MessageCreated):
    await event.message.answer("Чем могу помочь?")


@dp.message_created(F.message.body.text.regexp(r"^\d{4}$"))
async def on_code(event: MessageCreated):
    await event.message.answer("Принят 4-значный код")


@dp.message_created(F.message.body.attachments)
async def on_attachment(event: MessageCreated):
    await event.message.answer("Получено вложение")


@dp.message_created(F.message.body.text.len() > 20)
async def on_long_text(event: MessageCreated):
    await event.message.answer("Слишком длинное сообщение")


@dp.message_created(F.message.body.text.len() > 0)
async def on_non_empty(event: MessageCreated):
    await event.message.answer("Вы что-то написали.")


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Клавиатуры

Примеры работы с различными типами кнопок и клавиатур:

import asyncio
import logging

from maxapi import Bot, Dispatcher
from maxapi.types import (
    ChatButton,  # deprecated: 0.9.14
    LinkButton,
    CallbackButton,
    RequestGeoLocationButton,
    MessageButton,
    ButtonsPayload,
    RequestContactButton,
    OpenAppButton,
    MessageCreated,
    MessageCallback,
    MessageChatCreated,  # deprecated: 0.9.14
    CommandStart,
    Command,
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


@dp.message_created(CommandStart())
async def echo(event: MessageCreated):
    await event.message.answer(
        (
            "Привет! Мои команды:\n\n"
            "/builder - Клавиатура из InlineKeyboardBuilder\n"
            "/payload - Клавиатура из pydantic моделей\n"
        )
    )


@dp.message_created(Command("builder"))
async def builder(event: MessageCreated):
    builder = InlineKeyboardBuilder()

    builder.row(
        ChatButton(  # deprecated: 0.9.14
            text="Создать чат",
            chat_title="Test",
            chat_description="Test desc",
        ),
        LinkButton(
            text="Документация MAX",
            url="https://dev.max.ru/docs",
        ),
    )

    builder.row(
        RequestGeoLocationButton(text="Геолокация"),
        MessageButton(text="Сообщение"),
    )

    builder.row(
        RequestContactButton(text="Контакт"),
        OpenAppButton(
            text="Приложение",
            web_app=event.bot.me.username,
            contact_id=event.bot.me.user_id,
        ),
    )

    builder.row(
        CallbackButton(
            text="Callback",
            payload="test",
        )
    )

    await event.message.answer(
        text="Клавиатура из InlineKeyboardBuilder",
        attachments=[builder.as_markup()],
    )


@dp.message_created(Command("payload"))
async def payload(event: MessageCreated):
    buttons = [
        [
            ChatButton(  # deprecated: 0.9.14
                text="Создать чат",
                chat_title="Test",
                chat_description="Test desc",
            ),
            LinkButton(
                text="Документация MAX",
                url="https://dev.max.ru/docs",
            ),
        ],
        [
            RequestGeoLocationButton(text="Геолокация"),
            MessageButton(text="Сообщение"),
        ],
        [
            RequestContactButton(text="Контакт"),
            OpenAppButton(
                text="Приложение",
                web_app=event.bot.me.username,
                contact_id=event.bot.me.user_id,
            ),
        ],
        [
            CallbackButton(
                text="Callback",
                payload="test",
            )
        ],
    ]

    buttons_payload = ButtonsPayload(buttons=buttons).pack()

    await event.message.answer(
        text="Клавиатура из pydantic моделей",
        attachments=[buttons_payload],
    )


@dp.message_chat_created()  # deprecated: 0.9.14
async def message_chat_created(obj: MessageChatCreated):
    await obj.bot.send_message(
        chat_id=obj.chat.chat_id,
        text=f"Чат создан! Ссылка: {obj.chat.link}",
    )


@dp.message_callback()
async def message_callback(callback: MessageCallback):
    await callback.message.answer("Вы нажали на Callback!")


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Форматирование текста

Для сборки текста в HTML и Markdown можно использовать helper-ы из maxapi.utils.formatting.

Элемент	HTML	Markdown
Заголовок	<h1>Заголовок</h1>	# Заголовок
Жирный	<b>текст</b>	**текст**
Курсив	<i>текст</i>	*текст*
Подчёркнутый	<ins>текст</ins>	++текст++
Зачёркнутый	<s>текст</s>	~~текст~~
Моноширинный	<code>текст</code>	`текст`
Цитата	<blockquote>текст</blockquote>	> текст
Ссылка	<a href="https://example.com">текст</a>	[текст](https://example.com)

Пример HTML:

from maxapi.enums.format import Format
from maxapi.utils.formatting import (
    Blockquote,
    Bold,
    Heading,
    Italic,
    Link,
    as_html,
)

text = as_html(
    Heading("Проверка форматирования"),
    "\n",
    Bold("Жирный текст"),
    "\n",
    Italic("Курсивная строка"),
    "\n",
    Blockquote("Цитата"),
    "\n",
    Link("Документация", url="https://love-apples.github.io/maxapi/"),
)

await event.message.answer(text, format=Format.HTML)

Пример Markdown:

from maxapi.enums.format import Format
from maxapi.utils.formatting import Blockquote, Bold, Heading, as_markdown

text = as_markdown(
    Heading("Проверка форматирования"),
    "\n",
    Bold("Жирный текст"),
    "\n",
    Blockquote("Цитата"),
)

await event.message.answer(text, format=Format.MARKDOWN)

Если вы читаете входящее сообщение из API, можно получить каноническое представление текста через message.body.html_text и message.body.md_text. Это нормализованный вывод из markup, поэтому он может не совпадать с исходной строкой символ в символ.

Получение ID

Пример получения различных ID из событий:

import asyncio
import logging

from maxapi import Bot, Dispatcher, F
from maxapi.enums.format import Format
from maxapi.types import MessageCreated

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


@dp.message_created(F.message.link.type == "forward")
async def get_ids_from_forward(event: MessageCreated):
    text = (
        "Информация о пересланном сообщении:\n\n"
        f"Из чата: <b>{event.message.link.chat_id}</b>\n"
        f"От пользователя: <b>{event.message.link.sender.user_id}</b>"
    )
    await event.message.reply(text)


@dp.message_created()
async def get_ids(event: MessageCreated):
    from_user = await event.fetch_from_user()
    chat = await event.fetch_chat()

    if from_user is None or chat is None:
        return

    text = (
        f"Ваш ID: <b>{from_user.user_id}</b>\n"
        f"ID этого чата: <b>{chat.chat_id}</b>"
    )
    await event.message.answer(text, format=Format.HTML)


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Создание собственного фильтра

Пример создания кастомного фильтра на основе BaseFilter:

import asyncio
import logging

from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, CommandStart, UpdateUnion
from maxapi.filters import BaseFilter

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


class FilterChat(BaseFilter):
    """
    Фильтр, который срабатывает только в чате с названием `Test`
    """

    async def __call__(self, event: UpdateUnion):

        chat = await event.fetch_chat()
        if chat is None:
            return False

        return chat.title == "Test"


@dp.message_created(CommandStart(), FilterChat())
async def custom_data(event: MessageCreated):
    await event.message.answer("Привет!")


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Фильтр callback payload

Пример использования типизированных payload для callback-кнопок:

import asyncio
import logging

from maxapi import Bot, Dispatcher, F
from maxapi.filters.callback_payload import CallbackPayload
from maxapi.filters.command import CommandStart
from maxapi.types import (
    CallbackButton,
    MessageCreated,
    MessageCallback,
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


class MyPayload(CallbackPayload, prefix="mypayload"):
    foo: str
    action: str


class AnotherPayload(CallbackPayload, prefix="another"):
    bar: str
    value: int


@dp.message_created(CommandStart())
async def show_keyboard(event: MessageCreated):
    kb = InlineKeyboardBuilder()
    kb.row(
        CallbackButton(
            text="Первая кнопка",
            payload=MyPayload(foo="123", action="edit").pack(),
        ),
        CallbackButton(
            text="Вторая кнопка",
            payload=AnotherPayload(bar="abc", value=42).pack(),
        ),
    )
    await event.message.answer("Нажми кнопку!", attachments=[kb.as_markup()])


@dp.message_callback(MyPayload.filter(F.foo == "123"))
async def on_first_callback(event: MessageCallback, payload: MyPayload):
    await event.answer(
        new_text=f"Первая кнопка: foo={payload.foo}, action={payload.action}"
    )


@dp.message_callback(AnotherPayload.filter())
async def on_second_callback(event: MessageCallback, payload: AnotherPayload):
    await event.answer(
        new_text=f"Вторая кнопка: bar={payload.bar}, value={payload.value}"
    )


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Middleware в обработчиках

Пример использования middleware на уровне обработчиков:

import asyncio
import logging

from typing import Any, Awaitable, Callable, Dict

from maxapi import Bot, Dispatcher
from maxapi.filters.middleware import BaseMiddleware
from maxapi.types import MessageCreated, Command, UpdateUnion

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


class CheckChatTitleMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
        event_object: UpdateUnion,
        data: Dict[str, Any],
    ) -> Any:

        if event_object.chat.title == "MAXApi":
            return await handler(event_object, data)


class CustomDataMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
        event_object: UpdateUnion,
        data: Dict[str, Any],
    ) -> Any:

        data["custom_data"] = (
            f"Это ID того кто вызвал команду: {event_object.from_user.user_id}"
        )

        await handler(event_object, data)


@dp.message_created(Command("start"), CheckChatTitleMiddleware())
async def start(event: MessageCreated):
    await event.message.answer(
        'Это сообщение было отправлено, так как ваш чат называется "MAXApi"!'
    )


@dp.message_created(Command("custom_data"), CustomDataMiddleware())
async def custom_data(event: MessageCreated, custom_data: str):
    await event.message.answer(custom_data)


@dp.message_created(
    Command("many_middlewares"),
    CheckChatTitleMiddleware(),
    CustomDataMiddleware(),
)
async def many_middlewares(event: MessageCreated, custom_data: str):
    await event.message.answer(
        'Это сообщение было отправлено, так как ваш чат называется "MAXApi"!'
    )
    await event.message.answer(custom_data)


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Middleware в роутерах

Пример использования middleware на уровне роутера:

import asyncio
import logging

from typing import Any, Awaitable, Callable, Dict

from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, Command, UpdateUnion
from maxapi.filters.middleware import BaseMiddleware

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


class CustomDataForRouterMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
        event_object: UpdateUnion,
        data: Dict[str, Any],
    ) -> Any:

        data["custom_data"] = (
            f"Это ID того кто вызвал команду: {event_object.from_user.user_id}"
        )
        result = await handler(event_object, data)
        return result


@dp.message_created(Command("custom_data"))
async def custom_data(event: MessageCreated, custom_data: str):
    await event.message.answer(custom_data)


async def main():
    dp.register_outer_middleware(CustomDataForRouterMiddleware())

    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Роутеры, InputMedia и контекст

Пример использования роутеров, InputMedia и работы с контекстом и состояниями:

import asyncio
import logging

from maxapi import Bot, Dispatcher, F
from maxapi.context import MemoryContext, State, StatesGroup
from maxapi.types import (
    BotStarted,
    Command,
    MessageCreated,
    CallbackButton,
    MessageCallback,
    BotCommand,
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder

from router import router

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()

dp.include_routers(router)


start_text = """Пример чат-бота для MAX 💙

Мои команды:

/clear очищает ваш контекст
/state или /context показывают ваше контекстное состояние
/data показывает вашу контекстную память
"""


class Form(StatesGroup):
    name = State()
    age = State()


@dp.on_started()
async def _():
    logging.info("Бот стартовал!")


@dp.bot_started()
async def bot_started(event: BotStarted):
    await event.bot.send_message(
        chat_id=event.chat_id,
        text="Привет! Отправь мне /start",
    )


@dp.message_created(Command("clear"))
async def hello(event: MessageCreated, context: MemoryContext):
    await context.clear()
    await event.message.answer(f"Ваш контекст был очищен!")


@dp.message_created(Command("data"))
async def hello(event: MessageCreated, context: MemoryContext):
    data = await context.get_data()
    await event.message.answer(f"Ваша контекстная память: {str(data)}")


@dp.message_created(Command("context"))
@dp.message_created(Command("state"))
async def hello(event: MessageCreated, context: MemoryContext):
    data = await context.get_state()
    await event.message.answer(f"Ваше контекстное состояние: {str(data)}")


@dp.message_created(Command("start"))
async def hello(event: MessageCreated):
    builder = InlineKeyboardBuilder()

    builder.row(
        CallbackButton(text="Ввести свое имя", payload="btn_1"),
        CallbackButton(text="Ввести свой возраст", payload="btn_2"),
    )
    builder.row(CallbackButton(text="Не хочу", payload="btn_3"))

    await event.message.answer(
        text=start_text,
        attachments=[
            builder.as_markup(),
        ],
    )


@dp.message_callback(F.callback.payload == "btn_1")
async def hello(event: MessageCallback, context: MemoryContext):
    await context.set_state(Form.name)
    await event.message.delete()
    await event.message.answer(f"Отправьте свое имя:")


@dp.message_callback(F.callback.payload == "btn_2")
async def hello(event: MessageCallback, context: MemoryContext):
    await context.set_state(Form.age)
    await event.message.delete()
    await event.message.answer(f"Отправьте ваш возраст:")


@dp.message_callback(F.callback.payload == "btn_3")
async def hello(event: MessageCallback, context: MemoryContext):
    await event.message.delete()
    await event.message.answer(f"Ну ладно 🥲")


@dp.message_created(F.message.body.text, Form.name)
async def hello(event: MessageCreated, context: MemoryContext):
    await context.update_data(name=event.message.body.text)

    data = await context.get_data()

    await event.message.answer(
        f"Приятно познакомиться, {data['name'].title()}!"
    )


@dp.message_created(F.message.body.text, Form.age)
async def hello(event: MessageCreated, context: MemoryContext):
    await context.update_data(age=event.message.body.text)

    await event.message.answer(f"Ого! А мне всего пару недель 😁")


async def main():
    await bot.set_commands(
        BotCommand(
            name="/state", description="Показывают ваше контекстное состояние"
        ),
        BotCommand(
            name="/data", description="Показывает вашу контекстную память"
        ),
        BotCommand(
            name="/context",
            description="Показывают ваше контекстное состояние",
        ),
    )
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

И соответствующий роутер (router.py):

from maxapi import F, Router
from maxapi.types import Command, MessageCreated
from maxapi.types import InputMedia

router = Router()
file = __file__.split("\\")[-1]


@router.message_created(Command("router"))
async def hello(obj: MessageCreated):
    await obj.message.answer(f"Пишу тебе из роута {file}")


# новая команда для примера, /media,
# пример использования: /media image.png (медиафайл берется указанному пути)
@router.message_created(Command("media"))
async def hello(event: MessageCreated):
    await event.message.answer(
        attachments=[
            InputMedia(path=event.message.body.text.replace("/media ", ""))
        ]
    )

Отправка файлов

Два основных способа отправить файл:

1. Передать InputMedia напрямую в attachments.
2. Предварительно загрузить файл через bot.upload_media(...), а затем отправить полученный объект вложения.

import asyncio

from maxapi import Bot
from maxapi.types import InputMedia

bot = Bot()


async def main():
    # Вариант 1: отправка напрямую через InputMedia
    await bot.send_message(
        chat_id=...,
        attachments=[
            InputMedia(path="logo.png"),
        ],
    )

    # Вариант 2: ручная загрузка + отправка attachment. (Подходит для рассылок)
    media = InputMedia("logo.png")
    attachment = await bot.upload_media(media)
    await bot.send_message(
        chat_id=...,
        attachments=[attachment],
    )


if __name__ == "__main__":
    asyncio.run(main())

Ограничения на загружаемые файлы

Сервер MAX ограничивает загружаемые файлы по типу (см. POST /uploads):

Тип	Форматы	Лимиты
image	JPG, JPEG, PNG, GIF, TIFF, BMP, HEIC	до 50 МБ и не более 7680×7680 px
video	MP4, MOV, MKV, WEBM	до 250 МБ
audio	MP3, WAV, M4A и др.	до 256 МБ и не более 60 мин
file	TXT, DOC, PDF и распространённые форматы	до 4 ГБ

Для image и audio оба условия (размер и вторая величина) проверяются одновременно — должны выполняться оба. Актуальные значения — в Upload limits.

Библиотека проверяет размер файла при любой загрузке (bot.upload_media, attachments, ручной bot.upload_file) и лишь предупреждает в логе (bot) при его превышении, не выбрасывая исключение — окончательную проверку выполняет сервер MAX. Файлы читаются в память целиком, поэтому для больших файлов file практический потолок ниже 4 ГБ и зависит от доступной ОЗУ.

Webhook

Высокоуровневый подход

Простой способ работы с webhook через встроенный метод (использует aiohttp, входит в базовый пакет):

import asyncio
import logging

from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


@dp.message_created()
async def handle_message(event: MessageCreated):
    await event.message.answer("Бот работает через вебхук!")


async def main():
    webhook_url = "https://ваш-домен.рф/webhook"  # <-- укажите свой
    webhook_secret = "my-secret-token"  # <-- укажите свой (5–256 символов)

    # Регистрируем вебхук на стороне MAX — платформа будет отправлять
    # заголовок X-Max-Bot-Api-Secret с каждым запросом.
    await bot.subscribe_webhook(url=webhook_url, secret=webhook_secret)

    # Фреймворк автоматически проверяет X-Max-Bot-Api-Secret в каждом
    # входящем запросе и возвращает 403, если заголовок отсутствует
    # или не совпадает (защита от посторонних запросов).
    await dp.handle_webhook(
        bot=bot,
        host="0.0.0.0",
        port=8080,
        secret=webhook_secret,
    )


if __name__ == "__main__":
    asyncio.run(main())

Низкоуровневый подход

Более гибкий способ: получаем FastAPI-приложение, регистрируем собственные маршруты (например, healthcheck) и отдельно подключаем MAX webhook-модуль.

Требует опциональных зависимостей:

pip install maxapi[fastapi]

import asyncio
import logging

import uvicorn
from fastapi import FastAPI

from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated
from maxapi.webhook.fastapi import FastAPIMaxWebhook

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()


@dp.message_created()
async def handle_message(event: MessageCreated):
    await event.message.answer("Бот работает через вебхук!")


async def main():
    webhook_url = "https://ваш-домен.рф/webhook"  # <-- укажите свой
    webhook_secret = "my-secret-token"  # <-- укажите свой (5–256 символов)

    # Передаём secret в конструктор — он сохраняется в webhook.secret.
    # Фреймворк будет автоматически проверять заголовок X-Max-Bot-Api-Secret
    # в каждом входящем POST-запросе и возвращать 403 при несоответствии.
    webhook = FastAPIMaxWebhook(dp=dp, bot=bot, secret=webhook_secret)

    # Создаём FastAPI-приложение с lifespan-инициализацией диспетчера
    app = FastAPI(lifespan=webhook.lifespan)

    # Собственные маршруты — например, healthcheck
    @app.get("/health")
    async def health():
        return {"status": "ok"}

    # Подключаем MAX webhook-обработчик к нашему приложению
    webhook.setup(app, path="/webhook")

    # Подписываемся на webhook — передаём тот же secret,
    # чтобы платформа MAX добавляла X-Max-Bot-Api-Secret в каждый запрос.
    await bot.subscribe_webhook(url=webhook_url, secret=webhook_secret)

    # Запускаем сервер uvicorn
    config = uvicorn.Config(app=app, host="0.0.0.0", port=8080)
    server = uvicorn.Server(config)
    await server.serve()


if __name__ == "__main__":
    asyncio.run(main())

Настройка прокси

Подключение через прокси

Для использования прокси-сервера передайте параметр proxy в DefaultConnectionProperties:

import asyncio
from maxapi import Bot, Dispatcher
from maxapi.client import DefaultConnectionProperties
from maxapi.types import MessageCreated, Command

# URL прокси в формате: http://login:password@ip:port
proxy_url = "http://login:password@ip:port"

# Создание настроек соединения с прокси
connection_props = DefaultConnectionProperties(proxy=proxy_url)

# Инициализация бота с настройками соединения
bot = Bot(default_connection=connection_props)
dp = Dispatcher()


@dp.message_created(Command("start"))
async def start_handler(event: MessageCreated):
    await event.message.answer("Привет!")


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Использование прокси из переменных окружения

Для использования прокси из переменных окружения используйте параметр trust_env=True:

import asyncio
from maxapi import Bot, Dispatcher
from maxapi.client import DefaultConnectionProperties
from maxapi.types import MessageCreated, Command

bot = Bot(
    "YOUR-TOKEN",
    default_connection=DefaultConnectionProperties(trust_env=True),
)
dp = Dispatcher()


@dp.message_created(Command("start"))
async def start_handler(event: MessageCreated):
    await event.message.answer("Привет!")


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Что такое trust_env?

Параметр trust_env=True в aiohttp позволяет автоматически читать настройки прокси из переменных окружения системы. Когда этот параметр включен, aiohttp будет искать следующие переменные окружения:

• HTTP_PROXY — прокси для HTTP-запросов (например, http://proxy.example.com:8080)
• HTTPS_PROXY — прокси для HTTPS-запросов (например, https://proxy.example.com:8080)
• NO_PROXY — список доменов, для которых прокси не используется (например, localhost,127.0.0.1,*.local)

Важно: Если переменные окружения не установлены, trust_env=True не вызовет ошибку — просто прокси использоваться не будет.