Свой Telegram-бот на API

В приложении есть мастер: Главная → Создать бота. Он поднимает Telegram-бота под вашим токеном за минуту, даёт выбрать роль ассистента и дальше не требует ни строчки кода. Если задача — «чтобы коллеги спрашивали по нашим документам в чате», начните с него, а сюда возвращайтесь, когда упрётесь.

Когда нужен свой

Свой бот оправдан, если требуется:

собственный сценарий диалога — кнопки, формы, шаги, состояния;
своя логика доступа: кто из участников что может;
связь с вашими системами — CRM, тикеты, учёт;
не Telegram вовсе: Slack, Mattermost, почта, виджет на сайте.

Всё это — обычный клиент нашего API. Облако остаётся тем же: ваш бот и бот из приложения работают с одними файлами.

Скелет

import asyncio
from contextlib import aclosing

from agentum_cloud import AgentumClient

CLOUD = AgentumClient(base_url="https://api.cloud.agentums.ru", api_key="ak_…")


async def on_document(chat_id: int, filename: str, data: bytes) -> str:
    """Пользователь прислал файл: кладём в облако и отвечаем, когда он готов."""
    obj = await CLOUD.upload(filename, data)
    await CLOUD.wait_until_ready(obj.id)
    summary = await CLOUD.get_summary(obj.id)
    return f"Загрузил «{obj.filename}». {summary.summary}"


async def on_question(chat_id: int, text: str, session_id: str | None) -> tuple[str, str | None]:
    """Вопрос по документам: ответ с источниками, разговор продолжается session_id."""
    answer = await CLOUD.ask(text, session_id=session_id)
    sources = ", ".join(c.filename for c in answer.citations)
    reply = answer.answer + (f"\n\nИсточники: {sources}" if sources else "")
    return reply, str(answer.session_id) if answer.session_id else None

session_id храните рядом с идентификатором чата — это и есть память разговора. Без него каждая реплика отвечается с нуля.

Если нужен агент, а не ответы

/v1/agent/chat умеет действовать: конвертировать, собирать PDF, заводить поручения. Он отдаёт поток, и в мессенджере его удобно показывать редактированием одного сообщения — по кадрам tool_start видно, чем агент занят прямо сейчас.

async with aclosing(CLOUD.agent_chat(text, session_id=session_id)) as stream:
    async for ev in stream:
        if ev.type == "tool_start":
            await edit_message(chat_id, msg_id, f"⏳ {ev.name}…")
        elif ev.type == "token":
            buffer += ev.delta
        elif ev.type == "paused":
            await ask_user_to_confirm(chat_id, ev)   # поток на этом кончился
            break

Ограничение платформы, о которое спотыкаются все

Telegram не позволяет боту написать пользователю первым: пока человек сам не начал диалог с ботом, отправить ему сообщение нельзя. Любые уведомления, напоминания и «поручение назначено» работают только для тех, кто хотя бы раз нажал Start. Закладывайте это в сценарий: сначала подписка, потом рассылка.

Что почитать дальше

Как устроено облако — асинхронность загрузки и скрытые классы файлов
Ошибки и повторы — что повторять, а что бессмысленно
Справочник API