Aiogram Media Groups

aiogram media groups

Sending an album is easy. Receiving one is the part with no built-in support, and it is where bots break.

when to use this skill

• sending several photos or videos as one album
• handling a user who uploads several photos at once
• a handler that fires N times for one album
• captions appearing on the wrong item

sending

from aiogram.utils.media_group import MediaGroupBuilder

album = MediaGroupBuilder(caption="Trip photos")
album.add_photo(media="https://example.com/1.jpg")
album.add_photo(media="https://example.com/2.jpg")
album.add_photo(media="https://example.com/3.jpg", has_spoiler=True)

media = album.build()
assert len(media) == 3
assert media[0].caption == "Trip photos"
assert media[1].caption is None

Verified surface:

MediaGroupBuilder(media=None, caption=None, caption_entities=None)
  .add(**kwargs)          # explicit type= key
  .add_photo(media, caption=None, parse_mode=..., caption_entities=None,
             has_spoiler=None, **kwargs)
  .add_video(media, thumbnail=None, caption=None, ..., width=None, height=None,
             duration=None, supports_streaming=None, has_spoiler=None, **kwargs)
  .add_document(...)
  .add_audio(...)
  .build() -> list[InputMediaAudio | InputMediaPhoto | InputMediaVideo | InputMediaDocument]

The builder's caption lands on the first item, a convention for a shared caption. Telegram accepts captions on individual items; multiple captions may be displayed separately by the client.

from aiogram import Router
from aiogram.types import FSInputFile, Message
from aiogram.utils.formatting import Bold, Text
from aiogram.utils.media_group import MediaGroupBuilder

router = Router(name="send-album")


@router.message()
async def send_album(message: Message) -> None:
    caption = Text("Report for ", Bold("July"))
    text, entities = caption.render()
    album = MediaGroupBuilder(caption=text, caption_entities=entities)
    album.add_photo(media=FSInputFile("chart-1.png"))
    album.add_photo(media=FSInputFile("chart-2.png"))
    await message.answer_media_group(media=album.build())

the rules Telegram enforces

• 2–10 items. One item is rejected; use send_photo. Eleven is rejected; chunk.
• Photos and videos may be mixed in one album. Documents may only group with documents, and audio only with audio.
• An album cannot carry a reply_markup. Buttons need a separate message.
• Each item may be a file_id, an http URL, or an InputFile.

from aiogram.utils.media_group import MediaGroupBuilder


