1 марта 2026 Telegram добавил в Bot API метод sendMessageDraft - возможность потокового вывода сообщений. Тот самый эффект, к которому все привыкли в ChatGPT и Claude. Текст появляется по частям, в конце бегают анимированные точки, и ты видишь, что ответ ещё генерируется.
Я написал простенький рабочий пример на чистом Python - без каких-либо фреймворков. Только asyncio и urllib.
Как это работает
Принцип простой:
Бот создаёт черновик сообщения через
sendMessageDraftПо мере получения чанков обновляет этот черновик
Пока идёт стрим - Telegram показывает анимированные точки в конце текста
Когда стрим завершён - бот отправляет финальное сообщение через обычный
sendMessage
Черновик закрепляется в чате и обновляется в реальном времени. Пользователь видит, как текст печатается.
sendMessageDraft
await api.arequest( "sendMessageDraft", {"chat_id": chat_id, "draft_id": draft_id, "text": accumulated_text}, )
Параметры:
chat_id- куда отправляемdraft_id- уникальный идентификатор черновика (int). Все обновления одного стрима должны идти с однимdraft_idtext- текущий накопленный текст
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 делать, а также и в игрушках, например, применять. Думаю скоро много ботов внедрят это.
Поддержка
Если материал оказался полезным или просто зацепил — приглашаю в канал "На Дерево"
