Bots do Telegram autoalojados

Note

As ferramentas auxiliares para alojamento próprio de bots do Telegram em .NET chegarão em breve.

Note

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

agent-framework-hosting-telegram converte atualizações da API do Telegram Bot em valores de execução do Agent Framework e apresenta execuções finais ou em streaming como operações da API do Bot. Não fornece um cliente para bots, um ambiente de execução de sondagem, um encaminhador de webhooks, um registo de comandos nem uma estrutura de entrega.

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

Utilize qualquer biblioteca cliente para o Telegram que possa fornecer um payload de atualização e executar as operações devolvidas pelas funções auxiliares. O exemplo usa aiogram, mas as funções auxiliares não estão associadas a ele.

Processar uma atualização

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

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 sondagens e configuração de webhooks, gestão de comandos, política de media de entrada, edições em streaming e orientações para implementação em produção, consulte o exemplo local do Telegram.

Importante

Verifique as entregas de webhooks do Telegram antes de processar as atualizações. Um webhook secreto autentica a entrega do Telegram, mas não autoriza o utilizador ou chat do Telegram a aceder aos dados da aplicação. Trate os IDs do chat e dos utilizadores como não confiáveis até que a sua aplicação aplique a sua política de autorização.

Passos seguintes

Adicionar A2A

Vai mais fundo:

  • Last updated on