Dispatcher - класс aiogram отвечающий за обработку изменений

Dispatcher (диспетчер) в aiogram 3 принимает от Telegram обновления (новые сообщения, нажатия кнопок, inline-запросы) и передаёт каждое подходящему обработчику. Он же и корневой роутер: обработчики пишут в отдельных объектах Router и подключают к диспетчеру через dp.include_router(). Запускают бота вызовом await dp.start_polling(bot) внутри asyncio.run().

Примеры проверены на aiogram 3.x (версия 3.31.0) и Python 3.14.

Содержание

Что изменилось по сравнению с aiogram 2

Если вы видели старые примеры, вот главные отличия. В aiogram 2 бота передавали в диспетчер: Dispatcher(bot). В третьей версии диспетчер создают пустым, Dispatcher(), а бота передают при запуске. Методы register_message_handler и декораторы @dp.message_handler убрали, вместо них @router.message(...). Модуля executor больше нет: бота запускают через asyncio.run(). Код второй версии в третьей не запустится, его придётся переписать.

Dispatcher и Router

Роутер (Router) — это набор обработчиков. Удобно думать о нём как о папке: в одной лежат обработчики команд, в другой — обработчики оплаты, в третьей — админка. Диспетчер — тот же роутер, только главный. Он умеет всё, что умеет роутер, и ещё запускает бота.

Когда приходит обновление, диспетчер проверяет свои обработчики, потом обработчики подключённых роутеров в порядке подключения. Сработает первый, чьи фильтры подошли. Остальные это обновление уже не увидят.

Минимальный бот

import asyncio
import logging
import os

from aiogram import Bot, Dispatcher, F, Router
from aiogram.filters import CommandStart
from aiogram.types import ErrorEvent, Message

router = Router()


@router.message(CommandStart())
async def on_start(message: Message):
    await message.answer("Привет! Я бот на aiogram 3.")


@router.message(F.text)
async def on_text(message: Message):
    await message.answer(f"Вы написали: {message.text}")


@router.errors()
async def on_error(event: ErrorEvent):
    logging.exception("Ошибка при обработке обновления", exc_info=event.exception)


async def main():
    logging.basicConfig(level=logging.INFO)
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)

    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Токен бота (ключ от BotFather) берётся из переменной окружения BOT_TOKEN. В Windows её задают командой set BOT_TOKEN=ваш_токен, в Linux и macOS — export BOT_TOKEN=ваш_токен. В код токен не пишут: иначе он рано или поздно окажется в репозитории или на скриншоте.

Строка await bot.delete_webhook(drop_pending_updates=True) делает две вещи. Снимает вебхук, если он был настроен раньше: с включённым вебхуком опрос не работает. И выбрасывает сообщения, которые накопились, пока бот был выключен, чтобы он не отвечал на них разом после запуска.

Разбиение бота на модули

Когда обработчиков становится много, их раскладывают по файлам. В каждом файле свой роутер, а главный файл только собирает их.

# handlers/start.py
from aiogram import Router
from aiogram.filters import CommandStart
from aiogram.types import Message

router = Router(name="start")


@router.message(CommandStart())
async def on_start(message: Message):
    await message.answer("Привет!")

# bot.py
import asyncio
import os

from aiogram import Bot, Dispatcher

from handlers import start


async def main():
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_routers(start.router)
    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

include_routers() подключает сразу несколько роутеров: dp.include_routers(start.router, admin.router, payments.router). Роутер можно подключить и к другому роутеру, получится дерево. Подключить один роутер в два места нельзя, aiogram выдаст ошибку.

Обработчики событий

У диспетчера и у каждого роутера есть наблюдатели (observers) — по одному на тип события. К наблюдателю привязывают обработчики. Основные:

message — новые сообщения: текст, команды, фото, стикеры.
edited_message — пользователь отредактировал своё сообщение.
callback_query — нажатие встроенной (inline) кнопки под сообщением.
inline_query — пользователь набирает @имя_бота запрос в любом чате.
chosen_inline_result — пользователь выбрал один из вариантов inline-режима.
shipping_query — Telegram спрашивает, куда и как доставить товар (для платежей с доставкой).
pre_checkout_query — последняя проверка перед оплатой: бот должен подтвердить заказ за 10 секунд.
my_chat_member — бота добавили в группу, удалили или заблокировали.
errors — ошибка внутри любого обработчика.

Все наблюдатели работают одинаково, поэтому покажу два способа регистрации на примере message и callback_query.

Через декоратор, это основной способ:

