chore: add examples for telegram plugin

Author: Ang-dot

plugins/telegram/examples/test_telegram.py

@@ -0,0 +1,98 @@
+import os
+from typing import TypedDict
+import logging
+
+from telegram import Update
+from telegram.ext import ContextTypes, filters, MessageHandler
+
+from game_sdk.game.chat_agent import Chat, ChatAgent
+from telegram_plugin_gamesdk.telegram_plugin import TelegramPlugin
+from test_telegram_game_functions import send_message_fn, send_media_fn, create_poll_fn, pin_message_fn, unpin_message_fn, delete_message_fn
+
+game_api_key = os.environ.get("GAME_API_KEY")
+telegram_bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    level=logging.INFO,
+)
+
+logger = logging.getLogger(__name__)
+
+class ActiveUser(TypedDict):
+    chat_id: int
+    name: str
+
+chat_agent = ChatAgent(
+    prompt="You are a helpful assistant.",
+    api_key=game_api_key,
+)
+
+active_users: list[ActiveUser] = []
+active_chats: dict[int, Chat] = {}
+
+if __name__ == "__main__":
+    tg_plugin = TelegramPlugin(bot_token=telegram_bot_token)
+
+    agent_action_space = [
+        send_message_fn(tg_plugin),
+        send_media_fn(tg_plugin),
+        create_poll_fn(tg_plugin),
+        pin_message_fn(tg_plugin),
+        unpin_message_fn(tg_plugin),
+        delete_message_fn(tg_plugin),
+    ]
+
+    async def default_message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
+        """Handles incoming messages but ignores messages from the bot itself unless it's mentioned in a group chat."""
+
+        # Ignore messages from the bot itself
+        if update.message.from_user.id == tg_plugin.bot.id:
+            logger.info("Ignoring bot's own message.")
+            return
+
+        user = update.message.from_user
+        chat_id = update.message.chat.id
+        chat_type = update.message.chat.type  # "private", "group", "supergroup", or "channel"
+        bot_username = f"@{tg_plugin.bot.username}"
+
+        logger.info(f"Update received: {update}")
+        logger.info(f"Message received: {update.message.text}")
+
+        name = f"{user.first_name} (Telegram's chat_id: {chat_id}, this is not part of the partner's name but important for the telegram's function arguments)"
+
+        # Ignore group/supergroup messages unless the bot is mentioned
+        if chat_type in ["group", "supergroup"] and bot_username not in update.message.text:
+            logger.info(f"Ignoring group message not mentioning the bot: {update.message.text}")
+            return
+
+        if not any(u["chat_id"] == chat_id for u in active_users):
+            active_users.append({"chat_id": chat_id, "name": name})
+            logger.info(f"Active user added: {name}")
+            logger.info(f"Active users: {active_users}")
+            chat = chat_agent.create_chat(
+                partner_id=str(chat_id),
+                partner_name=name,
+                action_space=agent_action_space,
+            )
+            active_chats[chat_id] = chat
+
+        response = active_chats[chat_id].next(update.message.text.replace(bot_username, "").strip())  # Remove bot mention
+        logger.info(f"Response: {response}")
+
+        if response.message:
+            await update.message.reply_text(response.message)
+
+        if response.is_finished:
+            active_chats.pop(chat_id)
+            active_users.remove({"chat_id": chat_id, "name": name})
+            logger.info(f"Chat with {name} ended.")
+            logger.info(f"Active users: {active_users}")
+
+    tg_plugin.add_handler(MessageHandler(filters.ALL, default_message_handler))
+
+    # Start polling
+    tg_plugin.start_polling()
+
+    # Example of executing a function from Telegram Plugin to a chat without polling
+    #tg_plugin.send_message(chat_id=829856292, text="Hello! I am a helpful assistant. How can I assist you today?")

plugins/telegram/examples/test_telegram_game_functions.py

