[Plugin] Add Telegram Plugin

plugins/telegram/README.md

@@ -0,0 +1,130 @@
+# Telegram Plugin for GAME SDK
+
+## Overview
+
+The **Telegram Plugin** is an integration for the **Game SDK** that enables AI-driven interactions on Telegram. This plugin allows AI agents to handle messages, execute commands, and engage users through text, media, and polls.
+
+## Features
+
+- **Send Messages** – AI agents can send text messages to users.
+- **Send Media** – Supports sending photos, documents, videos, and audio.
+- **Create Polls** – AI agents can generate interactive polls.
+- **Pin & Unpin Messages** – Manage pinned messages in chats.
+- **Delete Messages** – Remove messages dynamically.
+- **AI-Powered Responses** – Leverages LLM to generate contextual replies.
+- **Real-Time Polling** – Runs asynchronously with Telegram’s polling system.
+- and more features to come!
+
+## Installation
+### Pre-requisites
+Ensure you have Python 3.9+ installed. Then, install the plugin via **PyPI**:
+### Steps
+1. Install the plugin:
+   ```sh bash
+   pip install telegram-plugin-gamesdk
+   ```
+2. Ensure you have a Telegram bot token and GAME API key and set them as environment variables:
+   ```sh bash
+   export TELEGRAM_BOT_TOKEN="your-telegram-bot-token"
+   export GAME_API_KEY="your-game-api-key"
+   ```
+3. Refer to the example and run the example bot:
+   ```sh bash
+   python examples/test_telegram.py
+   ```
+
+## Usage Examples
+### Initializing the Plugin
+
+```python
+from telegram_plugin_gamesdk.telegram_plugin import TelegramPlugin
+
+tg_bot = TelegramPlugin(bot_token='your-telegram-bot-token')
+tg_bot.start_polling()
+```
+
+### Sending a Message
+```python
+tg_bot.send_message(chat_id=123456789, text="Hello from the AI Agent!")
+```
+
+### Sending Media
+```python
+tg_bot.send_media(chat_id=123456789, media_type="photo", media="https://example.com/image.jpg", caption="Check this out!")
+```
+
+### Creating a Poll
+```python
+tg_bot.create_poll(chat_id=123456789, question="What's your favorite color?", options=["Red", "Blue", "Green"])
+```
+
+### Pinning and Unpinning Messages
+```python
+tg_bot.pin_message(chat_id=123456789, message_id=42)
+tg_bot.unpin_message(chat_id=123456789, message_id=42)
+```
+
+### Deleting a Message
+```python
+tg_bot.delete_message(chat_id=123456789, message_id=42)
+```
+
+## Integration with GAME Chat Agent
+Implement a message handler to integrate the Telegram Plugin with the GAME Chat Agent:
+```python
+from telegram import Update
+from telegram.ext import ContextTypes, filters, MessageHandler
+from game_sdk.game.chat_agent import ChatAgent
+
+chat_agent = ChatAgent(
+    prompt="You are a helpful assistant.",
+    api_key="your-game-api-key",
+)
+
+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)"
+
+    if not any(u["chat_id"] == chat_id for u in active_users):
+        # 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
+        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))
+```
+You can refer to [test_telegram.py](examples/test_telegram.py) for details.

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

plugins/telegram/plugin_metadata.yml

@@ -0,0 +1,14 @@
+# General Information
+plugin_name: "telegram_plugin_gamesdk"
+author: "Yang from Virtuals Protocol"
+logo_url: "https://drive.google.com/drive/folders/1AdRKQac0a-ORqdE6mtV4wGdF8y3DhIyo"
+release_date: "2025-03"
+
+# Description
+short_description: "Telegram Plugin for Game SDK"
+detailed_description: "This plugin provides a Telegram integration for the Game SDK. It allows developers to easily integrate Telegram functionalities like sending, deleting, pin, messages or media files etc. Credits to the yk(yenkhoon) for building the Telegram plugin for GAME Node SDK."
+
+# Contact & Support
+x_account_handle: "@GAME_Virtuals"
+support_contact: "https://discord.gg/virtualsio"
+community_link: "https://t.me/virtuals"

plugins/telegram/pyproject.toml

@@ -0,0 +1,16 @@
+[project]
+name = "telegram-plugin-gamesdk"
+version = "0.1.0"
+description = "Telegram Plugin for Python SDK for GAME by Virtuals"
+authors = [
+    {name = "Ang Weoy Yang", email = "[email protected]"}
+]
+readme = "README.md"
+requires-python = ">=3.9"
+dependencies = [
+    "python-telegram-bot>=21.11.1",
+]
+
+[build-system]
+requires = ["poetry-core>=2.0.0,<3.0.0"]
+build-backend = "poetry.core.masonry.api"