Telegram Bot for GPU Management

What We're Building

A Telegram bot that lets you browse GPUs, rent servers, monitor orders, and receive alerts — all from your phone. Perfect for managing Clore rentals on the go.

Prerequisites

• Clore.ai API key

• Telegram Bot Token (from @BotFather)

• Python 3.10+

• python-telegram-bot library

Step 1: Bot Setup

# clore_telegram_bot.py
"""Telegram bot for Clore.ai GPU management."""

import os
import logging
from datetime import datetime
from typing import Dict, List
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
    Application, CommandHandler, CallbackQueryHandler,
    ContextTypes, ConversationHandler, MessageHandler, filters
)
import requests

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# States for conversation
SELECTING_GPU, CONFIRMING_RENTAL = range(2)

class CloreBot:
    """Telegram bot for Clore.ai management."""
    
    BASE_URL = "https://api.clore.ai"
    
    def __init__(self, clore_api_key: str, telegram_token: str):
        self.clore_api_key = clore_api_key
        self.telegram_token = telegram_token
        self.headers = {"auth": clore_api_key}
        
        # User state
        self.user_selections: Dict[int, dict] = {}
    
    def _clore_request(self, method: str, endpoint: str, **kwargs) -> dict:
        """Make Clore API request."""
        url = f"{self.BASE_URL}{endpoint}"
        response = requests.request(method, url, headers=self.headers, **kwargs)
        return response.json()
    
    async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Handle /start command."""
        welcome = """
🚀 *Clore.ai GPU Manager Bot*

I can help you:
• 🔍 Browse available GPUs
• 💰 Check prices
• 🖥️ Rent servers
• 📊 Monitor your orders
• ⚡ Get alerts

