MaxAPI (love-apples/maxapi)

MaxAPI

https://github.com/love-apples/maxapi
Admin
MaxAPI is a Python library for developing chatbots for the MAX platform, supporting both polling and...

Tokens:34,451
Snippets:199
Trust Score:6.2
Update:1 week ago
Show doc for...
Context Summary (auto-generated)
Raw
# MaxAPI Documentation

MaxAPI is a modern asynchronous Python library for developing chatbots using the MAX messenger API. It provides a convenient, type-safe interface for working with the MAX API, supporting asynchronous operations, a flexible filter system, middleware, routers, finite state machines (FSM), and many other features for creating powerful chatbots.

The library follows patterns similar to aiogram (Telegram bot framework) and provides decorators for event handling, inline keyboards, callback processing, context management for user state, and multiple webhook integrations (aiohttp, FastAPI, Litestar). It requires Python 3.9+ and is fully type-annotated for IDE support.

## Installation

Install the stable version from PyPI with pip.

```bash
pip install maxapi

# Optional: FastAPI webhook support
pip install maxapi[fastapi]

# Optional: Litestar webhook support
pip install maxapi[litestar]
```

## Bot Initialization

The Bot class is the main interface for interacting with the MAX API. It handles authentication, session management, and provides methods for all API operations.

```python
import os
from maxapi import Bot
from maxapi.client import DefaultConnectionProperties

# Basic initialization (reads token from MAX_BOT_TOKEN env var)
bot = Bot()

# Explicit token
bot = Bot(token="your-bot-token-here")

# With custom settings
bot = Bot(
    token="your-bot-token-here",
    notify=False,  # Disable notifications by default
    disable_link_preview=True,  # Disable link previews
    auto_requests=True,  # Auto-fetch chat/user data
)

# With proxy configuration
proxy_url = "http://login:password@proxy-ip:port"
connection_props = DefaultConnectionProperties(proxy=proxy_url)
bot = Bot(default_connection=connection_props)

# Using environment proxy settings
bot = Bot(
    default_connection=DefaultConnectionProperties(trust_env=True)
)
```

## Dispatcher and Polling

The Dispatcher handles event routing, middleware application, and runs the bot in polling or webhook mode. It manages event handlers and provides decorators for registering them.

```python
import asyncio
import logging
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, Command, BotStarted

logging.basicConfig(level=logging.INFO)

bot = Bot()
dp = Dispatcher()

# Handle bot start button press
@dp.bot_started()
async def on_bot_started(event: BotStarted):
    await event.bot.send_message(
        chat_id=event.chat_id,
        text="Hello! Send me /start"
    )

# Handle /start command
@dp.message_created(Command('start'))
async def start_handler(event: MessageCreated):
    await event.message.answer("Welcome to the bot!")

# Handle all text messages
@dp.message_created()
async def echo_handler(event: MessageCreated):
    if event.message.body and event.message.body.text:
        await event.message.answer(f"You said: {event.message.body.text}")

# Startup callback
@dp.on_started()
async def on_startup():
    logging.info("Bot has started!")

async def main():
    # Skip old updates on restart
    await dp.start_polling(bot, skip_updates=True)

if __name__ == '__main__':
    asyncio.run(main())
```

## Sending Messages

The Bot class provides methods for sending, editing, and deleting messages. Messages can include text, attachments, and inline keyboards.

```python
import asyncio
from maxapi import Bot
from maxapi.types import InputMedia
from maxapi.enums.format import Format

bot = Bot()

async def send_examples():
    chat_id = 123456789

    # Send text message
    result = await bot.send_message(
        chat_id=chat_id,
        text="Hello, world!"
    )
    print(f"Sent message ID: {result.message.body.mid}")

    # Send with HTML formatting
    await bot.send_message(
        chat_id=chat_id,
        text="<b>Bold</b> and <i>italic</i> text",
        format=Format.HTML
    )

    # Send with Markdown formatting
    await bot.send_message(
        chat_id=chat_id,
        text="**Bold** and *italic* text",
        format=Format.MARKDOWN
    )

    # Send to user directly
    await bot.send_message(
        user_id=987654321,
        text="Direct message to user"
    )

    # Send with image attachment
    await bot.send_message(
        chat_id=chat_id,
        text="Check out this image!",
        attachments=[InputMedia(path="logo.png")]
    )

    # Edit message
    await bot.edit_message(
        message_id=result.message.body.mid,
        text="Updated message text"
    )

    # Delete message
    await bot.delete_message(message_id=result.message.body.mid)

asyncio.run(send_examples())
```

