Bot Telegram yang di-host sendiri

Note

Pembantu hosting mandiri untuk bot Telegram di .NET akan segera hadir.

Note

Pembantu hosting mandiri untuk bot Telegram saat ini tidak tersedia untuk Go.

agent-framework-hosting-telegram mengonversi pembaruan Telegram Bot API menjadi nilai eksekusi Agent Framework dan merender eksekusi akhir atau streaming sebagai operasi Bot API. Ini tidak menyediakan klien bot, runtime polling, router webhook, registri perintah, atau kerangka kerja pengiriman.

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

Gunakan pustaka klien Telegram apa pun yang dapat menyediakan payload pembaruan dan menjalankan operasi yang dikembalikan oleh pembantu. Sampel menggunakan aiogram, tetapi pembantu tidak terikat dengannya.

Memproses pembaruan

Sampel aiogram webhook memverifikasi header rahasia Telegram, meneruskan pembaruan, dan menggunakan ID sesi dalam cakupan bot untuk mempertahankan sesi agen untuk setiap obrolan privat atau obrolan grup bersama.

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)

Untuk polling dan penyiapan webhook, penanganan perintah, kebijakan media masuk, pengeditan streaming, dan panduan penyebaran produksi, lihat sampel Telegram lokal.

Important

Verifikasi pengiriman webhook Telegram sebelum memproses pembaruan. Rahasia webhook mengautentikasi pengiriman Telegram, tetapi tidak mengotorisasi pengguna Telegram atau obrolan untuk mengakses data aplikasi. Perlakukan obrolan dan ID pengguna sebagai tidak tepercaya sampai aplikasi Anda menerapkan kebijakan otorisasinya.

Langkah berikutnya

Tambahkan A2A

Masuk lebih dalam:

  • Last updated on