Bots Telegram auto-hébergés

agent-framework-hosting-telegram convertit les mises à jour de l’API Telegram Bot en valeurs d’exécution d’Agent Framework et convertit les exécutions finales ou en continu en opérations de l’API Bot. Il ne fournit pas de client de bot, d’environnement d’exécution pour l’interrogation, de routeur de webhooks, de registre de commandes ni d’infrastructure de distribution.

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

Utilisez n’importe quelle bibliothèque cliente pour Telegram capable de fournir un payload de mise à jour et d’exécuter les opérations renvoyées par les fonctions utilitaires. L’exemple utilise aiogram, mais les fonctions utilitaires n’y sont pas associées.

Traiter une mise à jour

L’exemple de webhook aiogram vérifie l’en-tête secret de Telegram, achemine la mise à jour et utilise un identifiant de session propre au bot pour préserver la session d’un agent pour chaque conversation privée ou chaque conversation de groupe partagée.

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)

Pour la scrutation et la configuration du webhook, la gestion des commandes, la politique relative aux médias entrants, les modifications du streaming et les conseils pour le déploiement en production, consultez l’exemple local Telegram.

Étapes suivantes

Aller plus loin :

• Vue d’ensemble de l’auto-hébergement
• Réponses OpenAI
• MCP
• Agents hébergés dans Foundry