## Event Types and Handlers

MaxAPI supports various event types through dedicated decorators. Each event type provides access to relevant data and helper methods.

```python
import asyncio
from maxapi import Bot, Dispatcher
from maxapi.types import (
    MessageCreated, MessageEdited, MessageRemoved,
    MessageCallback, BotAdded, BotRemoved, BotStarted, BotStopped,
    UserAdded, UserRemoved, ChatTitleChanged,
    DialogCleared, DialogMuted, DialogUnmuted
)

bot = Bot()
dp = Dispatcher()

@dp.message_created()
async def on_message(event: MessageCreated):
    # Access message data
    text = event.message.body.text if event.message.body else None
    sender = event.message.sender
    chat_id = event.message.recipient.chat_id
    await event.message.answer(f"Received: {text}")

@dp.message_edited()
async def on_edit(event: MessageEdited):
    await event.message.answer("You edited a message!")

@dp.message_removed()
async def on_delete(event: MessageRemoved):
    await bot.send_message(chat_id=event.chat_id, text="I saw that deletion!")

@dp.bot_added()
async def on_bot_added(event: BotAdded):
    chat = await event.fetch_chat()
    if chat:
        await bot.send_message(chat_id=event.chat_id, text=f"Hello, {chat.title}!")

@dp.user_added()
async def on_user_joined(event: UserAdded):
    await bot.send_message(
        chat_id=event.chat_id,
        text=f"Welcome, {event.user.first_name}!"
    )

@dp.user_removed()
async def on_user_left(event: UserRemoved):
    await bot.send_message(chat_id=event.chat_id, text="Goodbye!")

@dp.chat_title_changed()
async def on_title_change(event: ChatTitleChanged):
    await bot.send_message(
        chat_id=event.chat_id,
        text=f"New chat title: {event.title}"
    )

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

asyncio.run(main())
```

## MagicFilter (F)

The F object provides a powerful way to filter events based on their attributes using a fluent API. It supports attribute access, comparisons, string operations, and regex matching.

```python
import asyncio
from maxapi import Bot, Dispatcher, F
from maxapi.types import MessageCreated

bot = Bot()
dp = Dispatcher()

# Filter by exact text match
@dp.message_created(F.message.body.text == "hello")
async def on_hello(event: MessageCreated):
    await event.message.answer("Hello to you too!")

# Filter by text containing substring (case-insensitive)
@dp.message_created(F.message.body.text.lower().contains("help"))
async def on_help(event: MessageCreated):
    await event.message.answer("How can I help you?")

# Filter by regex pattern
@dp.message_created(F.message.body.text.regexp(r"^\d{4}$"))
async def on_four_digits(event: MessageCreated):
    await event.message.answer("You sent a 4-digit code!")

# Filter by text length
@dp.message_created(F.message.body.text.len() > 100)
async def on_long_message(event: MessageCreated):
    await event.message.answer("That's a long message!")

# Filter messages with attachments
@dp.message_created(F.message.body.attachments)
async def on_attachment(event: MessageCreated):
    await event.message.answer("You sent an attachment!")

# Filter forwarded messages
@dp.message_created(F.message.link.type == "forward")
async def on_forward(event: MessageCreated):
    await event.message.answer("That's a forwarded message!")

# Combine filters with & (and) and | (or)
@dp.message_created(
    (F.message.body.text.len() > 5) & (F.message.body.text.len() < 50)
)
async def on_medium_message(event: MessageCreated):
    await event.message.answer("Medium-length message received!")

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

asyncio.run(main())
```