from aiogram import F, Router
from aiogram.filters import Command
from aiogram.types import CallbackQuery, Message

router = Router()


@router.message(Command("start"))
async def on_start(message: Message):
    await message.answer("Привет!")


@router.callback_query(F.data == "button_pressed")
async def on_button_pressed(callback: CallbackQuery):
    await callback.answer("Кнопка нажата")

Без декоратора, через register(). Это замена старых register_message_handler, register_callback_query_handler и других. Пригодится, если обработчик подключают по условию или он лежит в чужом модуле:

async def on_help(message: Message):
    await message.answer("Справка")


router.message.register(on_help, Command("help"))

В скобках декоратора — фильтры. Command("start") ловит команду, F.text — любой текст, F.photo — фото, F.data == "button_pressed" — нажатие кнопки с этим значением. С F (магическим фильтром) условие пишется одной строкой: F.text.lower().contains("привет"), F.from_user.id == 123456789. Несколько фильтров через запятую работают как «и».

Обработка ошибок

В минимальном боте выше есть обработчик @router.errors(). Он ловит исключения из всех обработчиков роутера. Само исключение лежит в event.exception, а обновление, на котором бот споткнулся, — в event.update. Без такого обработчика aiogram всё равно запишет ошибку в лог и продолжит работу, но своей реакции, например сообщения администратору, не будет.

Атрибуты и данные диспетчера

В aiogram 2 у диспетчера были атрибуты middlewares и loop. В третьей версии атрибута loop нет: цикл событий (event loop) создаёт asyncio.run(). Промежуточные слои подключают к конкретному наблюдателю, об этом ниже.

Зато диспетчер работает как словарь. Всё, что в него положить, придёт в обработчик по имени параметра. Так передают подключение к базе, настройки, общий клиент HTTP:

dp = Dispatcher()
dp["admin_name"] = "Виктор"


@router.message(Command("whoami"))
async def whoami(message: Message, admin_name: str):
    await message.answer(f"Мой администратор: {admin_name}")

Тот же эффект дают именованные аргументы при запуске: await dp.start_polling(bot, admin_name="Виктор"). Объект bot приходит в обработчик так же, достаточно добавить параметр bot: Bot.

Метод start_polling: как бот получает обновления

Telegram доставляет боту обновления двумя способами.

Опрос (long polling). Бот сам спрашивает сервер Telegram: «есть что-нибудь новое?». Если нового нет, сервер держит запрос открытым до 10 секунд и отвечает, как только что-то появится. Внутри это метод getUpdates из Bot API. Ничего настраивать не нужно, бот работает хоть на домашнем ноутбуке.

Вебхук (webhook). Бот сообщает Telegram адрес своего сервера, и Telegram сам отправляет туда каждое обновление. Для этого нужен сервер с публичным адресом и HTTPS-сертификатом.

dp.start_polling(bot) включает первый способ. Метод асинхронный: пока один обработчик ждёт ответа от базы или сайта, бот обслуживает других пользователей. Полезные параметры:

allowed_updates — список типов обновлений, которые нужны боту. По умолчанию aiogram сам вычисляет его по зарегистрированным обработчикам.
polling_timeout — сколько секунд держать запрос к Telegram открытым, по умолчанию 10.
можно передать несколько ботов: await dp.start_polling(bot1, bot2). Один диспетчер с одними обработчиками обслужит обоих.

Остановить бота можно сочетанием Ctrl+C: aiogram корректно закроет соединения.

Вебхуки

Как это работает:

Бот вызывает set_webhook и передаёт адрес, например https://bot.example.ru/webhook.
Пользователь пишет боту, Telegram собирает обновление в JSON (текстовый формат данных).
Telegram отправляет этот JSON POST-запросом на указанный адрес.
Сервер бота принимает запрос и передаёт обновление в диспетчер, а тот вызывает обработчик.

Плюсы вебхука: обновление приходит сразу, бот не тратит запросы на пустые опросы, проще масштабировать под большую нагрузку. Минусы: нужен сервер с доменом и HTTPS, запустить такого бота на ноутбуке без дополнительных инструментов не выйдет. Для учебного бота и для бота на одну группу опроса хватает с запасом.

Пример вебхука на aiohttp

В старой версии статьи вебхук принимал сервер на Flask. В aiogram 3 для этого есть готовая связка с веб-сервером aiohttp: он уже стоит вместе с aiogram, ставить отдельно не нужно.

import os

from aiohttp import web

from aiogram import Bot, Dispatcher, Router
from aiogram.filters import CommandStart
from aiogram.types import Message
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application

