""" Telegram Bot Integration for Hermes Agent Supports both webhook and polling modes """ import os import asyncio import logging from typing import Optional from telegram import Update, BotCommand from telegram.ext import ( Application, CommandHandler, MessageHandler, filters, ContextTypes, ) import config from hermes_agent import process_query, get_agent logger = logging.getLogger("telegram-bot") # Bot commands COMMANDS = [ BotCommand("start", "Start the bot and see welcome message"), BotCommand("help", "Show available commands and features"), BotCommand("weather", "Get weather for a city - /weather London"), BotCommand("crypto", "Get crypto price - /crypto bitcoin"), BotCommand("joke", "Get a random joke"), BotCommand("fact", "Get a random interesting fact"), BotCommand("quote", "Get an inspirational quote"), BotCommand("pokemon", "Get Pokemon info - /okemon pikachu"), BotCommand("anime", "Search anime - /anime attack on titan"), BotCommand("clear", "Clear conversation history"), BotCommand("about", "About Hermes Agent"), ] # Welcome message WELCOME_MESSAGE = """ 🌟 *Welcome to Hermes Agent!* 🌟 I'm your AI assistant powered by *Nous Research Hermes* with access to **hundreds of real-world APIs** and tools! 🚀 *What I can do:* • 🌤️ Weather forecasts for any city worldwide • 💰 Crypto prices & market data • 📚 Book searches & quotes • 🎮 Pokemon, anime, games & D&D info • 🍕 Recipes & cocktail instructions • 😂 Jokes, memes & fun facts • 🚀 Space data (ISS location, SpaceX, NASA) • ⚽ Sports scores & F1 standings • 🎵 Song lyrics & music search • 💱 Currency conversion • 📖 Dictionary & word definitions • 🎨 Art from museums worldwide • 🎬 TV shows & movie info • 🗺️ Location & geocoding • 🔬 Science facts & periodic table • ➕ Much more! Just ask me anything! Try: • "What's the weather in Tokyo?" • "Get me a random joke" • "Search anime One Piece" • "What's the price of Bitcoin?" • "Define serendipity" Type /help for more details. """ HELP_MESSAGE = """ 📖 *Hermes Agent Commands* */start* - Start the bot */help* - Show this help message */weather [city]* - Get weather forecast */crypto [coin]* - Get cryptocurrency price */joke* - Random joke */fact* - Random interesting fact */quote* - Inspirational quote */pokemon [name/id]* - Pokemon info */anime [title]* - Search anime */clear* - Clear your chat history */about* - About this bot *You can also ask me anything naturally!* Examples: • "What's the weather in Paris?" • "Tell me a Chuck Norris joke" • "Get recipe for lasagna" • "Who's in space right now?" • "Latest news from Hacker News" • "Convert 100 USD to EUR" • "Search for Breaking Bad quotes" • "Get F1 standings" • "Draw 5 random cards" • "What's Pokemon #25?" """ ABOUT_MESSAGE = """ 🤖 *About Hermes Agent* Built with: • 🧠 *Nous Research Hermes* - Function calling LLM • 🐍 Python + FastAPI backend • 🌐 OpenRouter API for AI inference • 🔗 100+ integrated APIs • ☁️ Deployed on Hugging Face Spaces *Features:* • Multi-turn conversations with memory • 100+ real-world API integrations • Automatic tool selection • Weather, crypto, news, entertainment • Games, anime, sports, science • And much more! The agent uses the *Hermes Function Calling* standard to intelligently select and execute tools based on your questions. *Open Source:* Based on NousResearch/Hermes-Function-Calling """ async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /start command""" await update.message.reply_text( WELCOME_MESSAGE, parse_mode="Markdown", disable_web_page_preview=True ) async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /help command""" await update.message.reply_text( HELP_MESSAGE, parse_mode="Markdown" ) async def about_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /about command""" await update.message.reply_text( ABOUT_MESSAGE, parse_mode="Markdown" ) async def weather_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /weather command""" city = " ".join(context.args) if context.args else "London" processing_msg = await update.message.reply_text(f"🌤️ Getting weather for *{city}*...", parse_mode="Markdown") try: response = process_query(f"What's the weather in {city}?", str(update.effective_user.id)) await processing_msg.delete() await update.message.reply_text(response[:4000] if len(response) > 4000 else response) except Exception as e: await processing_msg.delete() await update.message.reply_text(f"❌ Error getting weather: {str(e)[:200]}") async def crypto_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /crypto command""" coin = context.args[0] if context.args else "bitcoin" processing_msg = await update.message.reply_text(f"💰 Fetching {coin} price...") try: response = process_query(f"What's the current price of {coin}?", str(update.effective_user.id)) await processing_msg.delete() await update.message.reply_text(response[:4000] if len(response) > 4000 else response) except Exception as e: await processing_msg.delete() await update.message.reply_text(f"❌ Error: {str(e)[:200]}") async def joke_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /joke command""" try: response = process_query("Tell me a random joke", str(update.effective_user.id)) await update.message.reply_text(response) except Exception as e: await update.message.reply_text(f"❌ Error: {str(e)[:200]}") async def fact_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /fact command""" try: response = process_query("Tell me an interesting random fact", str(update.effective_user.id)) await update.message.reply_text(response) except Exception as e: await update.message.reply_text(f"❌ Error: {str(e)[:200]}") async def quote_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /quote command""" try: response = process_query("Give me an inspirational quote of the day", str(update.effective_user.id)) await update.message.reply_text(response) except Exception as e: await update.message.reply_text(f"❌ Error: {str(e)[:200]}") async def pokemon_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /pokemon command""" name_or_id = context.args[0] if context.args else "25" processing_msg = await update.message.reply_text(f"🎮 Looking up Pokemon...") try: response = process_query(f"Get Pokemon info for {name_or_id}", str(update.effective_user.id)) await processing_msg.delete() await update.message.reply_text(response[:4000] if len(response) > 4000 else response) except Exception as e: await processing_msg.delete() await update.message.reply_text(f"❌ Error: {str(e)[:200]}") async def anime_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /anime command""" title = " ".join(context.args) if context.args else "Attack on Titan" processing_msg = await update.message.reply_text(f"🔍 Searching anime: *{title}*...", parse_mode="Markdown") try: response = process_query(f"Search anime {title}", str(update.effective_user.id)) await processing_msg.delete() await update.message.reply_text(response[:4000] if len(response) > 4000 else response) except Exception as e: await processing_msg.delete() await update.message.reply_text(f"❌ Error: {str(e)[:200]}") async def clear_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /clear command - clear conversation history""" user_id = str(update.effective_user.id) agent = get_agent() agent.clear_history(user_id) await update.message.reply_text("🗑️ *Conversation history cleared!*", parse_mode="Markdown") async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle regular text messages""" user_id = str(update.effective_user.id) user_message = update.message.text # Show typing indicator await context.bot.send_chat_action( chat_id=update.effective_chat.id, action="typing" ) try: # Process through Hermes agent response = process_query(user_message, user_id) # Send response (Telegram has 4096 char limit per message) if len(response) > 4000: # Split into chunks chunks = [response[i:i+4000] for i in range(0, len(response), 4000)] for chunk in chunks: await update.message.reply_text(chunk) else: await update.message.reply_text(response) except Exception as e: logger.error(f"Error processing message: {e}") await update.message.reply_text( f"❌ Sorry, I encountered an error:\n```\n{str(e)[:300]}\n```\n\n" "Please try again or use /clear to reset the conversation.", parse_mode="Markdown" ) async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle errors""" logger.error(f"Update {update} caused error: {context.error}") if update and update.effective_message: await update.effective_message.reply_text( "❌ An unexpected error occurred. Please try again later." ) class TelegramBot: """Telegram Bot wrapper for Hermes Agent""" def __init__(self, token: str = None): self.token = token or config.TELEGRAM_BOT_TOKEN self.application = None def setup(self): """Initialize the bot application""" self.application = Application.builder().token(self.token).build() # Register handlers self.application.add_handler(CommandHandler("start", start_command)) self.application.add_handler(CommandHandler("help", help_command)) self.application.add_handler(CommandHandler("about", about_command)) self.application.add_handler(CommandHandler("weather", weather_command)) self.application.add_handler(CommandHandler("crypto", crypto_command)) self.application.add_handler(CommandHandler("joke", joke_command)) self.application.add_handler(CommandHandler("fact", fact_command)) self.application.add_handler(CommandHandler("quote", quote_command)) self.application.add_handler(CommandHandler("pokemon", pokemon_command)) self.application.add_handler(CommandHandler("anime", anime_command)) self.application.add_handler(CommandHandler("clear", clear_command)) self.application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) self.application.add_error_handler(error_handler) logger.info("Telegram bot setup complete") def run_polling(self): """Run the bot in polling mode (for local development)""" if not self.application: self.setup() logger.info("Starting bot in polling mode...") self.application.run_polling( drop_pending_updates=True, allowed_updates=Update.ALL_TYPES ) async def setup_webhook(self, webhook_url: str): """Setup webhook for production""" if not self.application: self.setup() # Set webhook await self.application.bot.set_webhook( url=webhook_url, allowed_updates=Update.ALL_TYPES ) # Set bot commands await self.application.bot.set_my_commands(COMMANDS) logger.info(f"Webhook set to: {webhook_url}") def get_webhook_handler(self): """Get the webhook handler for FastAPI integration""" if not self.application: self.setup() return self.application def process_update(self, update_data: dict): """Process an update from webhook""" if not self.application: self.setup() update = Update.de_json(update_data, self.application.bot) self.application.process_update(update) # Global bot instance _bot = None def get_bot() -> TelegramBot: """Get or create the global bot instance""" global _bot if _bot is None: _bot = TelegramBot() return _bot