Bot di Telegram ospitati autonomamente

agent-framework-hosting-telegram converte gli aggiornamenti dell'API Bot di Telegram in valori di esecuzione di Agent Framework e trasforma le esecuzioni finali o in streaming in operazioni dell'API Bot. Non fornisce un client per bot, un runtime di polling, un router di webhook, un registro dei comandi o un framework di consegna.

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

Utilizza qualsiasi libreria client Telegram in grado di fornire un payload di aggiornamento ed eseguire le operazioni restituite dagli helper. L'esempio usa aiogram, ma gli helper non sono legati ad esso.

Elaborare un aggiornamento

L'esempio aiogram di webhook verifica l'intestazione del segreto di Telegram, invia l'aggiornamento e usa un ID sessione con ambito bot per mantenere una sessione agente per ogni chat privata o chat di gruppo condiviso.

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)

Per la configurazione di polling e webhook, la gestione dei comandi, i criteri per i contenuti multimediali in ingresso, le modifiche in streaming e le linee guida per la distribuzione in produzione, consulta l'esempio locale di Telegram.

Passaggi successivi

Approfondimento:

• Panoramica dell'hosting autonomo
• Risposte OpenAI
• MCP
• Agenti ospitati di Foundry