""" Hermes Agent - FastAPI Backend Server Serves both the API endpoints and Telegram webhook """ import os import logging from contextlib import asynccontextmanager from fastapi import FastAPI, Request, HTTPException, BackgroundTasks from fastapi.responses import JSONResponse, HTMLResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn import config from hermes_agent import process_query, get_agent from telegram_bot import get_bot, TelegramBot # Configure logging logging.basicConfig( level=getattr(logging, config.LOG_LEVEL), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("hermes-api") # Check if we have the API key set try: config.validate_config() logger.info("Configuration validated successfully") except ValueError as e: logger.warning(f"Configuration issue: {e}") # App metadata for documentation APP_METADATA = { "title": "Hermes Agent API", "description": """ 🤖 Hermes Agent - AI-powered assistant with 100+ API integrations Built with Nous Research Hermes function calling standard. ## Features - **Weather** - Global forecasts for any city - **Crypto** - Real-time prices for 1000+ coins - **Anime** - Search MyAnimeList database - **Games** - Pokemon, D&D, chess data - **Food** - 300,000+ recipes and cocktails - **Space** - ISS tracking, SpaceX launches, NASA APOD - **Sports** - Football, NBA, F1 standings - **Quotes** - Inspirational, movie, TV quotes - **Science** - Periodic table, number facts - **News** - Hacker News, space news - **Entertainment** - Jokes, memes, trivia - **Utility** - Translation, currency conversion, QR codes ## Usage Send a POST request to `/chat` with your message and get an AI response that intelligently uses tools when needed. """, "version": "1.0.0", "contact": { "name": "Hermes Agent", "url": "https://github.com/NousResearch/Hermes-Function-Calling" } } @asynccontextmanager async def lifespan(app: FastAPI): """Lifespan context manager for startup/shutdown events""" # Startup logger.info("🚀 Hermes Agent starting up...") logger.info(f"Model: {config.DEFAULT_MODEL}") logger.info(f"Max recursion depth: {config.MAX_RECURSION_DEPTH}") # Initialize agent agent = get_agent() logger.info("✅ Agent initialized") # Setup Telegram bot in polling mode for maximum reliability on HF if config.TELEGRAM_BOT_TOKEN: try: bot = get_bot() # Run polling in a separate thread to not block FastAPI import threading def start_polling(): bot.run_polling() polling_thread = threading.Thread(target=start_polling, daemon=True) polling_thread.start() logger.info("✅ Telegram bot started in polling mode (background thread)") except Exception as e: logger.warning(f"⚠️ Could not start Telegram bot: {e}") yield # Shutdown logger.info("🛑 Hermes Agent shutting down...") # Create FastAPI app app = FastAPI( title=APP_METADATA["title"], description=APP_METADATA["description"], version=APP_METADATA["version"], contact=APP_METADATA["contact"], lifespan=lifespan ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ============================================================ # HTML DASHBOARD (Landing Page) # ============================================================ DASHBOARD_HTML = """ Hermes Agent - AI with 100+ APIs

Hermes Agent

AI-Powered Assistant with 100+ Real-World API Integrations

Online
100+
APIs Integrated
50+
Categories
∞
Possibilities
🌤️

Weather & Environment

Global weather forecasts, air quality, UV index, and environmental data for any location on Earth.

💰

Crypto & Finance

Real-time cryptocurrency prices, market data, exchange rates, and financial information.

🎮

Games & Entertainment

Pokemon database, D&D references, anime search, game stats, and entertainment data.

🚀

Space & Science

ISS tracking, SpaceX launches, NASA imagery, periodic table, and scientific facts.

🍕

Food & Drink

300,000+ recipes, cocktail instructions, brewery database, and food information.

⚽

Sports

Football standings, NBA games, F1 results, chess ratings, and bike sharing data.

📚

Books & Knowledge

Book search, dictionary definitions, quotes, trivia questions, and educational content.

😂

Fun & Social

Jokes, memes, Reddit posts, Chuck Norris facts, Rick & Morty data, and more.

🔗 API Endpoints

GET / - Dashboard (this page)
GET /health - Health check
POST /chat - Send a message to Hermes
POST /telegram/webhook - Telegram bot webhook
GET /tools - List all available tools

Powered by Nous Research Hermes | Built with FastAPI | Deployed on Hugging Face Spaces

Based on Hermes-Function-Calling

""" # ============================================================ # API ENDPOINTS # ============================================================ @app.get("/", response_class=HTMLResponse) async def dashboard(): """Serve the dashboard HTML page""" return HTMLResponse(content=DASHBOARD_HTML) @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "version": "1.0.0", "model": config.DEFAULT_MODEL, "features": { "agent": True, "telegram": bool(config.TELEGRAM_BOT_TOKEN), "tools_count": 100, "api_categories": 50 } } @app.post("/chat") async def chat(request: Request): """ Main chat endpoint - send a message to Hermes Agent Request body: { "message": "Your question here", "user_id": "optional_user_id" } """ try: data = await request.json() message = data.get("message", "") user_id = data.get("user_id", "default") if not message: raise HTTPException(status_code=400, detail="Message is required") # Process through Hermes agent response = process_query(message, user_id) return { "response": response, "user_id": user_id, "status": "success" } except HTTPException: raise except Exception as e: logger.error(f"Chat error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/tools") async def list_tools(): """List all available tools/functions""" from api_functions import get_all_function_names, get_function_schemas return { "functions": get_all_function_names(), "count": len(get_all_function_names()), "schemas": get_function_schemas() } @app.post("/telegram/webhook") async def telegram_webhook(request: Request): """ Telegram webhook endpoint Receives updates from Telegram when the bot is in webhook mode """ try: update_data = await request.json() bot = get_bot() bot.process_update(update_data) return {"status": "ok"} except Exception as e: logger.error(f"Webhook error: {e}") return {"status": "error", "message": str(e)} @app.get("/telegram/webhook-info") async def telegram_webhook_info(): """Get Telegram webhook information""" try: bot = get_bot() info = await bot.application.bot.get_webhook_info() return info.to_dict() except Exception as e: return {"error": str(e)} @app.get("/demo") async def demo_page(): """Interactive demo page""" return HTMLResponse(content=""" Hermes Agent - Demo

🤖 Hermes Agent Demo

""") # Error handlers @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): """Handle all unhandled exceptions""" logger.error(f"Unhandled exception: {exc}") return JSONResponse( status_code=500, content={"detail": "Internal server error", "error": str(exc)} ) # Startup validation @app.on_event("startup") async def startup_event(): """Run on startup""" logger.info("=" * 60) logger.info("HERMES AGENT STARTED") logger.info(f"Port: {config.PORT}") logger.info(f"Debug: {config.DEBUG}") logger.info("=" * 60) if __name__ == "__main__": # Run the server uvicorn.run( "main:app", host=config.HOST, port=config.PORT, reload=config.DEBUG, log_level="info" )