# Aiogram Streaming

> Use when streaming aiogram 3 text/progress with message drafts, rich messages, throttled edits, or chat actions.

- Skill: `ballisarium/aiogram-streaming` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add ballisarium/aiogram-streaming`
- Raw SKILL.md: https://api.skillmd.com/api/skills/ballisarium/aiogram-streaming/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-streaming

---


# aiogram streaming

Telegram has no streaming socket. Producing incremental output means one of three things,
and only the first two are real API features.

| Approach | Real API? | Best for |
|---|---|---|
| **message drafts** (`sendMessageDraft`) | yes, Bot API 10.x | LLM output, animated in-progress text |
| **rich messages** (`sendRichMessage`) | yes, Bot API 10.x | structured long-form output |
| **edit throttling** | pattern you write | anything, any Bot API version |
| chat actions (`typing`) | yes | short waits with no partial output |

Verified present in aiogram 3.31.0: `Bot.send_message_draft`, `Bot.send_rich_message`,
`Bot.send_rich_message_draft`.

```bash
python skills/using-aiogram/tools/get-method.py sendMessageDraft
python skills/using-aiogram/tools/get-type.py InputRichMessage
```

## when to use this skill

- streaming an LLM response into a chat
- showing progress on a long job
- a bot hitting flood limits while editing a message repeatedly
- deciding between a draft, a rich message, and edit throttling

## message drafts

`sendMessageDraft` shows evolving text in the chat before a real message exists. Quoted
from the field documentation carried by aiogram 3.31.0:

- `draft_id` — *"Unique identifier of the message draft; must be non-zero. Changes to
  drafts with the same identifier are animated"*
- `text` — *"Text of the message to be sent, 0-4096 characters after entities parsing.
  Pass an empty text to show a 'Thinking…' placeholder"*
- `chat_id` — *"Unique identifier for the target private chat"*

Two consequences that decide the design: **reuse one `draft_id`** so updates animate
rather than replace, and drafts are for **private chats**.

```python
import time
from collections.abc import AsyncIterator

from aiogram import Bot


async def stream_with_draft(
    bot: Bot,
    chat_id: int,
    chunks: AsyncIterator[str],
    draft_id: int,
) -> str:
    """Single-message example; the caller allocates a non-zero id per generation."""
    await bot.send_message_draft(chat_id=chat_id, draft_id=draft_id, text="", parse_mode=None)
    buffer = ""
    last_update = time.monotonic()
    async for chunk in chunks:
        buffer += chunk
        if len(buffer.encode("utf-16-le")) // 2 > 4096:
            raise ValueError("Split long output into messages in the caller")
        if buffer and time.monotonic() - last_update >= 1.2:
            await bot.send_message_draft(
                chat_id=chat_id,
                draft_id=draft_id,
                text=buffer,
                parse_mode=None,
            )
            last_update = time.monotonic()
    if buffer:
        await bot.send_message(chat_id=chat_id, text=buffer, parse_mode=None)
    return buffer
```

`send_message_draft` returns `bool`, not a `Message` — a draft is not a message. Send the
final text separately when you need a message id to edit, reply to, or pin.

Draft updates are paced here, but 429 and transport errors propagate. Apply the bounded
`with_flood_retry` pattern from `aiogram-rate-limits` at the call boundary where needed.
Pacing does not guarantee avoiding Telegram limits.

## rich messages

`sendRichMessage` takes an `InputRichMessage` and returns a real `Message`. Content is
given as `html`, `markdown`, or a list of typed `blocks`:

```python
from aiogram import Bot
from aiogram.types import (
    InputRichBlockParagraph,
    InputRichBlockPreformatted,
    InputRichBlockSectionHeading,
    InputRichMessage,
)


async def send_report(bot: Bot, chat_id: int) -> None:
    content = InputRichMessage(
        blocks=[
            InputRichBlockSectionHeading(text="Build report"),
            InputRichBlockParagraph(text="All checks passed."),
            InputRichBlockPreformatted(text="pytest: 42 passed"),
        ]
    )
    await bot.send_rich_message(chat_id=chat_id, rich_message=content)


block = InputRichBlockParagraph(text="hello")
assert block.type == "paragraph"
assert InputRichMessage(blocks=[block]).blocks is not None
```

`InputRichMessage` fields: `html`, `markdown`, `is_rtl`, `skip_entity_detection`,
`blocks`, `media`. Use `blocks` for structure you control; use `html`/`markdown` when the
content already exists in that form, with `media` supplying the `tg://photo?id=` targets.

Block types available in aiogram 3.31.0 include paragraph, section heading, preformatted,
list and list item, table, block quotation, pull quotation, details, divider, footer,
anchor, collage, slideshow, map, mathematical expression, thinking, and the media blocks
(photo, video, audio, animation, voice note).

`send_rich_message_draft(chat_id, draft_id, rich_message, ...)` is the
draft form — the same animation behaviour, with structured content.

## edit throttling: the portable pattern

