1 марта 2026 Telegram добавил в Bot API метод sendMessageDraft - возможность потокового вывода сообщений. Тот самый эффект, к которому все привыкли в ChatGPT и Claude. Текст появляется по частям, в конце бегают анимированные точки, и ты видишь, что ответ ещё генерируется.

Я написал простенький рабочий пример на чистом Python - без каких-либо фреймворков. Только asyncio и urllib.

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

Принцип простой:

1. Бот создаёт черновик сообщения через sendMessageDraft

2. По мере получения чанков обновляет этот черновик

3. Пока идёт стрим - Telegram показывает анимированные точки в конце текста

4. Когда стрим завершён - бот отправляет финальное сообщение через обычный sendMessage

Черновик закрепляется в чате и обновляется в реальном времени. Пользователь видит, как текст печатается.

sendMessageDraft

await api.arequest(
    "sendMessageDraft",
    {"chat_id": chat_id, "draft_id": draft_id, "text": accumulated_text},
)

Параметры:

• chat_id - куда отправляем

• draft_id - уникальный идентификатор черновика (int). Все обновления одного стрима должны идти с одним draft_id

• text - текущий накопленный текст

draft_id генерируется один раз на стрим. Я использал timestamp в миллисекундах с обрезкой до int32:

draft_id = int(time.time() * 1000) % 2147483647

Управление частотой обновлений

Отправлять sendMessageDraft на каждый чанк - плохая идея. Telegram зарейтлимитит. Поэтому обновляем черновик не чаще, чем раз в N миллисекунд:

now = time.monotonic()
if now - last_draft_ts >= config.draft_interval:
    await api.arequest("sendMessageDraft", {...})
    last_draft_ts = now

Значения DRAFT_INTERVAL_MS, CHUNK_SIZE, CHUNK_DELAY_MS в коде - подобраны на глаз. Подкрутите под свои условия.

Полный код

Один файл. Никаких зависимостей кроме стандартной библиотеки Python.

import asyncio
import json
import time
from dataclasses import dataclass
from typing import AsyncIterator, Callable
from urllib import error, request


DEFAULT_BASE_URL = "https://api.telegram.org"
TOKEN = ""           # ВАШ ТОКЕН БОТА
DRAFT_INTERVAL_MS = 80
CHUNK_SIZE = 6
CHUNK_DELAY_MS = 30


@dataclass
class AppConfig:
    token: str
    request_timeout: int = 35
    poll_timeout: int = 30
    poll_interval: float = 0.2
    draft_interval: float = 0.08
    stream_chunk_size: int = 6
    stream_chunk_delay: float = 0.03


class TelegramApi:
    def __init__(self, config: AppConfig):
        self.config = config
        self.base_url = (
            f"{DEFAULT_BASE_URL}/bot{config.token}" if config.token else ""
        )

	def request(self, method: str, payload: dict):  
	    url = f"{self.base_url}/{method}"  
	    data = json.dumps(payload).encode("utf-8")  
	    req = request.Request(  
	        url=url,  
	        data=data,  
	        headers={"Content-Type": "application/json"},  
	        method="POST",  
	    )  
	    with request.urlopen(req, timeout=self.config.request_timeout) as response:  
	        raw_body = response.read().decode("utf-8")  
	        body = json.loads(raw_body)  
	  
	    if not body.get("ok"):  
	        print(f"[telegram] api error in {method}: {body.get('description')}")  
	        return None  
	    return body.get("result")  
	  
	async def arequest(self, method: str, payload: dict):  
	    return await asyncio.to_thread(self.request, method, payload)


StreamFactory = Callable[[str, AppConfig], AsyncIterator[str]]


#---------ЗАГЛУШКА-----------
async def fake_llm_stream(
    prompt: str, config: AppConfig
) -> AsyncIterator[str]:
    """Демо-стрим. Замените на реальный SDK вашей LLM."""
    text = (
        f"Запрос: {prompt}\n\n"
        "Это демонстрация стримингового ответа. "
        "Текст появляется частями, draft в Telegram "
        "обновляется на лету. После завершения "
        "отправляется финальное сообщение."
    )
    for i in range(0, len(text), config.stream_chunk_size):
        yield text[i : i + config.stream_chunk_size]
        await asyncio.sleep(config.stream_chunk_delay)
#---------ЗАГЛУШКА-----------

async def stream_with_draft(
    api: TelegramApi,
    chat_id: int,
    prompt: str,
    config: AppConfig,
    stream_factory: StreamFactory = fake_llm_stream,
) -> str:
    draft_id = int(time.time() * 1000) % 2147483647
    full_text = ""
    last_draft_ts = 0.0

    async for chunk in stream_factory(prompt, config):
        if not chunk:
            continue
        full_text += chunk
        now = time.monotonic()

        if now - last_draft_ts >= config.draft_interval:
            await api.arequest(
                "sendMessageDraft",
                {
                    "chat_id": chat_id,
                    "draft_id": draft_id,
                    "text": full_text,
                },
            )
            last_draft_ts = now

    if full_text:
        await api.arequest(
            "sendMessageDraft",
            {"chat_id": chat_id, "draft_id": draft_id, "text": full_text},
        )
        await api.arequest(
            "sendMessage",
            {"chat_id": chat_id, "text": full_text},
        )
    return full_text


async def poll_updates(api: TelegramApi, config: AppConfig):
    print("Polling started. Send a message to your bot.")
    offset = 0
    active_tasks: set[asyncio.Task] = set()

    while True:
        updates = await api.arequest(
            "getUpdates",
            {
                "offset": offset,
                "timeout": config.poll_timeout,
                "allowed_updates": ["message"],
            },
        )
        if updates and isinstance(updates, list):
            for update in updates:
                offset = update["update_id"] + 1
                msg = update.get("message") or {}
                text = msg.get("text")
                chat_id = (msg.get("chat") or {}).get("id")
                if not chat_id or not text:
                    continue

                await api.arequest(
                    "sendChatAction",
                    {"chat_id": chat_id, "action": "typing"},
                )
                task = asyncio.create_task(
                    stream_with_draft(api, chat_id, text, config)
                )
                active_tasks.add(task)
                task.add_done_callback(active_tasks.discard)

        await asyncio.sleep(config.poll_interval)


async def main():
    config = AppConfig(
        token=TOKEN,
        draft_interval=max(DRAFT_INTERVAL_MS, 0) / 1000.0,
        stream_chunk_size=max(CHUNK_SIZE, 1),
        stream_chunk_delay=max(CHUNK_DELAY_MS, 0) / 1000.0,
    )
    api = TelegramApi(config)
    await poll_updates(api, config)


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

Нюансы

• parse_mode параметр используйте только в финальной отправке сообщения, чтобы телега не ругалась на незакрытые тэги.

• В коде на LLM стрим стоит заглушка, ее не используйте.

• draft_id должен быть уникальным для каждого стрима, но одинаковым для всех обновлений внутри одного стрима

• Финальный sendMessage обязателен. Черновик - это черновик. Без финального сообщения текст не сохранится в истории чата, а просто пропадет

• Анимированные точки добавляет сам телеграмм, пока черновик обновляется. Ничего дополнительного делать не нужно. Но можно и закостылить какой-нибудь мигающий курсор, но нафиг надо, если за нас уже все сделали.

sendMessageDraft - классное простенько обновление API. С которым теперь можно хорошие интеграции с LLM делать, а также и в игрушках, например, применять. Думаю скоро много ботов внедрят это. 

Поддержка

Если материал оказался полезным или просто зацепил — приглашаю в канал "На Дерево"