# 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.