## Command Filter

The Command filter matches messages that start with a command prefix (default: /). It supports multiple commands, case sensitivity, and bot username mentions.

```python
import asyncio
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, Command, CommandStart

bot = Bot()
dp = Dispatcher()

# Simple command
@dp.message_created(Command("start"))
async def start_cmd(event: MessageCreated):
    await event.message.answer("Bot started!")

# CommandStart shortcut for /start
@dp.message_created(CommandStart())
async def start_shortcut(event: MessageCreated):
    await event.message.answer("Welcome!")

# Multiple commands
@dp.message_created(Command(["help", "h", "?"]))
async def help_cmd(event: MessageCreated):
    await event.message.answer("Available commands: /start, /help")

# Command with arguments - args passed to handler
@dp.message_created(Command("echo"))
async def echo_cmd(event: MessageCreated, args: list[str]):
    # /echo hello world -> args = ["hello", "world"]
    if args:
        await event.message.answer(" ".join(args))
    else:
        await event.message.answer("Usage: /echo <text>")

# Custom prefix
@dp.message_created(Command("ping", prefix="!"))
async def ping_cmd(event: MessageCreated):
    # Matches !ping
    await event.message.answer("Pong!")

# Case-sensitive command
@dp.message_created(Command("CaseSensitive", check_case=True))
async def case_cmd(event: MessageCreated):
    await event.message.answer("Exact case matched!")

# Require bot username mention (@botname /command)
@dp.message_created(Command("private", only_with_bot_username=True))
async def private_cmd(event: MessageCreated):
    await event.message.answer("Command with @mention received!")

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

asyncio.run(main())
```

## Inline Keyboards and Callback Buttons

Create interactive keyboards with various button types. The InlineKeyboardBuilder provides a fluent API for constructing keyboards.

```python
import asyncio
from maxapi import Bot, Dispatcher, F
from maxapi.types import (
    MessageCreated, MessageCallback, Command,
    CallbackButton, LinkButton, RequestContactButton,
    RequestGeoLocationButton, MessageButton
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder

bot = Bot()
dp = Dispatcher()

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

    # Add buttons in rows
    builder.row(
        CallbackButton(text="Option 1", payload="opt_1"),
        CallbackButton(text="Option 2", payload="opt_2")
    )
    builder.row(
        CallbackButton(text="Option 3", payload="opt_3")
    )
    builder.row(
        LinkButton(text="Documentation", url="https://dev.max.ru/docs")
    )

    await event.message.answer(
        text="Choose an option:",
        attachments=[builder.as_markup()]
    )

@dp.message_created(Command("contacts"))
async def show_contacts(event: MessageCreated):
    builder = InlineKeyboardBuilder()
    builder.row(
        RequestContactButton(text="Share Contact"),
        RequestGeoLocationButton(text="Share Location")
    )
    builder.row(
        MessageButton(text="Send Message")
    )

    await event.message.answer(
        text="Share your info:",
        attachments=[builder.as_markup()]
    )

# Handle callback button presses
@dp.message_callback(F.callback.payload == "opt_1")
async def on_option_1(event: MessageCallback):
    await event.answer(
        notification="You selected Option 1!",
        new_text="Option 1 selected"
    )

@dp.message_callback(F.callback.payload == "opt_2")
async def on_option_2(event: MessageCallback):
    # Edit the message
    await event.edit(text="You chose Option 2!")

@dp.message_callback(F.callback.payload == "opt_3")
async def on_option_3(event: MessageCallback):
    # Just acknowledge without changing message
    await event.ack(notification="Option 3 acknowledged!")

# Catch-all callback handler
@dp.message_callback()
async def on_any_callback(event: MessageCallback):
    payload = event.callback.payload
    await event.answer(notification=f"Received: {payload}")

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

asyncio.run(main())
```

## Typed Callback Payloads