*Commands:*
/browse - Browse available GPUs
/orders - View your active orders
/balance - Check wallet balance
/cancel <order_id> - Cancel an order
/help - Show this help
        """
        await update.message.reply_text(welcome, parse_mode="Markdown")
    
    async def browse(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Browse available GPUs."""
        await update.message.reply_text("🔍 Fetching available GPUs...")
        
        try:
            data = self._clore_request("GET", "/v1/marketplace")
            servers = data.get("servers", [])
            
            # Filter available servers
            available = [s for s in servers if not s.get("rented")]
            
            # Group by GPU type
            by_gpu: Dict[str, List] = {}
            for server in available:
                gpus = server.get("gpu_array", [])
                for gpu in gpus:
                    # Normalize GPU name
                    gpu_name = gpu.split()[0] + " " + gpu.split()[-1] if " " in gpu else gpu
                    if gpu_name not in by_gpu:
                        by_gpu[gpu_name] = []
                    by_gpu[gpu_name].append(server)
            
            # Build message
            message = "🖥️ *Available GPUs:*\n\n"
            
            for gpu_name, servers in sorted(by_gpu.items()):
                prices = [s.get("price", {}).get("usd", {}).get("on_demand_clore", 999) for s in servers]
                min_price = min(prices)
                count = len(servers)
                message += f"• *{gpu_name}*: {count} available from ${min_price:.2f}/hr\n"
            
            message += "\nUse /rent <gpu_type> to see specific options."
            
            await update.message.reply_text(message, parse_mode="Markdown")
            
        except Exception as e:
            await update.message.reply_text(f"❌ Error: {e}")
    
    async def rent_gpu(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Show rental options for specific GPU."""
        if not context.args:
            await update.message.reply_text("Usage: /rent <gpu_type>\nExample: /rent RTX 4090")
            return
        
        gpu_type = " ".join(context.args)
        await update.message.reply_text(f"🔍 Searching for {gpu_type}...")
        
        try:
            data = self._clore_request("GET", "/v1/marketplace")
            servers = data.get("servers", [])
            
            # Filter matching GPUs
            matching = []
            for server in servers:
                if server.get("rented"):
                    continue
                
                gpus = server.get("gpu_array", [])
                if any(gpu_type.lower() in g.lower() for g in gpus):
                    price = server.get("price", {}).get("usd", {}).get("on_demand_clore", 999)
                    matching.append({
                        "id": server["id"],
                        "gpus": gpus,
                        "price": price,
                        "reliability": server.get("reliability", 0)
                    })
            
            if not matching:
                await update.message.reply_text(f"No {gpu_type} available right now. Try /browse to see options.")
                return
            
            # Sort by price
            matching.sort(key=lambda x: x["price"])
            
            # Create inline keyboard with top 5 options
            keyboard = []
            for server in matching[:5]:
                text = f"${server['price']:.2f}/hr - {server['reliability']:.0f}% reliable"
                callback = f"rent_{server['id']}"
                keyboard.append([InlineKeyboardButton(text, callback_data=callback)])
            
            keyboard.append([InlineKeyboardButton("❌ Cancel", callback_data="cancel")])
            
            message = f"🖥️ *Available {gpu_type}:*\n\n"
            message += "Select a server to rent:\n"
            
            await update.message.reply_text(
                message,
                parse_mode="Markdown",
                reply_markup=InlineKeyboardMarkup(keyboard)
            )
            
        except Exception as e:
            await update.message.reply_text(f"❌ Error: {e}")
    
    async def handle_rental_selection(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Handle rental selection from inline keyboard."""
        query = update.callback_query
        await query.answer()
        
        if query.data == "cancel":
            await query.edit_message_text("Cancelled.")
            return
        
        if query.data.startswith("rent_"):
            server_id = int(query.data.replace("rent_", ""))
            
            # Store selection
            self.user_selections[query.from_user.id] = {"server_id": server_id}
            
            # Show confirmation
            keyboard = [
                [
                    InlineKeyboardButton("✅ Confirm", callback_data=f"confirm_{server_id}"),
                    InlineKeyboardButton("❌ Cancel", callback_data="cancel")
                ]
            ]
            
            await query.edit_message_text(
                f"🖥️ Rent server {server_id}?\n\nThis will start an on-demand rental.",
                reply_markup=InlineKeyboardMarkup(keyboard)
            )
        
        elif query.data.startswith("confirm_"):
            server_id = int(query.data.replace("confirm_", ""))
            
            await query.edit_message_text("⏳ Creating order...")
            
            try:
                order = self._clore_request("POST", "/v1/create_order", json={
                    "renting_server": server_id,
                    "type": "on-demand",
                    "currency": "CLORE-Blockchain",
                    "image": "nvidia/cuda:12.8.0-base-ubuntu22.04",
                    "ports": {"22": "tcp"},
                    "env": {"NVIDIA_VISIBLE_DEVICES": "all"},
                    "ssh_password": "TelegramRent123!"
                })
                
                order_id = order.get("order_id")
                
                message = f"""
✅ *Order Created!*

Order ID: `{order_id}`
Server: {server_id}

Your server will be ready in 1-2 minutes.
Use /orders to check status.
Use /cancel {order_id} to stop rental.
                """
                
                await query.edit_message_text(message, parse_mode="Markdown")
                
            except Exception as e:
                await query.edit_message_text(f"❌ Failed: {e}")
    
    async def orders(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Show active orders."""
        try:
            data = self._clore_request("GET", "/v1/my_orders")
            orders = data.get("orders", [])
            
            if not orders:
                await update.message.reply_text("No active orders.")
                return
            
            message = "📊 *Your Orders:*\n\n"
            
            for order in orders:
                status_emoji = {
                    "running": "🟢",
                    "creating_order": "🟡",
                    "paused": "🟠",
                    "expired": "🔴"
                }.get(order.get("status"), "⚪")
                
                message += f"{status_emoji} Order `{order['order_id']}`\n"
                message += f"   Status: {order.get('status')}\n"
                message += f"   Server: {order.get('renting_server')}\n"
                
                if order.get("connection", {}).get("ssh"):
                    message += f"   SSH: `{order['connection']['ssh']}`\n"
                
                message += "\n"
            
            await update.message.reply_text(message, parse_mode="Markdown")
            
        except Exception as e:
            await update.message.reply_text(f"❌ Error: {e}")
    
    async def balance(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Show wallet balance."""
        try:
            data = self._clore_request("GET", "/v1/wallets")
            wallets = data.get("wallets", [])
            
            message = "💰 *Wallet Balances:*\n\n"
            
            for wallet in wallets:
                name = wallet.get("name", "Unknown")
                balance = wallet.get("balance", 0)
                
                if "CLORE" in name:
                    message += f"🔹 CLORE: {balance:.2f}\n"
                elif "bitcoin" in name.lower():
                    message += f"🟠 BTC: {balance:.6f}\n"
                elif "USD" in name:
                    message += f"💵 USD: ${balance:.2f}\n"
            
            await update.message.reply_text(message, parse_mode="Markdown")
            
        except Exception as e:
            await update.message.reply_text(f"❌ Error: {e}")
    
    async def cancel_order(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Cancel an order."""
        if not context.args:
            await update.message.reply_text("Usage: /cancel <order_id>")
            return
        
        order_id = int(context.args[0])
        
        try:
            self._clore_request("POST", "/v1/cancel_order", json={"id": order_id})
            await update.message.reply_text(f"✅ Order {order_id} cancelled.")
        except Exception as e:
            await update.message.reply_text(f"❌ Failed to cancel: {e}")
    
    async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Show help."""
        await self.start(update, context)
    
    def run(self):
        """Run the bot."""
        app = Application.builder().token(self.telegram_token).build()
        
        # Command handlers
        app.add_handler(CommandHandler("start", self.start))
        app.add_handler(CommandHandler("browse", self.browse))
        app.add_handler(CommandHandler("rent", self.rent_gpu))
        app.add_handler(CommandHandler("orders", self.orders))
        app.add_handler(CommandHandler("balance", self.balance))
        app.add_handler(CommandHandler("cancel", self.cancel_order))
        app.add_handler(CommandHandler("help", self.help_command))
        
        # Callback handler for inline buttons
        app.add_handler(CallbackQueryHandler(self.handle_rental_selection))
        
        # Start bot
        logger.info("Starting Telegram bot...")
        app.run_polling()


if __name__ == "__main__":
    import sys
    
    clore_key = os.environ.get("CLORE_API_KEY") or sys.argv[1]
    telegram_token = os.environ.get("TELEGRAM_TOKEN") or sys.argv[2]
    
    bot = CloreBot(clore_key, telegram_token)
    bot.run()

Step 2: Alert System

# alerts.py
"""Alert system for Clore monitoring."""

import asyncio
import time
from datetime import datetime, timedelta
from typing import List, Dict, Callable
import requests
from telegram import Bot

class CloreAlerts:
    """Monitor Clore and send Telegram alerts."""
    
    BASE_URL = "https://api.clore.ai"
    
    def __init__(self, clore_api_key: str, telegram_token: str, chat_id: int):
        self.clore_api_key = clore_api_key
        self.telegram_token = telegram_token
        self.chat_id = chat_id
        self.headers = {"auth": clore_api_key}
        self.bot = Bot(token=telegram_token)
        
        # Track state
        self.last_order_status: Dict[int, str] = {}
        self.last_balance: Dict[str, float] = {}
        self.price_alerts: List[Dict] = []
    
    def _clore_request(self, endpoint: str) -> dict:
        url = f"{self.BASE_URL}{endpoint}"
        response = requests.get(url, headers=self.headers)
        return response.json()
    
    async def send_alert(self, message: str):
        """Send Telegram alert."""
        await self.bot.send_message(
            chat_id=self.chat_id,
            text=message,
            parse_mode="Markdown"
        )
    
    def add_price_alert(self, gpu_type: str, max_price: float):
        """Add price alert for GPU type."""
        self.price_alerts.append({
            "gpu_type": gpu_type,
            "max_price": max_price,
            "triggered": False
        })
    
    async def check_orders(self):
        """Check order status changes."""
        try:
            data = self._clore_request("/v1/my_orders")
            orders = data.get("orders", [])
            
            for order in orders:
                order_id = order["order_id"]
                status = order.get("status", "unknown")
                
                # Check for status change
                if order_id in self.last_order_status:
                    old_status = self.last_order_status[order_id]
                    if old_status != status:
                        await self.send_alert(
                            f"⚡ *Order Status Changed*\n"
                            f"Order: `{order_id}`\n"
                            f"Status: {old_status} → {status}"
                        )
                else:
                    # New order
                    await self.send_alert(
                        f"🆕 *New Order Created*\n"
                        f"Order: `{order_id}`\n"
                        f"Status: {status}"
                    )
                
                self.last_order_status[order_id] = status
                
        except Exception as e:
            print(f"Order check error: {e}")
    
    async def check_balance(self):
        """Check for low balance."""
        try:
            data = self._clore_request("/v1/wallets")
            wallets = data.get("wallets", [])
            
            for wallet in wallets:
                name = wallet["name"]
                balance = wallet["balance"]
                
                # Alert if CLORE balance drops below 10
                if "CLORE" in name and balance < 10:
                    if name not in self.last_balance or self.last_balance[name] >= 10:
                        await self.send_alert(
                            f"⚠️ *Low Balance Alert*\n"
                            f"CLORE balance: {balance:.2f}\n"
                            f"Consider topping up!"
                        )
                
                self.last_balance[name] = balance
                
        except Exception as e:
            print(f"Balance check error: {e}")
    
    async def check_prices(self):
        """Check price alerts."""
        try:
            data = self._clore_request("/v1/marketplace")
            servers = data.get("servers", [])
            
            for alert in self.price_alerts:
                if alert["triggered"]:
                    continue
                
                for server in servers:
                    if server.get("rented"):
                        continue
                    
                    gpus = server.get("gpu_array", [])
                    if not any(alert["gpu_type"].lower() in g.lower() for g in gpus):
                        continue
                    
                    price = server.get("price", {}).get("usd", {}).get("on_demand_clore", 999)
                    
                    if price <= alert["max_price"]:
                        await self.send_alert(
                            f"💰 *Price Alert!*\n"
                            f"{alert['gpu_type']} available at ${price:.2f}/hr\n"
                            f"Server ID: {server['id']}"
                        )
                        alert["triggered"] = True
                        break
                        
        except Exception as e:
            print(f"Price check error: {e}")
    
    async def run_monitoring(self, interval: int = 60):
        """Run monitoring loop."""
        print(f"Starting monitoring (every {interval}s)...")
        
        while True:
            await self.check_orders()
            await self.check_balance()
            await self.check_prices()
            await asyncio.sleep(interval)


# Run monitoring
if __name__ == "__main__":
    import sys
    
    clore_key = sys.argv[1]
    telegram_token = sys.argv[2]
    chat_id = int(sys.argv[3])
    
    alerts = CloreAlerts(clore_key, telegram_token, chat_id)
    
    # Add price alerts
    alerts.add_price_alert("RTX 4090", 0.35)
    alerts.add_price_alert("A100", 1.50)
    
    asyncio.run(alerts.run_monitoring())

Quick Start

# Install dependencies
pip install python-telegram-bot requests

# Set environment variables
export CLORE_API_KEY="your_clore_key"
export TELEGRAM_TOKEN="your_bot_token"  # Get from @BotFather

# Run the bot
python clore_telegram_bot.py

# Or run monitoring separately
python alerts.py $CLORE_API_KEY $TELEGRAM_TOKEN $YOUR_CHAT_ID

Bot Commands

Command	Description
/start	Welcome and help
/browse	List available GPUs
/rent <gpu>	Rent specific GPU type
/orders	View active orders
/balance	Check wallet
/cancel <id>	Cancel order

Features

• ✅ Browse GPUs from Telegram

• ✅ One-tap rentals

• ✅ Order monitoring

• ✅ Balance checking

• ✅ Price alerts

• ✅ Status change notifications

Next Steps

• Price Tracking Dashboard

• Webhook-Based Order Management

• Multi-Cloud GPU Orchestrator