# Aiogram Media Groups

> Use when sending Telegram albums or aggregating incoming media_group_id updates with aiogram 3.

- Skill: `ballisarium/aiogram-media-groups` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add ballisarium/aiogram-media-groups`
- Raw SKILL.md: https://api.skillmd.com/api/skills/ballisarium/aiogram-media-groups/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: ballisarium (https://skillmd.com/u/ballisarium)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/ballisarium/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

```python
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:

```text
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.

```python
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`.

```python
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:

<!-- example: run -->
```python
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.

```python
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

| Concern | Reality |
|---|---|
| latency | too short splits albums; too long makes the bot feel slow. 0.5–1.0 s is typical |
| memory | an in-process dict grows if a group never completes; cap it or expire entries |
| multiple replicas | items of one album can land on different processes — a dict cannot see them. Use Redis keyed by `media_group_id`, with a short TTL |
| holding the handler | `asyncio.sleep` inside a middleware holds that update's task for the whole latency |
| filters | outer middleware sees messages reaching its router |
| concurrency | this example needs concurrent update tasks; FSM isolation may serialise the same user before it reaches this middleware |

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

```python
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