Define structured callback payloads using Pydantic models for type-safe callback handling with automatic parsing and filtering.

```python
import asyncio
from maxapi import Bot, Dispatcher, F
from maxapi.types import MessageCreated, MessageCallback, Command, CallbackButton
from maxapi.filters.callback_payload import CallbackPayload
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder

bot = Bot()
dp = Dispatcher()

# Define typed payload classes with prefix
class ProductPayload(CallbackPayload, prefix="product"):
    product_id: int
    action: str  # "view", "buy", "favorite"

class PagePayload(CallbackPayload, prefix="page"):
    page_num: int
    category: str

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

    # Create buttons with typed payloads
    builder.row(
        CallbackButton(
            text="View Product #1",
            payload=ProductPayload(product_id=1, action="view").pack()
        ),
        CallbackButton(
            text="Buy Product #1",
            payload=ProductPayload(product_id=1, action="buy").pack()
        )
    )
    builder.row(
        CallbackButton(
            text="Next Page",
            payload=PagePayload(page_num=2, category="electronics").pack()
        )
    )

    await event.message.answer(
        text="Welcome to the shop!",
        attachments=[builder.as_markup()]
    )

# Filter by payload type and fields
@dp.message_callback(ProductPayload.filter(F.action == "view"))
async def on_view_product(event: MessageCallback, payload: ProductPayload):
    await event.answer(
        new_text=f"Viewing product #{payload.product_id}"
    )

@dp.message_callback(ProductPayload.filter(F.action == "buy"))
async def on_buy_product(event: MessageCallback, payload: ProductPayload):
    await event.answer(
        notification=f"Added product #{payload.product_id} to cart!"
    )

# Filter by payload type only
@dp.message_callback(PagePayload.filter())
async def on_page_change(event: MessageCallback, payload: PagePayload):
    await event.answer(
        new_text=f"Page {payload.page_num} of {payload.category}"
    )

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

asyncio.run(main())
```

## Custom Filters

Create custom filters by extending BaseFilter for complex filtering logic that can't be expressed with MagicFilter.

```python
import asyncio
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, Command, UpdateUnion
from maxapi.filters import BaseFilter

bot = Bot()
dp = Dispatcher()

# Custom filter checking chat title
class ChatTitleFilter(BaseFilter):
    def __init__(self, allowed_titles: list[str]):
        self.allowed_titles = allowed_titles

    async def __call__(self, event: UpdateUnion) -> bool:
        chat = await event.fetch_chat()
        if chat is None:
            return False
        return chat.title in self.allowed_titles

# Custom filter checking user permissions
class AdminOnlyFilter(BaseFilter):
    async def __call__(self, event: UpdateUnion) -> dict | bool:
        from_user = await event.fetch_from_user()
        if from_user is None:
            return False

        # Return dict to pass data to handler
        if from_user.user_id in [123456, 789012]:  # Admin IDs
            return {"is_admin": True, "user": from_user}
        return False

# Custom filter with data injection
class MessageLengthFilter(BaseFilter):
    def __init__(self, min_length: int = 0, max_length: int = 1000):
        self.min_length = min_length
        self.max_length = max_length

    async def __call__(self, event: UpdateUnion) -> dict | bool:
        if not isinstance(event, MessageCreated):
            return False

        text = event.message.body.text if event.message.body else ""
        length = len(text) if text else 0

        if self.min_length <= length <= self.max_length:
            return {"text_length": length}
        return False

# Use custom filters
@dp.message_created(Command("admin"), AdminOnlyFilter())
async def admin_command(event: MessageCreated, is_admin: bool, user):
    await event.message.answer(f"Hello admin {user.first_name}!")

@dp.message_created(
    ChatTitleFilter(["Support", "Help Desk"]),
    MessageLengthFilter(min_length=10)
)
async def support_message(event: MessageCreated, text_length: int):
    await event.message.answer(
        f"Support ticket received ({text_length} chars)"
    )

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

asyncio.run(main())
```

## Middleware

