Boty Telegrama hostowane samodzielnie

Uwaga / Notatka

Narzędzia do samodzielnego hostowania botów Telegrama w środowisku .NET będą wkrótce dostępne.

Uwaga / Notatka

Narzędzia do samodzielnego hostowania botów Telegrama nie są obecnie dostępne w języku Go.

agent-framework-hosting-telegram przekształca aktualizacje Telegram Bot API w wartości uruchomień platformy Agent Framework oraz przekształca końcowe lub strumieniowane uruchomienia w operacje Bot API. Nie udostępnia klienta bota, środowiska uruchomieniowego do odpytywania, routera webhooków, rejestru poleceń ani mechanizmu dostarczania.

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

Użyj dowolnej biblioteki klienckiej Telegrama, która może dostarczyć dane aktualizacji i wykonać operacje zwrócone przez funkcje pomocnicze. W przykładzie użyto elementu aiogram, ale funkcje pomocnicze nie są z nim powiązane.

Przetwórz aktualizację

Przykład aiogram webhooka weryfikuje tajny nagłówek Telegrama, przekazuje aktualizację dalej i używa identyfikatora sesji ograniczonego do bota, aby zachować sesję agenta dla każdego czatu prywatnego lub współdzielonego czatu grupowego.

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:
            username = (await bot.me()).username
            if not username or telegram_command(update, bot_username=username) is None:
                return
            if 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)

Przed wysłaniem polecenia przekaż bieżącą nazwę użytkownika bota do telegram_command(update, bot_username=...). Funkcja pomocnicza zwraca wartość None dla poleceń adresowanych do innego bota, w tym poleceń w tekście wiadomości, danych wywołania zwrotnego i podpisach multimediów. Pominięcie bot_username zachowuje parsowanie niezależne od celu dla zachowania zgodności.

Aby uzyskać informacje na temat konfiguracji odpytywania i webhooka, obsługi poleceń, zasad dotyczących multimediów przychodzących, edycji strumieniowych oraz wskazówek dotyczących wdrożenia produkcyjnego, zobacz lokalny przykład Telegrama.

Important

Przed przetworzeniem aktualizacji sprawdź dostawy elementu webhook usługi Telegram. Sekret webhooka uwierzytelnia dostarczanie przez Telegram, ale nie autoryzuje użytkownika ani czatu Telegrama do uzyskiwania dostępu do danych aplikacji. Traktuj identyfikatory czatów i użytkowników jako niezaufane, dopóki aplikacja nie zastosuje zasad autoryzacji.

Następne kroki

Dodawanie A2A

Głębiej:

  • Ostatnia aktualizacja o