Boty Telegrama hostowane samodzielnie

Note

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

Note

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 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)

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.

Ważna

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:

  • Last updated on