Middleware allows pre/post processing of events. Apply middleware globally to the dispatcher, to routers, or to individual handlers.

```python
import asyncio
import time
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

bot = Bot()
dp = Dispatcher()

# Logging middleware
class LoggingMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
        event: UpdateUnion,
        data: Dict[str, Any],
    ) -> Any:
        start = time.time()
        print(f"Processing event: {event.update_type}")

        result = await handler(event, data)

        elapsed = time.time() - start
        print(f"Event processed in {elapsed:.3f}s")
        return result

# Data injection middleware
class UserDataMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
        event: UpdateUnion,
        data: Dict[str, Any],
    ) -> Any:
        # Inject custom data for handlers
        from_user = await event.fetch_from_user()
        data["user_name"] = from_user.first_name if from_user else "Unknown"
        data["timestamp"] = time.time()

        return await handler(event, data)

# Access control middleware
class ChatFilterMiddleware(BaseMiddleware):
    def __init__(self, allowed_chat_ids: list[int]):
        self.allowed_chat_ids = allowed_chat_ids

    async def __call__(
        self,
        handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
        event: UpdateUnion,
        data: Dict[str, Any],
    ) -> Any:
        chat = await event.fetch_chat()
        if chat and chat.chat_id in self.allowed_chat_ids:
            return await handler(event, data)
        # Don't call handler - silently ignore

# Register global middleware
dp.middleware(LoggingMiddleware())
dp.middleware(UserDataMiddleware())

# Handler with injected data from middleware
@dp.message_created(Command("whoami"))
async def whoami(event: MessageCreated, user_name: str, timestamp: float):
    await event.message.answer(f"Hello {user_name}! (at {timestamp})")

# Handler-level middleware
@dp.message_created(
    Command("restricted"),
    ChatFilterMiddleware(allowed_chat_ids=[123456789])
)
async def restricted_cmd(event: MessageCreated):
    await event.message.answer("You have access!")

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

asyncio.run(main())
```

## Routers

Routers allow organizing handlers into separate modules with their own middleware and filters. They can be nested and included in the main dispatcher.

```python
# main.py
import asyncio
from maxapi import Bot, Dispatcher
from maxapi.types import Command, MessageCreated
from admin_router import admin_router
from user_router import user_router

bot = Bot()
dp = Dispatcher()

# Include routers
dp.include_routers(admin_router, user_router)

@dp.message_created(Command("start"))
async def start(event: MessageCreated):
    await event.message.answer("Welcome! Use /help for commands.")

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

asyncio.run(main())
```

```python
# admin_router.py
from maxapi import Router
from maxapi.types import Command, MessageCreated
from maxapi.filters.middleware import BaseMiddleware

admin_router = Router(router_id="admin")

class AdminCheckMiddleware(BaseMiddleware):
    async def __call__(self, handler, event, data):
        from_user = await event.fetch_from_user()
        if from_user and from_user.user_id in [123456]:  # Admin IDs
            return await handler(event, data)

# Apply middleware to entire router
admin_router.middleware(AdminCheckMiddleware())

@admin_router.message_created(Command("ban"))
async def ban_user(event: MessageCreated, args: list[str]):
    if args:
        await event.message.answer(f"User {args[0]} banned!")
    else:
        await event.message.answer("Usage: /ban <user_id>")

@admin_router.message_created(Command("stats"))
async def show_stats(event: MessageCreated):
    await event.message.answer("Admin stats: ...")
```

```python
# user_router.py
from maxapi import Router, F
from maxapi.types import Command, MessageCreated

user_router = Router(router_id="user")

@user_router.message_created(Command("profile"))
async def show_profile(event: MessageCreated):
    from_user = await event.fetch_from_user()
    if from_user:
        await event.message.answer(
            f"Name: {from_user.first_name}\n"
            f"ID: {from_user.user_id}"
        )

@user_router.message_created(Command("settings"))
async def show_settings(event: MessageCreated):
    await event.message.answer("User settings...")
```

