Telegram-боты с собственным хостингом

Note

Инструменты для самостоятельного размещения ботов Telegram на .NET скоро появятся.

Note

Средства для самостоятельного хостинга ботов Telegram в настоящее время недоступны для Go.

agent-framework-hosting-telegram преобразует обновления Telegram Bot API в значения запуска Agent Framework и преобразует окончательные запуски или потоки выполнения в операции Bot API. Он не предоставляет клиент бота, среду выполнения опроса, маршрутизатор веб-перехватчика, реестр команд или платформу доставки.

pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-telegram azure-identity

Используйте любую клиентскую библиотеку Telegram, способную предоставить данные обновления и выполнять операции, возвращаемые вспомогательными функциями. В примере используется aiogram, но вспомогательные функции не привязаны к нему.

Обработка обновления

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

async def handle_update(update: Mapping[str, Any]) -> None:
    """Process one Telegram update through the sample agent."""
    callback_query_id = telegram_callback_query_id(update)
    if callback_query_id is not None:
        await bot.answer_callback_query(callback_query_id=callback_query_id)

    chat_id = telegram_chat_id(update)
    session_id = telegram_session_id(update, bot_id=bot.id)
    if chat_id is None or session_id is None:
        return

    # Background webhook tasks may overlap. Serialize each chat so /new cannot
    # delete a session while an earlier response is still updating it.
    async with session_locks.setdefault(session_id, asyncio.Lock()):
        if (command := telegram_command(update)) is not None and await handle_command(update, command):
            return

        async def resolve_file_url(file_id: str) -> str | None:
            file = await bot.get_file(file_id)
            if file.file_path is None or (file.file_size is not None and file.file_size > MAX_MEDIA_BYTES):
                return None
            destination = BytesIO()
            await bot.download_file(file.file_path, destination=destination)
            data = destination.getvalue()
            if len(data) > MAX_MEDIA_BYTES:
                return None
            encoded = base64.b64encode(data).decode("ascii")
            return f"data:application/octet-stream;base64,{encoded}"

        try:
            run = await telegram_to_run(update, resolve_file_url=resolve_file_url, stream=True)
        except ValueError:
            LOGGER.debug("Ignoring non-actionable Telegram update", exc_info=True)
            return

        await bot.send_chat_action(chat_id=chat_id, action="typing")
        placeholder = await bot.send_message(chat_id=chat_id, text=PLACEHOLDER_TEXT)

        target = await state.get_target()
        # Reuse one AgentSession per Telegram chat. The /new command removes this
        # mapping so get_or_create_session creates a clean session next time.
        session = await state.get_or_create_session(session_id)
        stream = target.run(
            run["messages"],
            stream=True,
            session=session,
            options=run["options"],
        )
        if not isinstance(stream, ResponseStream):
            raise RuntimeError("agent did not return a response stream")

        last_edit_at = 0.0
        async for operation in telegram_from_streaming_run(
            stream,
            chat_id=chat_id,
            message_id=placeholder.message_id,
            initial_text=PLACEHOLDER_TEXT,
        ):
            if operation["method"] == "editMessageText":
                delay = EDIT_INTERVAL_SECONDS - (time.monotonic() - last_edit_at)
                if delay > 0:
                    await asyncio.sleep(delay)
                last_edit_at = time.monotonic()
            await execute_operation(operation)

        # Persist the updated AgentSession back under the stable per-chat key after
        # streaming has finalized and the history provider has recorded the turn.
        await state.set_session(session_id, session)

Сведения о настройке опроса и вебхука, обработке команд, политике входящих мультимедиа, редактировании в потоковом режиме и рекомендациях по промышленному развертыванию см. в локальном примере Telegram.

Important

Проверяйте доставку вебхука Telegram перед обработкой обновлений. Секрет вебхука подтверждает подлинность запросов от Telegram, но не предоставляет пользователю или чату Telegram права доступа к данным приложения. Обрабатывать идентификаторы чата и пользователей как ненадежные, пока приложение не применяет политику авторизации.

Дальнейшие действия

Добавление A2A

Вернитесь глубже:

  • Last updated on