Hospede você mesmo bots do Telegram

Note

As ferramentas de hospedagem própria para bots do Telegram em .NET estarão disponíveis em breve.

Note

Os auxiliares de auto-hospedagem para bots do Telegram não estão disponíveis no momento para o Go.

agent-framework-hosting-telegram converte atualizações da Telegram Bot API em valores de execução do Agent Framework e renderiza execuções finais ou em streaming como operações da Bot API. Ele não fornece um cliente de bot, ambiente de execução para polling, roteador de webhooks, registro de comandos nem framework de entrega.

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

Use qualquer biblioteca cliente do Telegram que seja capaz de fornecer um payload de atualização e executar as operações retornadas pelas funções auxiliares. O exemplo usa aiogram, mas os auxiliares não estão vinculados a ele.

Processar uma atualização

O aiogram exemplo de webhook verifica o cabeçalho secreto do Telegram, envia a atualização e usa uma ID de sessão com escopo de bot para preservar uma sessão de agente para cada chat privado ou chat em grupo compartilhado.

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)

Para configuração de polling e webhook, processamento de comandos, política para mídia de entrada, edição em streaming e orientações para implantação em produção, consulte o exemplo local do Telegram.

Importante

Verifique as entregas do webhook do Telegram antes de processar atualizações. Um segredo de webhook autentica a entrega feita pelo Telegram, mas não autoriza o usuário do Telegram nem o chat a acessar dados do aplicativo. Trate as IDs de chat e de usuário como não confiáveis até que seu aplicativo aplique sua política de autorização.

Próximas Etapas 

Adicionar A2A

Vá mais fundo:

  • Last updated on