## Finite State Machine (FSM)

The context system provides FSM capabilities for multi-step conversations. Define states using StatesGroup and State classes.

```python
import asyncio
from maxapi import Bot, Dispatcher, F
from maxapi.types import MessageCreated, Command, CallbackButton, MessageCallback
from maxapi.context import MemoryContext, State, StatesGroup
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder

bot = Bot()
dp = Dispatcher()

# Define states
class RegistrationForm(StatesGroup):
    waiting_name = State()
    waiting_age = State()
    waiting_confirm = State()

@dp.message_created(Command("register"))
async def start_registration(event: MessageCreated, context: MemoryContext):
    await context.set_state(RegistrationForm.waiting_name)
    await event.message.answer("Let's register! What's your name?")

@dp.message_created(F.message.body.text, RegistrationForm.waiting_name)
async def process_name(event: MessageCreated, context: MemoryContext):
    name = event.message.body.text
    await context.update_data(name=name)
    await context.set_state(RegistrationForm.waiting_age)
    await event.message.answer(f"Nice to meet you, {name}! How old are you?")

@dp.message_created(F.message.body.text, RegistrationForm.waiting_age)
async def process_age(event: MessageCreated, context: MemoryContext):
    text = event.message.body.text
    if not text.isdigit():
        await event.message.answer("Please enter a valid number.")
        return

    age = int(text)
    await context.update_data(age=age)
    await context.set_state(RegistrationForm.waiting_confirm)

    data = await context.get_data()

    builder = InlineKeyboardBuilder()
    builder.row(
        CallbackButton(text="Confirm", payload="confirm"),
        CallbackButton(text="Cancel", payload="cancel")
    )

    await event.message.answer(
        f"Please confirm:\nName: {data['name']}\nAge: {data['age']}",
        attachments=[builder.as_markup()]
    )

@dp.message_callback(
    F.callback.payload == "confirm",
    RegistrationForm.waiting_confirm
)
async def confirm_registration(event: MessageCallback, context: MemoryContext):
    data = await context.get_data()
    await context.clear()  # Clear state and data
    await event.answer(
        new_text=f"Registration complete!\nWelcome, {data['name']}!"
    )

@dp.message_callback(
    F.callback.payload == "cancel",
    RegistrationForm.waiting_confirm
)
async def cancel_registration(event: MessageCallback, context: MemoryContext):
    await context.clear()
    await event.answer(new_text="Registration cancelled.")

# Utility commands
@dp.message_created(Command("cancel"))
async def cancel_any(event: MessageCreated, context: MemoryContext):
    state = await context.get_state()
    if state:
        await context.clear()
        await event.message.answer("Operation cancelled.")
    else:
        await event.message.answer("Nothing to cancel.")

@dp.message_created(Command("state"))
async def show_state(event: MessageCreated, context: MemoryContext):
    state = await context.get_state()
    data = await context.get_data()
    await event.message.answer(f"State: {state}\nData: {data}")

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

asyncio.run(main())
```

## Redis Context Storage

For production applications, use Redis for persistent state storage across bot restarts.

```python
import asyncio
from redis.asyncio import Redis
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, Command
from maxapi.context import RedisContext, State, StatesGroup

bot = Bot()

# Create Redis client
redis_client = Redis(host="localhost", port=6379, db=0)

# Initialize dispatcher with Redis storage
dp = Dispatcher(
    storage=RedisContext,
    redis_client=redis_client,
    key_prefix="mybot"  # Optional prefix for Redis keys
)

class UserState(StatesGroup):
    active = State()

@dp.message_created(Command("start"))
async def start(event: MessageCreated, context: RedisContext):
    await context.set_state(UserState.active)
    await context.update_data(visits=1)
    await event.message.answer("State saved to Redis!")

@dp.message_created(Command("visits"))
async def show_visits(event: MessageCreated, context: RedisContext):
    data = await context.get_data()
    visits = data.get("visits", 0)
    await context.update_data(visits=visits + 1)
    await event.message.answer(f"Total visits: {visits + 1}")

async def main():
    try:
        await dp.start_polling(bot)
    finally:
        await redis_client.close()

asyncio.run(main())
```