WEBHOOK_PATH = "/webhook"
BASE_URL = os.environ["BASE_URL"]  # например, https://bot.example.ru
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"]

router = Router()


@router.message(CommandStart())
async def on_start(message: Message):
    await message.answer("Я работаю через вебхук")


async def on_startup(bot: Bot):
    await bot.set_webhook(
        f"{BASE_URL}{WEBHOOK_PATH}",
        secret_token=WEBHOOK_SECRET,
        drop_pending_updates=True,
    )


def main():
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)
    dp.startup.register(on_startup)

    app = web.Application()
    SimpleRequestHandler(dispatcher=dp, bot=bot, secret_token=WEBHOOK_SECRET).register(
        app, path=WEBHOOK_PATH
    )
    setup_application(app, dp, bot=bot)
    web.run_app(app, host="127.0.0.1", port=8080)


if __name__ == "__main__":
    main()

Что здесь происходит. При старте диспетчер вызывает on_startup, и бот сообщает Telegram свой адрес. SimpleRequestHandler принимает POST-запросы на /webhook и передаёт обновления в диспетчер. secret_token — строка-пароль: Telegram кладёт её в заголовок каждого запроса, и обработчик отбрасывает запросы без неё. Так посторонний не сможет прислать боту поддельное обновление.

Сервер слушает только локальный адрес 127.0.0.1:8080. Наружу его выводят через nginx (веб-сервер, который принимает HTTPS-запросы и пересылает их приложению). Сертификат для домена бесплатно выдаёт Let’s Encrypt. Telegram принимает вебхуки только на портах 443, 80, 88 и 8443.

Если захотите вернуться к опросу, запустите бота через start_polling: строка delete_webhook снимет вебхук.

Промежуточные слои (middleware)

Middleware (промежуточный слой) — код, который выполняется до и после обработчика. Это похоже на охрану на входе: она видит каждого посетителя, может записать его в журнал, выдать пропуск или не пустить. В боте так делают логирование, проверку доступа, подсчёт запросов, подключение к базе.

В aiogram 3 middleware — это класс с методом __call__. Он получает три вещи: handler — следующий шаг цепочки, event — само событие (например, Message), data — словарь с данными, которые уйдут в обработчик. Чтобы пропустить событие дальше, вызывают await handler(event, data). Не вызвали — обработчик не сработает.

Подключают двумя способами:

dp.message.outer_middleware(...) — внешний слой. Срабатывает на каждое сообщение, ещё до проверки фильтров.
dp.message.middleware(...) — внутренний слой. Срабатывает, только когда фильтры подошли и обработчик найден.

Для всех типов событий сразу есть dp.update.outer_middleware(...). Старого dispatcher.middleware.setup() в третьей версии нет.

Пример: логирование, подсчёт запросов и доступ по списку

import asyncio
import logging
import os
from typing import Any, Awaitable, Callable

from aiogram import BaseMiddleware, Bot, Dispatcher, F, Router
from aiogram.filters import Command
from aiogram.types import Message, TelegramObject

ALLOWED_USERS = {123456789, 987654321}

router = Router()


class LoggingMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: dict[str, Any],
    ) -> Any:
        logging.info("Сообщение от %s: %s", event.from_user.id, event.text)
        return await handler(event, data)


class CounterMiddleware(BaseMiddleware):
    def __init__(self) -> None:
        self.count = 0

    async def __call__(self, handler, event, data):
        self.count += 1
        data["request_count"] = self.count
        return await handler(event, data)


class AuthMiddleware(BaseMiddleware):
    async def __call__(self, handler, event: Message, data):
        if event.from_user.id not in ALLOWED_USERS:
            await event.answer("Доступ закрыт")
            return None
        return await handler(event, data)


@router.message(Command("stats"))
async def stats(message: Message, request_count: int, admin_name: str):
    await message.answer(f"Запрос №{request_count}. Админ бота: {admin_name}")


@router.message(F.text)
async def echo(message: Message):
    await message.answer(message.text)


async def main():
    logging.basicConfig(level=logging.INFO)
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp["admin_name"] = "Виктор"

    dp.message.outer_middleware(LoggingMiddleware())
    dp.message.middleware(AuthMiddleware())
    dp.message.middleware(CounterMiddleware())

    dp.include_router(router)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Три слоя работают по очереди. LoggingMiddleware внешний: пишет в лог любое сообщение, даже если обработчика для него нет. AuthMiddleware пропускает только пользователей из ALLOWED_USERS, остальным отвечает «Доступ закрыт» и дальше событие не передаёт. CounterMiddleware считает запросы и кладёт номер в data["request_count"], а обработчик /stats получает его как обычный параметр. Порядок подключения важен: счётчик стоит после проверки доступа, поэтому чужие сообщения он не считает.