This works on any Bot API version and in any chat type. It is **your code**, not an
aiogram feature.

```python
import asyncio
import time
from collections.abc import AsyncIterator

from aiogram import Bot
from aiogram.exceptions import TelegramBadRequest, TelegramRetryAfter
from aiogram.types import Message

MIN_INTERVAL = 1.2  # application pacing; Telegram may require a longer delay
MAX_LENGTH = 4096


async def stream_by_editing(
    bot: Bot,
    chat_id: int,
    chunks: AsyncIterator[str],
) -> Message:
    """Coalesce a single message; reject overflow instead of silently losing text."""
    message = await bot.send_message(chat_id=chat_id, text="…", parse_mode=None)
    buffer = ""
    shown = ""
    last_edit = time.monotonic()

    async def flush() -> None:
        nonlocal shown, last_edit
        if buffer == shown or not buffer:
            return
        await asyncio.sleep(max(0.0, MIN_INTERVAL - (time.monotonic() - last_edit)))
        for attempt in range(3):
            try:
                await bot.edit_message_text(
                    chat_id=chat_id,
                    message_id=message.message_id,
                    text=buffer,
                    parse_mode=None,
                )
            except TelegramRetryAfter as error:
                if attempt == 2:
                    raise
                await asyncio.sleep(error.retry_after)
                continue
            except TelegramBadRequest as error:
                if "message is not modified" not in str(error):
                    raise
            shown = buffer
            last_edit = time.monotonic()
            return

    async for chunk in chunks:
        buffer += chunk
        if len(buffer.encode("utf-16-le")) // 2 > MAX_LENGTH:
            raise ValueError("Split long output into messages in the caller")
        if time.monotonic() - last_edit >= MIN_INTERVAL:
            await flush()
    if not buffer:
        buffer = "No output was generated."
    await flush()  # retry the tail too; exhaustion is reported to the caller
    return message
```

Every line there exists for a reason:

- **Coalesce.** Token-by-token edits are the fastest way to a flood limit.
- **Compare before editing.** Identical text raises `message is not modified`.
- **Check UTF-16 length.** This example rejects overflow; the caller must split long output.
- **Honour `retry_after`.** Never a hardcoded sleep.
- **Flush the tail.** Otherwise the last few tokens never appear.

### streaming partial markup

Never send partially generated HTML or MarkdownV2. A half-written `<b>` raises
`can't parse entities` and the edit is lost. Two safe options:

1. Stream as **plain text** (`parse_mode=None`) and send one final formatted edit.
2. Compose with `aiogram.utils.formatting` entities, which cannot produce malformed
   markup — see `aiogram-formatting`.

## chat actions

When there is no partial output to show, an action indicator is enough:

```python
import asyncio

from aiogram import Bot, Router
from aiogram.types import Message
from aiogram.utils.chat_action import ChatActionSender

router = Router(name="thinking")


@router.message()
async def slow(message: Message, bot: Bot) -> None:
    async with ChatActionSender.typing(bot=bot, chat_id=message.chat.id):
        await asyncio.sleep(0)  # the real work
    await message.answer("finished")
```

`ChatActionSender` re-sends every 5 seconds, to refresh the indicator, which Telegram keeps for at most five seconds. `@flags.chat_action("typing")` plus `ChatActionMiddleware` does the same
declaratively — see `aiogram-middlewares`.

Use `ChatActionSender.upload_document`, `.upload_photo`, `.record_voice`, and friends when
the work is an upload; the indicator then matches what the user is waiting for.

## choosing

| Situation | Use |
|---|---|
| LLM output, private chat, Bot API 10.x | message draft, then a final `send_message` |
| structured long-form output | rich message with `blocks` |
| group chat, or older Bot API | edit throttling |
| under ~3 seconds, no partial output | `ChatActionSender` |
| output over 4096 characters | split into several messages |

## cancellation

The application owns generation tasks. Keep a task per conversation (including bot and
thread when relevant), cancel and await a previous task before replacing it, and remove
it on completion only if the registry still points to that same task. Propagate
`CancelledError` after cleanup and settle the visible message deliberately.

Bot API 10.3 adds `can_stop` and `keep_on_stop` to text/rich drafts, plus
`router.stopped_message_generation()` updates carrying `chat`, `draft_id`, and optional
`message_thread_id`. Map these to your own task registry; the event has no `from_user`.
Include this observer in webhook `allowed_updates` when offering stop controls.

## checklist

- edits or draft updates coalesced to roughly one per second or slower
- text compared before every edit
- long output split deliberately; never silently truncated
- `retry_after` honoured
- the tail always flushed
- no partially generated HTML or Markdown sent
- drafts used only in private chats, with a stable non-zero `draft_id`
- a real message sent at the end when a message id is needed
- one stream per chat, previous ones cancelled

## see also

- `aiogram-formatting` — entity-safe text
- `aiogram-rate-limits` — flood limits and pacing
- `aiogram-errors` — `message is not modified`, `can't parse entities`
- `aiogram-middlewares` — `@flags.chat_action`