@@ -0,0 +1,123 @@
+import os
+import sys
+from typing import Tuple
+
+project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../..'))
+sys.path.append(project_root)
+
+from plugins.telegram.telegram_plugin_gamesdk.telegram_plugin import TelegramPlugin
+from src.game_sdk.game.custom_types import Function, FunctionResultStatus, Argument
+
+
+def send_message_executable(tg_plugin: TelegramPlugin, chat_id: str, text: str) -> Tuple[FunctionResultStatus, str, dict]:
+    try:
+        tg_plugin.send_message(chat_id=chat_id, text=text)
+        return FunctionResultStatus.DONE, "Message sent successfully", {}
+    except Exception as e:
+        return FunctionResultStatus.FAILED, str(e), {}
+
+def send_message_fn(bot: TelegramPlugin) -> Function:
+    return Function(
+        fn_name="send_message",
+        fn_description="Send a text message to a Telegram chat",
+        args=[
+            Argument(name="chat_id", description="ID of the chat to send the message to", type="str"),
+            Argument(name="text", description="Text message to send", type="str"),
+        ],
+        executable=lambda chat_id, text: send_message_executable(bot, chat_id, text),
+    )
+
+def send_media_executable(tg_plugin: TelegramPlugin, chat_id: str, media_type: str, media: str, caption: str = None) -> Tuple[FunctionResultStatus, str, dict]:
+    try:
+        tg_plugin.send_media(chat_id=chat_id, media_type=media_type, media=media, caption=caption)
+        return FunctionResultStatus.DONE, "Media sent successfully", {}
+    except Exception as e:
+        return FunctionResultStatus.FAILED, str(e), {}
+
+def send_media_fn(bot: TelegramPlugin) -> Function:
+    return Function(
+        fn_name="send_media",
+        fn_description="Send a media message to a Telegram chat",
+        args=[
+            Argument(name="chat_id", description="ID of the chat to send the message to", type="str"),
+            Argument(name="media_type", description="Type of media to send (photo, document, video, audio)", type="str"),
+            Argument(name="media", description="Media URL or file path to send", type="str"),
+            Argument(name="caption", description="Optional caption for the media", type="str", optional=True),
+        ],
+        executable=lambda chat_id, media_type, media, caption=None: send_media_executable(bot, chat_id, media_type, media, caption),
+    )
+
+def create_poll_executable(tg_plugin: TelegramPlugin, chat_id: str, question: str, options: list[str], is_anonymous: bool = True, allows_multiple_answers: bool = False) -> Tuple[FunctionResultStatus, str, dict]:
+    try:
+        tg_plugin.create_poll(chat_id=chat_id, question=question, options=options, is_anonymous=is_anonymous, allows_multiple_answers=allows_multiple_answers)
+        return FunctionResultStatus.DONE, "Poll created successfully", {}
+    except Exception as e:
+        return FunctionResultStatus.FAILED, str(e), {}
+
+def create_poll_fn(bot: TelegramPlugin) -> Function:
+    return Function(
+        fn_name="create_poll",
+        fn_description="Create a poll in a Telegram chat",
+        args=[
+            Argument(name="chat_id", description="ID of the chat to create the poll in", type="str"),
+            Argument(name="question", description="Question to ask in the poll", type="str"),
+            Argument(name="options", description="List of options for the poll", type="List[str]"),
+            Argument(name="is_anonymous", description="Whether the poll is anonymous (default: True)", type="bool", optional=True),
+            Argument(name="allows_multiple_answers", description="Whether multiple answers are allowed (default: False)", type="bool", optional=True),
+        ],
+        executable=lambda chat_id, question, options, is_anonymous=False, allows_multiple_answers=False: create_poll_executable(bot, chat_id, question, options, is_anonymous, allows_multiple_answers),
+    )
+
+def pin_message_executable(tg_plugin: TelegramPlugin, chat_id: str, message_id: int) -> Tuple[FunctionResultStatus, str, dict]:
+    try:
+        tg_plugin.pin_message(chat_id=chat_id, message_id=message_id)
+        return FunctionResultStatus.DONE, "Message pinned successfully", {}
+    except Exception as e:
+        return FunctionResultStatus.FAILED, str(e), {}
+
+def pin_message_fn(bot: TelegramPlugin) -> Function:
+    return Function(
+        fn_name="pin_message",
+        fn_description="Pin a message in a Telegram chat",
+        args=[
+            Argument(name="chat_id", description="ID of the chat to pin the message in", type="str"),
+            Argument(name="message_id", description="ID of the message to pin", type="int"),
+        ],
+        executable=lambda chat_id, message_id: pin_message_executable(bot, chat_id, message_id),
+    )
+
+def unpin_message_executable(tg_plugin: TelegramPlugin, chat_id: str, message_id: int) -> Tuple[FunctionResultStatus, str, dict]:
+    try:
+        tg_plugin.unpin_message(chat_id=chat_id, message_id=message_id)
+        return FunctionResultStatus.DONE, "Message unpinned successfully", {}
+    except Exception as e:
+        return FunctionResultStatus.FAILED, str(e), {}
+
+def unpin_message_fn(bot: TelegramPlugin) -> Function:
+    return Function(
+        fn_name="unpin_message",
+        fn_description="Unpin a message in a Telegram chat",
+        args=[
+            Argument(name="chat_id", description="ID of the chat to unpin the message in", type="str"),
+            Argument(name="message_id", description="ID of the message to unpin", type="int"),
+        ],
+        executable=lambda chat_id, message_id: unpin_message_executable(bot, chat_id, message_id),
+    )
+
+def delete_message_executable(tg_plugin: TelegramPlugin, chat_id: str, message_id: int) -> Tuple[FunctionResultStatus, str, dict]:
+    try:
+        tg_plugin.delete_message(chat_id=chat_id, message_id=message_id)
+        return FunctionResultStatus.DONE, "Message deleted successfully", {}
+    except Exception as e:
+        return FunctionResultStatus.FAILED, str(e), {}
+
+def delete_message_fn(bot: TelegramPlugin) -> Function:
+    return Function(
+        fn_name="delete_message",
+        fn_description="Delete a message in a Telegram chat",
+        args=[
+            Argument(name="chat_id", description="ID of the chat to delete the message in", type="str"),
+            Argument(name="message_id", description="ID of the message to delete", type="int"),
+        ],
+        executable=lambda chat_id, message_id: delete_message_executable(bot, chat_id, message_id),
+    )