maxo 0.6.0

Асинхронный Python-фреймворк для разработки ботов мессенджера MAX (max.ru) с поддержкой long-polling, вебхуков, FSM, диалогов, фильтров и DI

maxo - асинхронный Python-фреймворк для ботов MAX (max.ru)

Асинхронный Python-фреймворк для разработки ботов в MAX

Документация

Интерфейс основан на aiogram
maxo/dialogs сделано из aiogram_dialog
maxo/transport/webhook сделано из aiogram-webhook

Установка

Через pip:

pip install maxo

В pyproject.toml:

[project]
dependencies = [
    "maxo",
]

Особенности

• Асинхронность на базе aiohttp и unihttp (asyncio, PEP 492)
• 100% покрытие типами, adaptix для валидации данных
• Роутеры, фильтры, милдвари
• Встроенная машина состояний (FSM) и диалоги поверх них
• Поддержка лонг-поллинга и вебхуков через aiohttp и fastapi
• Интеграции с dishka и magic_filter
• Автогенерация методов, типов и апдейтов по официальной документации

Быстрый старт

Больше примеров в примерах

Эхо-бот

from maxo import Bot, Dispatcher
from maxo.routing.updates import MessageCreated
from maxo.transport.long_polling import LongPolling

bot = Bot("TOKEN")
dp = Dispatcher()

@dp.message_created()
async def echo_handler(message: MessageCreated) -> None:
    text = message.text or "Текста нет"
    await message.answer(text)

LongPolling(dp).run(bot)

Команды

from maxo import Bot, Dispatcher
from maxo.routing.filters import Command, DeeplinkFilter
from maxo.routing.updates import BotStarted, MessageCreated
from maxo.transport.long_polling import LongPolling

bot = Bot("TOKEN")
dp = Dispatcher()

@dp.bot_started(DeeplinkFilter())
async def deeplink_handler(bot_started: BotStarted, deeplink: str) -> None:
    await bot_started.send_message(f"Привет! Я бот. Диплинк: {deeplink}")

@dp.bot_started()
async def start_handler(bot_started: BotStarted) -> None:
    await bot_started.send_message(f"Привет! Я бот. А ты {bot_started.user.fullname}")

@dp.message(Command("help"))
async def help_handler(message: MessageCreated) -> None:
    await message.send_message("За помощью обращайтесь в t.me/maxo_py")

LongPolling(dp).run(bot)

Клавиатуры

from magic_filter import F

from maxo import Bot, Dispatcher
from maxo.integrations.magic_filter import MagicFilter
from maxo.routing.filters import CommandStart
from maxo.routing.updates import MessageCallback, MessageCreated
from maxo.transport.long_polling import LongPolling
from maxo.utils.builders import KeyboardBuilder

bot = Bot("TOKEN")
dp = Dispatcher()

@dp.message_created(CommandStart())
async def start_handler(message: MessageCreated) -> None:
    maxo_url = "https://github.com/K1rL3s/maxo"
    keyboard = (
        KeyboardBuilder()
        .add_callback(text="Колбэк", payload="click_me")
        .add_message(text="Сообщение")
        .add_link(text="Перейти в maxo", url=maxo_url)
        .add_clipboard(text="Скопировать maxo", payload=maxo_url)
        .add_request_contact(text="Поделиться контактами")
        .add_request_geo_location(text="Поделиться гео позицией")
        .adjust(2, 2, 1, 1)
    )
    await message.answer(text="Кнопочки :3", keyboard=keyboard.build())

@dp.message_callback(MagicFilter(F.payload == "callback_payload"))
async def button_handler(callback: MessageCallback) -> None:
    await callback.callback_answer("Вы нажали на кнопку!")

LongPolling(dp).run(bot)

Вебхук

import logging

from aiohttp import web

from maxo import Bot, Dispatcher, Router
from maxo.enums import TextFormat
from maxo.routing.updates import BotStarted, MessageCreated
from maxo.routing.utils import collect_used_updates
from maxo.transport.webhook.adapters.aiohttp import AiohttpWebAdapter
from maxo.transport.webhook.engines import SimpleEngine, WebhookEngine
from maxo.transport.webhook.routing import StaticRouting
from maxo.transport.webhook.security import Security, StaticSecretToken

bot = Bot("TOKEN")
router = Router()

@router.bot_started()
async def start_handler(bot_started: BotStarted) -> None:
    await bot_started.send_message(
        text=f"Привет из вебхука, {bot_started.user.first_name}!",
    )

@router.message_created()
async def echo_handler(message: MessageCreated) -> None:
    await message.answer(
        text=message.message.body.html_text,
        format=TextFormat.HTML,
    )

@router.after_startup()
async def on_startup(dispatcher: Dispatcher, webhook_engine: WebhookEngine) -> None:
    await webhook_engine.set_webhook(update_types=collect_used_updates(dispatcher))

def main() -> None:
    dispatcher = Dispatcher()
    dispatcher.include(router)

    engine = SimpleEngine(
        dispatcher,
        bot,
        web_adapter=AiohttpWebAdapter(),
        routing=StaticRouting(url="https://example.com/webhook"),
        security=Security(secret_token=StaticSecretToken("pepa_pig")),
    )
    app = web.Application()
    engine.register(app)

    web.run_app(app, host="127.0.0.1", port=8080)

if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    main()

FAQ

Что такое MAX?

MAX - российский мессенджер. У него есть открытое Bot API, для работы с которым и создан maxo.

Чем maxo отличается от aiogram?

maxo - отдельный фреймворк именно для ботов MAX, но интерфейс намеренно близок к aiogram, чтобы переход был максимально безболезненным. Диалоги (maxo.dialogs) портированы из aiogram_dialog, вебхуки (maxo.transport.webhook) - из aiogram-webhook.

Поддерживает ли maxo вебхуки?

Да. Поддерживается и long-polling, и webhook через aiohttp или fastapi - см. примеры выше.

Какой Python нужен?

Python 3.12, 3.13 или 3.14.

Где взять токен бота MAX?

На платформе для партнёров.

Как добавить FSM?

FSM встроена в maxo - есть MemoryStorage из коробки и опциональное хранилище в Redis (maxo[redis]). Подробности - в документации.

Связь

Если у вас есть вопросы, вы можете задать их в Телеграме @maxo_py или Максе