Bots de Telegram autoalojados

Note

Las herramientas de autoalojamiento para bots de Telegram en .NET llegarán pronto.

Note

Las utilidades de alojamiento propio para bots de Telegram en Go no están disponibles actualmente.

agent-framework-hosting-telegram convierte las actualizaciones de Telegram Bot API en valores de ejecución de Agent Framework y representa las ejecuciones finales o de streaming como operaciones de Bot API. No proporciona un cliente de bot, un entorno de ejecución de sondeo, un enrutador de webhook, un registro de comandos ni un marco de entrega.

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

Use cualquier biblioteca cliente de Telegram que pueda proporcionar una carga de actualización y ejecutar las operaciones devueltas por los asistentes. En el ejemplo se usa aiogram, pero las funciones auxiliares no están ligadas a él.

Procesar una actualización

El aiogram ejemplo de webhook verifica la cabecera secreta de Telegram, redirige la actualización y usa un identificador de sesión asociado al bot para mantener una sesión de agente para cada chat privado o chat de grupo compartido.

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 la configuración del sondeo y del webhook, la gestión de comandos, la política de medios entrantes, las ediciones en streaming y la guía para el despliegue en producción, consulte el ejemplo local de Telegram.

Importante

Compruebe las entregas de webhook de Telegram antes de procesar las actualizaciones. Un secreto de webhook autentica el envío de Telegram, pero no autoriza al usuario ni al chat de Telegram a acceder a los datos de la aplicación. Trate los identificadores de usuario y chat como que no son de confianza hasta que la aplicación aplique su directiva de autorización.

Pasos siguientes

Agregar A2A

Vaya más profundamente:

  • Last updated on