## File Uploads and Downloads

Send files using InputMedia or pre-upload for bulk messaging. Download files from message attachments.

```python
import asyncio
from pathlib import Path
from maxapi import Bot
from maxapi.types import InputMedia, InputMediaBuffer

bot = Bot()

async def file_operations():
    chat_id = 123456789

    # Send file from path
    await bot.send_message(
        chat_id=chat_id,
        text="Here's an image!",
        attachments=[InputMedia(path="photo.jpg")]
    )

    # Send file from bytes (useful for generated content)
    image_bytes = Path("photo.jpg").read_bytes()
    await bot.send_message(
        chat_id=chat_id,
        attachments=[
            InputMediaBuffer(data=image_bytes, filename="generated.jpg")
        ]
    )

    # Pre-upload file (efficient for bulk sending)
    media = InputMedia(path="document.pdf")
    attachment = await bot.upload_media(media)

    # Send to multiple chats without re-uploading
    for chat_id in [111, 222, 333]:
        await bot.send_message(
            chat_id=chat_id,
            text="Document for everyone!",
            attachments=[attachment]
        )

    # Download file from message attachment
    # Assuming you have a message with an attachment
    messages = await bot.get_messages(chat_id=chat_id, count=1)
    if messages.messages:
        msg = messages.messages[0]
        if msg.body and msg.body.attachments:
            for att in msg.body.attachments:
                if hasattr(att.payload, 'url'):
                    downloaded_path = await bot.download_file(
                        url=att.payload.url,
                        destination="./downloads/"
                    )
                    print(f"Downloaded to: {downloaded_path}")

asyncio.run(file_operations())
```

## Webhook Setup

Configure webhooks for production deployments. MaxAPI supports aiohttp (built-in), FastAPI, and Litestar.

```python
# High-level webhook (aiohttp built-in)
import asyncio
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated

bot = Bot()
dp = Dispatcher()

@dp.message_created()
async def handle_message(event: MessageCreated):
    await event.message.answer("Webhook bot is working!")

async def main():
    webhook_url = "https://your-domain.com/webhook"
    webhook_secret = "your-secret-token-here"  # 5-256 characters

    # Register webhook with MAX
    await bot.subscribe_webhook(url=webhook_url, secret=webhook_secret)

    # Start webhook server
    await dp.handle_webhook(
        bot=bot,
        host="0.0.0.0",
        port=8080,
        path="/webhook",
        secret=webhook_secret  # Validates X-Max-Bot-Api-Secret header
    )

asyncio.run(main())
```

```python
# Low-level webhook with FastAPI
# pip install maxapi[fastapi]
import asyncio
import uvicorn
from fastapi import FastAPI
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated
from maxapi.webhook.fastapi import FastAPIMaxWebhook

bot = Bot()
dp = Dispatcher()

@dp.message_created()
async def handle_message(event: MessageCreated):
    await event.message.answer("FastAPI webhook!")

async def main():
    webhook_url = "https://your-domain.com/webhook"
    webhook_secret = "your-secret-token"

    webhook = FastAPIMaxWebhook(dp=dp, bot=bot, secret=webhook_secret)
    app = FastAPI(lifespan=webhook.lifespan)

    # Custom routes
    @app.get("/health")
    async def health():
        return {"status": "ok"}

    webhook.setup(app, path="/webhook")
    await bot.subscribe_webhook(url=webhook_url, secret=webhook_secret)

    config = uvicorn.Config(app=app, host="0.0.0.0", port=8080)
    server = uvicorn.Server(config)
    await server.serve()

asyncio.run(main())
```

## Chat and User Management

Access chat information, manage members, and perform administrative actions.