def split_sizes(count: int) -> list[int]:
    """Chunk into groups of 2-10. A lone tail item borrows one from the group before it."""
    if count == 1:
        raise ValueError("Use send_photo for one item")
    if count < 1:
        return []
    sizes = [10] * (count // 10)
    remainder = count % 10
    if remainder == 1 and sizes:
        sizes[-1] -= 1  # 10 + 1 -> 9 + 2
        remainder = 2
    if remainder:
        sizes.append(remainder)
    return sizes


def chunk_album(urls: list[str]) -> list[list[object]]:
    groups: list[list[object]] = []
    start = 0
    for size in split_sizes(len(urls)):
        builder = MediaGroupBuilder()
        for url in urls[start : start + size]:
            builder.add_photo(media=url)
        groups.append(list(builder.build()))
        start += size
    return groups


assert split_sizes(11) == [9, 2]
assert split_sizes(12) == [10, 2]
assert split_sizes(20) == [10, 10]
assert split_sizes(21) == [10, 9, 2]
assert [len(g) for g in chunk_album([f"u{i}" for i in range(11)])] == [9, 2]
assert all(2 <= len(g) <= 10 for g in chunk_album([f"u{i}" for i in range(37)]))

receiving: the actual problem

Telegram sends an album as separate message updates that share a media_group_id. There is no count, no terminator, and no guaranteed ordering window. A naive handler runs once per item:

from aiogram import F, Router
from aiogram.types import Message

router = Router(name="naive")


@router.message(F.photo)
async def wrong(message: Message) -> None:
    """Fires 5 times for a 5-photo album, and answers 5 times."""
    await message.answer("Got your photo!")

aiogram has no built-in album aggregator. Anything you write here is your own code, not an aiogram feature. The standard approach is a debounce: buffer by media_group_id, wait a short quiet period, then dispatch once.

import asyncio
from collections import defaultdict
from typing import Any

from aiogram import BaseMiddleware, Router
from aiogram.types import Message, TelegramObject

router = Router(name="album")


class AlbumMiddleware(BaseMiddleware):
    """Collects album items and calls the handler once with `album` injected.

    Not an aiogram feature - a debounce you own. `latency` trades responsiveness
    against the risk of splitting an album across two calls.
    """

    def __init__(self, latency: float = 0.6) -> None:
        self.latency = latency
        self.parts: dict[tuple[int, int, str], list[Message]] = defaultdict(list)

    async def __call__(
        self,
        handler: Any,
        event: TelegramObject,
        data: dict[str, Any],
    ) -> Any:
        if not isinstance(event, Message) or event.media_group_id is None:
            return await handler(event, data)

        key = (data["bot"].id, event.chat.id, event.media_group_id)
        album = self.parts[key]
        album.append(event)

        # Only the most recent arrival in this exact bucket may dispatch.
        try:
            await asyncio.sleep(self.latency)
        except asyncio.CancelledError:
            if self.parts.get(key) is album and album[-1] is event:
                self.parts.pop(key, None)
            raise
        if self.parts.get(key) is not album or album[-1] is not event:
            return None
        self.parts.pop(key)

        album.sort(key=lambda item: item.message_id)
        data["album"] = album
        return await handler(album[0], data)


router.message.outer_middleware(AlbumMiddleware())


@router.message()
async def on_album(message: Message, album: list[Message] | None = None) -> None:
    """`album` is present only for grouped messages; keep the default."""
    if album is None:
        await message.answer("single item")
        return
    await message.answer(f"received {len(album)} items")

Points that matter in that middleware:

• It must be an outer middleware. An inner one runs after filters, so items rejected by a filter never reach the buffer.
• Inspect captions on every item; a sender may caption an item other than the first.
• Sort by message_id. Arrival order is not guaranteed.
• Handlers must tolerate album=None, because non-grouped messages still flow through.

the trade-offs, stated honestly

Do not disable FSM isolation just to make this collector work. When updates are serialised, use a collector outside the locked handler path and a managed dispatch task. A quiet period is a heuristic, never proof that Telegram delivered the entire album.

If you need this to be reliable across replicas, buffer in Redis and dispatch from a short-lived task keyed on the group id — but the same debounce logic and the same trade-offs apply.

sending an album back

from aiogram import Router
from aiogram.types import Message
from aiogram.utils.media_group import MediaGroupBuilder

router = Router(name="echo-album")


@router.message()
async def echo_album(message: Message, album: list[Message] | None = None) -> None:
    if not album:
        return
    builder = MediaGroupBuilder(caption="Here they are")
    for item in album:
        if item.photo:
            builder.add_photo(media=item.photo[-1].file_id)
    media = builder.build()
    if len(media) >= 2:
        await message.answer_media_group(media=media)

Re-sending by file_id means no upload at all — see aiogram-files.

checklist

• outgoing albums contain 2–10 items, chunked when longer
• caption set once, via the builder, landing on the first item
• no reply_markup on an album
• incoming albums aggregated deliberately, with the debounce documented as your code
• aggregation done in outer middleware
• items sorted by message_id
• captions inspected on all incoming items
• handlers tolerate a missing album
• multi-replica deployments use a shared store, not a process-local dict

see also

• aiogram-files — file_id reuse and upload sources
• aiogram-middlewares — outer versus inner
• skills/using-aiogram/references/telegram-quirks.md — album delivery semantics