# 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="Bold and italic 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 ")
# 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 ")
@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="Bold and italic with link",
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.