Spaces:
Configuration error
Configuration error
| """ | |
| 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 | |