AioScam — Documentation (EN)¶

Table of Contents¶

Installation¶

git clone https://github.com/alex-di-96/aioscam.git
cd aioscam
pip install -e .

pip install aioscam[fastapi]   # FastAPI webhook
pip install aioscam[litestar]  # Litestar webhook
pip install aioscam[dev]       # pytest, ruff, mypy

aioscam.webapp (WebApp/Mini App support) needs no extra install — it only uses aiohttp and pydantic, both already required by the base package. There is no aioscam[webapp] extra because there is nothing extra to install; a plain pip install aioscam already gives you validate_init_data, EventStreamManager, and WebAppMiddleware.

Requirements: Python 3.9–3.12, aiohttp>=3.9, aiofiles>=23.0, pydantic>=2.0, magic-filter>=1.0

Quick Start¶

import asyncio
from aioscam import Bot, Dispatcher, Router
from aioscam.filters import Command

dp = Dispatcher()
router = Router()

@router.message_created(Command("start"))
async def cmd_start(event):
    await event.answer("Hello! Send me anything.")

@router.message_created()
async def echo(event):
    await event.answer(event.text)

dp.include_router(router)

async def main():
    bot = Bot()  # token from MAX_BOT_TOKEN env var
    await dp.start_polling(bot)

asyncio.run(main())

Architecture¶

aioscam/
├── bot/          # Bot — all API methods
├── client/       # HTTP client (aiohttp + rate limiter + upload/download)
├── dispatcher/   # Dispatcher, Router, EventContext, StateGuard
├── enums/        # 15 enum files
├── exceptions/   # 12 exception classes
├── filters/      # Command, Text, StartCommand, StateFilter, ContentType, F
├── fsm/          # State, StatesGroup, MemoryStorage
├── handler/      # MessageHandler, CallbackHandler, EventHandler
├── i18n/         # I18n — JSON translations, locale detection
├── limiter/      # RateLimiter, RateLimitConfig
├── methods/      # BaseMethod, GetMe, SendMessage, GetUpdates
├── middleware/   # BaseMiddleware, MiddlewareManager
├── types/        # Pydantic models — User, Chat, Message, Attachment, etc.
├── utils/        # KeyboardBuilder, formatting, deep_linking, media, BotCapabilities
└── webapp/       # validate_init_data, validate_contact, EventStreamManager, WebAppMiddleware

Bot Methods¶

Messages¶

await bot.send_message(chat_id=123, user_id=456, text="Hello!", format="markdown")
await bot.edit_message(message_id="mid.abc", text="Updated text")
await bot.delete_message(message_id="mid.abc")
await bot.send_action(chat_id=123, action=SenderAction.TYPING_ON)

Chats & Users¶

me = await bot.get_me()
chat = await bot.get_chat_by_id(id=123)   # get_chats() is deprecated — use ChatRegistry
members = await bot.get_chat_members(chat_id=123)
await bot.add_chat_members(chat_id=123, user_ids=[456])
await bot.remove_member_chat(chat_id=123, user_id=456)

Bot Info¶

await bot.set_my_commands([BotCommand(name="start", description="Start the bot")])
await bot.set_bot_info(name="My Bot", description="Bot description")

Media¶

Sending¶

from aioscam import InputMedia, InputMediaBuffer, UploadType

# From file path — type auto-detected from extension
await bot.send_photo(chat_id=123, user_id=456, photo="photo.jpg", caption="Look!")
await bot.send_video(chat_id=123, user_id=456, video="video.mp4")
await bot.send_audio(chat_id=123, user_id=456, audio="song.mp3")
await bot.send_document(chat_id=123, user_id=456, document="report.pdf")
await bot.send_media(chat_id=123, user_id=456, media="any_file.ext")  # auto-type

# From bytes buffer (no temp file on disk)
data = open("photo.jpg", "rb").read()
await bot.send_photo(chat_id=123, user_id=456, photo=data)

Extension → UploadType mapping:

.jpg .jpeg .png .gif .webp .bmp → IMAGE
.mp4 .mov .avi .mkv .webm → VIDEO
.mp3 .ogg .wav .m4a .flac .aac .opus → AUDIO
everything else → FILE

Downloading¶

# In-memory (for processing, PIL, ffmpeg, etc.)
data = await bot.download_file_bytes(url, token)   # → bytes | None

# To disk with unique datetime-stamped filename
path = Bot.make_temp_path(".jpg")  # /tmp/aioscam_20260528_143022_847291.jpg
await bot.download_file(path, url, token)           # → HTTP status code

Receiving in handlers¶

@router.message_created(F.message.body.text == "")
async def handle_media(event):
    raw = event.data.get("raw_update", {})
    attachments = raw.get("message", {}).get("body", {}).get("attachments", [])
    for att in attachments:
        url   = att.get("payload", {}).get("url")
        token = att.get("payload", {}).get("token")
        if url and token:
            data = await event.bot.download_file_bytes(url, token)