```python
import asyncio
from maxapi import Bot

bot = Bot()

async def chat_management():
    chat_id = 123456789

    # Get bot info
    me = await bot.get_me()
    print(f"Bot: @{me.username} (ID: {me.user_id})")

    # Get chat by ID
    chat = await bot.get_chat_by_id(id=chat_id)
    print(f"Chat: {chat.title}")

    # Get chat by link
    chat = await bot.get_chat_by_link(link="https://max.ru/join/abc123")

    # List bot's chats
    chats = await bot.get_chats(count=50)
    for chat in chats.chats:
        print(f"- {chat.title} ({chat.chat_id})")

    # Get chat members
    members = await bot.get_chat_members(chat_id=chat_id, count=20)
    for member in members.members:
        print(f"Member: {member.user.first_name}")

    # Get specific member
    member = await bot.get_chat_member(chat_id=chat_id, user_id=987654321)

    # Add members to chat
    await bot.add_chat_members(chat_id=chat_id, user_ids=[111, 222, 333])

    # Kick member
    await bot.kick_chat_member(chat_id=chat_id, user_id=999, block=False)

    # Get chat admins
    admins = await bot.get_list_admin_chat(chat_id=chat_id)

    # Edit chat
    await bot.edit_chat(chat_id=chat_id, title="New Chat Title")

    # Pin/unpin messages
    await bot.pin_message(chat_id=chat_id, message_id="msg_123", notify=True)
    await bot.delete_pin_message(chat_id=chat_id)

    # Get pinned message
    pinned = await bot.get_pin_message(chat_id=chat_id)

    # Send typing indicator
    from maxapi.enums.sender_action import SenderAction
    await bot.send_action(chat_id=chat_id, action=SenderAction.TYPING_ON)

    # Leave chat
    await bot.delete_me_from_chat(chat_id=chat_id)

asyncio.run(chat_management())
```

## Text Formatting Helpers

Build formatted messages using HTML or Markdown helpers for consistent styling.

```python
import asyncio
from maxapi import Bot
from maxapi.enums.format import Format
from maxapi.utils.formatting import (
    Bold, Italic, Underline, Strikethrough,
    Code, Blockquote, Heading, Link,
    as_html, as_markdown
)

bot = Bot()

async def formatting_examples():
    chat_id = 123456789

    # HTML formatting
    html_text = as_html(
        Heading("Welcome!"),
        "\n\n",
        Bold("Important notice:"), " This is a ", Italic("test"), " message.\n",
        Underline("Underlined"), " and ", Strikethrough("strikethrough"), "\n",
        Code("inline_code()"), "\n",
        Blockquote("This is a quote"),
        "\n",
        Link("Documentation", url="https://dev.max.ru/docs")
    )

    await bot.send_message(
        chat_id=chat_id,
        text=html_text,
        format=Format.HTML
    )

    # Markdown formatting
    md_text = as_markdown(
        Heading("Features"),
        "\n",
        Bold("Feature 1"), " - Description\n",
        Bold("Feature 2"), " - Description\n",
        Blockquote("Note: Beta feature")
    )

    await bot.send_message(
        chat_id=chat_id,
        text=md_text,
        format=Format.MARKDOWN
    )

    # Direct HTML/Markdown strings also work
    await bot.send_message(
        chat_id=chat_id,
        text="<b>Bold</b> and <i>italic</i> with <a href='https://max.ru'>link</a>",
        format=Format.HTML
    )

asyncio.run(formatting_examples())
```

## Summary

MaxAPI provides a comprehensive toolkit for building MAX messenger bots with modern Python patterns. The library excels at handling complex conversational flows through its FSM context system, building interactive interfaces with inline keyboards and callback handlers, and organizing large bots with routers and middleware. Its async-first design ensures excellent performance for high-traffic bots.

For production deployments, MaxAPI supports multiple webhook frameworks (aiohttp, FastAPI, Litestar), Redis-backed state persistence, and proxy configurations. The MagicFilter system and custom filter classes enable precise event targeting, while the middleware system allows centralized logging, authentication, and data injection. The type-safe API with Pydantic models ensures IDE support and catches errors early in development.