Свой id в Telegram можно узнать у бота @userinfobot. Список ALLOWED_USERS в настоящем проекте лучше хранить вне кода: в переменной окружения или в файле настроек, как и токен. Счётчик живёт в памяти и обнулится при перезапуске бота.

Примеры обработки обновлений

Обработка сообщений

Обработчик только текстовых сообщений. Фильтр F.text заменяет старый content_types=types.ContentTypes.TEXT: фото, стикеры и голосовые до этой функции не дойдут.

@router.message(F.text)
async def handle_text_message(message: Message):
    await message.answer(f"Вы написали: {message.text}")

Для других типов: F.photo, F.document, F.voice, F.sticker. Полный бот с таким обработчиком — в разделе «Минимальный бот» выше.

Обработка inline-кнопок

Бот по команде /menu присылает сообщение с тремя кнопками под ним. Нажатие кнопки приходит как CallbackQuery (запрос от кнопки) со строкой callback_data.

import asyncio
import os

from aiogram import Bot, Dispatcher, F, Router
from aiogram.filters import Command
from aiogram.types import CallbackQuery, Message
from aiogram.utils.keyboard import InlineKeyboardBuilder

router = Router()


@router.message(Command("menu"))
async def show_menu(message: Message):
    builder = InlineKeyboardBuilder()
    for i in range(1, 4):
        builder.button(text=f"Кнопка {i}", callback_data=f"button_{i}")
    builder.adjust(1)
    await message.answer("Выберите кнопку:", reply_markup=builder.as_markup())


@router.callback_query(F.data.startswith("button_"))
async def handle_inline_button(callback: CallbackQuery):
    number = callback.data.split("_")[1]
    await callback.answer()
    await callback.message.answer(f"Вы нажали кнопку {number}")


async def main():
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

InlineKeyboardBuilder собирает клавиатуру, adjust(1) ставит кнопки по одной в ряд. Фильтр F.data.startswith("button_") ловит все три кнопки одним обработчиком. callback.answer() убирает часики загрузки на кнопке. В старой версии примера бот отправлял меню сам при запуске на заранее известный chat_id. Здесь меню приходит по команде, поэтому номер чата знать не нужно.

Как по нажатию кнопки отредактировать то же сообщение, а не присылать новое, разобрано в статье про управление сообщениями.

Обработчик inline-режима

В inline-режиме бота вызывают в любом чате, даже там, где бота нет. Пользователь набирает в строке ввода @имя_бота текст, Telegram показывает список вариантов от бота, пользователь выбирает один, и тот уходит в чат от имени пользователя.

import asyncio
import os

from aiogram import Bot, Dispatcher, Router
from aiogram.types import InlineQuery, InlineQueryResultArticle, InputTextMessageContent

router = Router()


@router.inline_query()
async def handle_inline_query(inline_query: InlineQuery):
    query = inline_query.query or "пусто"
    results = [
        InlineQueryResultArticle(
            id="upper",
            title="Большими буквами",
            description=query.upper(),
            input_message_content=InputTextMessageContent(message_text=query.upper()),
        ),
        InlineQueryResultArticle(
            id="reverse",
            title="Задом наперёд",
            description=query[::-1],
            input_message_content=InputTextMessageContent(message_text=query[::-1]),
        ),
    ]
    await inline_query.answer(results, cache_time=1)


async def main():
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Текст запроса лежит в inline_query.query. Бот возвращает два варианта: тот же текст большими буквами и задом наперёд. Каждый вариант — объект InlineQueryResultArticle: title и description видны в списке, input_message_content уйдёт в чат. cache_time=1 просит Telegram не запоминать ответ надолго, иначе при отладке вы будете видеть старые варианты. В настоящем боте варианты обычно ищут в базе или во внешнем сервисе.

Как включить и проверить inline-режим

По умолчанию inline-режим у бота выключен. Откройте BotFather, отправьте /setinline, выберите бота и задайте подсказку, которую пользователь увидит в строке ввода. Затем запустите скрипт, откройте любой чат и наберите @имя_вашего_бота привет. Через секунду над строкой ввода появятся два варианта. Добавлять бота в этот чат не нужно.

Уроки по теме

Спокойный вход с мини-проектами — уроки курса (в нём более простая библиотека telebot):

Бот здоровается: команда /start — курс «Создание чат-ботов»