Stickers¶

Bots cannot send stickers. On receive:

if att.get("type") == "sticker":
    code = att.get("payload", {}).get("code")
    url  = att.get("payload", {}).get("url")

Dispatcher & Router¶

dp = Dispatcher(
    state_guard_commands={'/cancel', '/start'},
    state_guard_callbacks={'action:cancel'},
)
child = Router(name="child")
dp.include_router(child)
# Backlog policy: what to do with updates accumulated while the bot was down
#   "skip"     — drop them (default; equals the legacy skip_updates=True)
#   "process"  — dispatch everything
#   "collapse" — dedupe repeats: 50 stale /start presses from one user → one
await dp.start_polling(bot, backlog="collapse")

EventContext¶

Property	Type	Description
`user_id`	int\|None	Sender user ID
`chat_id`	int\|None	Chat ID
`text`	str\|None	Message text
`payload`	str\|None	Deep link payload
`locale`	str\|None	User locale (BCP 47)
`callback_data`	str\|None	Callback button data
`from_user`	User\|None	Sender object

await event.answer("Text", keyboard=kb, format="markdown")
await event.hide_keyboard("New text")

Filters¶

@router.message_created(Command("start"))
@router.message_created(Command(["start", "go"]))

@router.bot_started(StartCommand())              # any deep link payload
@router.message_created(StartCommand())          # repeat deep link visit
async def handler(event, start_payload: str = None): ...

@router.message_created(Text(contains=["hello"]))
@router.message_created(StateFilter(MyState.waiting_name))
@router.message_created(F.message.body.text.func(lambda t: "hi" in t.lower()))
@router.callback_query(F.callback_data.startswith("action:"))

FSM¶

from aioscam.fsm import State, StatesGroup

class MyState(StatesGroup):
    waiting_name  = State()
    waiting_phone = State()

await state.set_state(MyState.waiting_name)
await state.update_data(name="John")
data = await state.get_data()
await state.set_state(None)

StateGuard — blocks commands and callback buttons during an active FSM state:

from magic_filter import F

dp = Dispatcher(
    state_guard_commands={'/cancel', '/start'},
    state_guard_callbacks=[
        'action:cancel',                # exact match
        F.startswith('confirm_'),       # also matches "confirm_yes|52507"
        F.regexp(r'^nav:(back|next)$'),
    ],
    state_guard_hint_func=lambda s: "expected input",
)

state_guard_callbacks — a list of strings (exact match) and/or magic_filter.F expressions (.startswith(), .contains(), .regexp(), combinable via & | ~). Use a list, not a set — F expressions are unhashable.

Middleware¶

async def my_middleware(event, handler):
    result = await handler(event)
    return result

router.middleware()(my_middleware)

Keyboards¶

from aioscam.utils.keyboard import KeyboardBuilder

builder = KeyboardBuilder(inline=True)
builder.callback("Button", "action:click")
builder.link("Website", "https://example.com")
builder.request_contact("📱 Share Phone")
builder.request_location("📍 Share Location")
builder.row()

await event.answer("Choose:", keyboard=builder.build().to_dict())

Deep Links¶

from aioscam.utils.deep_linking import create_deep_link, parse_deep_link

link = create_deep_link("mybot", "ref_123")  # https://max.ru/mybot?start=ref_123
info = parse_deep_link(link)  # {"bot_username": "mybot", "start": "ref_123"}

I18n¶

from aioscam import I18n
i18n = I18n(path="locales/", default_locale="en")
text = i18n.get("welcome", locale=event.locale or "en")

Rate Limiter¶

from aioscam.limiter import RateLimitConfig
bot = Bot(rate_limit=RateLimitConfig.strict())   # 5 req/s
bot = Bot(rate_limit=RateLimitConfig.relaxed())  # 30 req/s

Webhook¶

await dp.handle_webhook(bot=bot, host="0.0.0.0", port=8080,
                        path="/webhook", secret_token="secret")

WebApp (Mini Apps)¶

Max opens WebApps (mini apps) as plain HTML/CSS/JS inside the client’s WebView. aioscam.webapp provides the server-side half: validating what the WebApp page sends you, and pushing events back to it over SSE.

Validating `initData`¶

Every Max WebApp page receives a signed initData string from window.WebApp.initData. Validate it server-side before trusting any of it:

from aioscam.webapp import validate_init_data, WebAppSignatureError, WebAppExpiredError

try:
    data = validate_init_data(raw_init_data, bot_token, max_age=3600)
except WebAppSignatureError:
    ...  # tampered or wrong bot_token
except WebAppExpiredError:
    ...  # auth_date older than max_age

print(data.user.id, data.start_param)

validate_contact(...) does the same HMAC check for the payload returned by requestContact().

Pushing events to the WebApp (SSE)¶

from aioscam.webapp import EventStreamManager

events = EventStreamManager()

# in your /api/events route handler:
async def sse_handler(request):
    return await events.stream(request, user_id=data.user.id)

# anywhere else in the bot:
await events.publish(user_id=123, {"type": "bot_message", "text": "hi from the bot"})
await events.broadcast({"type": "announcement", "text": "deploy done"})

EventSource (the browser API behind SSE) cannot set custom headers, so the WebApp page passes initData as a query parameter on the SSE request: GET /api/events?initData=<raw>. WebAppMiddleware accepts initData from Authorization: MaxWebApp <raw>, the X-Webapp-Init-Data header, or the ?initData= query param — in that order.

Protecting your `/api/*` routes¶

from aioscam.webapp.aiohttp import WebAppMiddleware

app.middlewares.append(WebAppMiddleware(bot_token=bot.token))

The middleware only validates requests whose path starts with /api; everything else (including /static/* and your HTML pages) is left untouched and stays publicly reachable. It also splits its error responses on purpose: a request with no initData at all gets a plain 404 (indistinguishable from a route that doesn’t exist), while a request with a present-but-invalid initData (bad signature, expired, malformed) gets 401. A blind path scanner that never sends initData can’t tell /api/me apart from a 404 on a path that was never registered.

Landing page¶

HomePage gives the bare server root something safe to show instead of a hand-written index.html: bot name/description (from bot.get_me()) and an “Open in Max” deep link, no JS required, nothing that hints at /api/*. Mount your actual Mini App frontend under its own path and register that path — not the bare root — as the Mini App URL in the Max bot dashboard:

from aioscam.webapp.aiohttp import HomePage

app.router.add_get("/", HomePage(bot).handler)              # public landing page
app.router.add_get("/app", serve_index)                     # Mini App URL registered in the dashboard
app.router.add_static("/app", path=str(STATIC_DIR))

HomePage(bot, title=..., description=..., extra_head=..., extra_body=...) lets you override the copy or inject your own markup/scripts on top of the default shell — see examples/webapp_bot.py.

Masking `/api/*` from scanners¶

The 404/401 split above already hides whether a route exists from anyone who never sends initData. To raise the bar further against wordlist-style scanners (which try common names like /api, /api/me, /api/auth), move the API off the well-known /api prefix with api_prefix, and add a WebAppFailGuard to stop responding altogether to addresses that keep failing auth:

from aioscam.webapp.aiohttp import WebAppFailGuard, WebAppMiddleware

guard = WebAppFailGuard(max_failures=20, window=60, ban_seconds=300)
app.middlewares.append(
    WebAppMiddleware(bot_token=bot.token, api_prefix="/a8f3e1", fail_guard=guard)
)

Once an address crosses max_failures failed validations within window seconds, every request from it gets a flat 404 for ban_seconds — no signature check is even attempted. This is a defense-in-depth speed bump for automated probing, not a substitute for the HMAC check itself. Remember that your frontend’s fetch/EventSource calls must target the same api_prefix — see how examples/webapp_bot.py does it below.

A full working example — REST endpoints (/api/auth, /api/contact, /api/send), the SSE endpoint, and 4 frontend pages — lives in examples/webapp_bot.py + examples/webapp/*.html. Set WEBAPP_API_PREFIX=/your-secret before launching it and the demo moves its whole API there: the server rewrites the const API_PREFIX = "/api"; line in each served page on the fly, so the frontend picks up the real prefix without a build step or any hardcoded duplicate.

BotCapabilities¶

Max’s GET /me does not expose a permissions/capabilities field, so BotCapabilities builds a best-effort picture from the bot’s profile and configuration instead of letting you assume features that may not be set up:

from aioscam.utils.capabilities import BotCapabilities

caps = await BotCapabilities.probe(bot, webapp_url="https://example.com/webapp")
caps.log_report(logger)  # structured banner at startup, including warnings

Bridge SDK features (haptics, biometrics, NFC, QR, contacts) are client-side only — the server has no way to know the end user’s platform, so they are not part of this report; check window.WebApp.platform in the frontend instead.

ChatRegistry¶

In June 2026 Max removed GET /chats: a bot can no longer ask the server which chats it is in. ChatRegistry rebuilds that knowledge bot-side and persists it in SQLite (.aioscam/bot.db — the shared database of all framework components):

from aioscam import Bot, Dispatcher, ChatRegistry

registry = ChatRegistry()
dp = Dispatcher(registry=registry)
await dp.start_polling(bot, backlog="collapse")

await registry.chats()      # everything known — zero API calls
await registry.groups()     # groups only (type="chat")
await registry.dialogs()
await registry.get(chat_id)

How the registry fills itself:

bot_added / bot_started / chat_title_changed events are applied automatically; bot_removed / dialog_removed soft-delete the row (kept for history);
lazy discovery — any event from an unknown chat registers it;
persisted marker — the long-polling position survives restarts, so downtime events (e.g. being added to a group while offline) are not lost;
registry events are applied in all backlog modes, even "skip".

Manual reconciliation (Max has no “permissions changed” event — periodic checks are the only way):

stats = await registry.sync(bot)
# {'bootstrapped': N, 'checked': N, 'updated': N, 'removed': N}

sync() does a best-effort bootstrap via the deprecated GET /chats (while Max still answers it), then per-chat GET /chats/{id} lookups (403/404 → removed) and refreshes the bot’s permissions via GET /chats/{id}/members/me with a TTL cache (1 hour by default).

Example: examples/registry_bot.py.

Polls & Quizzes (PollManager)¶

Max Bot API has no native polls (unlike Telegram). PollManager emulates them over inline keyboards: votes live in SQLite (surviving restarts), the message shows live result bars, hint strings are localized (ru/en bundled — notifications follow the clicking user’s client locale, the shared message follows the creator’s; poll content is never translated).

from aioscam import PollManager

polls = PollManager()               # same .aioscam/bot.db
polls.attach(dp, command="poll")    # vote handler + /poll command + StateGuard allowlist

Users create polls right from the chat:

/poll Where do we eat? | Cafe | Canteen | Stay in
/poll priv Release score? | 5 | 4 | 3
/poll pub Friday deploy? | Yes | No

Visibility modes:

Mode	What the message shows
`pub`	bars + voter names under each option
`anon`	aggregate bars and numbers only (default)
`priv`	just “Votes: N”; the creator reads the breakdown via the private “📊 Results” button

Programmatic (bot-driven):

poll_id = await polls.send_poll(bot, chat_id, "Question?", ["Yes", "No"],
                                visibility="pub", creator_id=admin_id)
# creator_id=None → an "ownerless" poll: no control buttons at all,
# closing is possible only from code

await polls.send_quiz(bot, chat_id, "2+2?", ["3", "4"], correct_option=1,
                      explanation="Arithmetic.")   # instant private ✅/❌ feedback

results = await polls.results(poll_id)
await polls.close_poll(bot, poll_id)   # keyboard removed; a quiz reveals its answer

Voting mechanics: single choice — the vote can be moved; multiple=True — toggle; quiz — one answer forever. attach() adds the poll payload prefix and the /poll command to the StateGuard allowlists, so a user stuck inside an FSM dialog can still vote.

Example: examples/poll_bot.py.

Max API v2 migration & certificates¶

By July 19, 2026 every bot must move to platform-api2.max.ru — the old domains (platform-api.max.ru, botapi.max.ru) are being shut down. aioscam ≥0.2.2 handles this out of the box:

the default base URL is platform-api2.max.ru; /answers (callback replies) now lives on the same domain — the separate callback host is gone;
the new server is signed by the Mintsifry (Russian Trusted) CA, absent from the trust stores of most non-Russian systems. The framework bundles the official certificates (aioscam/certs/, downloaded from gosuslugi.ru) and trusts them for the bot’s own API connections only — your system trust store is never modified, nothing to install;
override with Bot(ssl_context=...) / AioScamClient(ssl_context=...);
the new server’s rate limit is 30 rps (the RateLimiter default of 10 rps leaves headroom);
GET /chats was removed from the API — Bot.get_chats() is deprecated, use ChatRegistry instead.

Exceptions & Hints¶

Every framework exception carries a .hint — a concrete reason or fix — appended automatically when you str() it:

try:
    bot = Bot()
except BotTokenError as e:
    print(e)  # "Bot token is not provided — pass token=... to Bot(), or set the MAX_BOT_TOKEN environment variable"
    print(e.hint)  # just the hint

ApiError, NetworkError, TimeoutError, RetryAfter, UnauthorizedError, ForbiddenError, NotFoundError, BotTokenError, DispatcherError, and the WebApp* exceptions (WebAppSignatureError, WebAppExpiredError, WebAppMissingFieldError, WebAppParseError, FeatureUnavailableError) all carry a sensible default hint — override it with hint=... on the raise if you need something more specific.

Configuration¶

Variable	Required	Description
`MAX_BOT_TOKEN`	✅	Bot token from Max
`AIOSCAM_ENV`	❌	`debug` / `test` / `prod` (default: `prod`)

License¶

PolyForm Noncommercial License 1.0.0 — free for personal, educational, and other noncommercial use. Commercial use requires a separate license — contact the author.