from fastapi import FastAPI, HTTPException, Request from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import HTMLResponse, JSONResponse from pydantic import BaseModel import asyncio import json from datetime import datetime from typing import List, Dict, Any, Optional import os from dotenv import load_dotenv load_dotenv() from src.agent.research_agent import Web3ResearchAgent from src.api.airaa_integration import AIRAAIntegration from src.utils.logger import get_logger from src.utils.config import config logger = get_logger(__name__) app = FastAPI( title="Web3 Research Co-Pilot", description="AI-powered cryptocurrency research assistant", version="1.0.0" ) # Pydantic models for request/response class QueryRequest(BaseModel): query: str chat_history: Optional[List[Dict[str, str]]] = [] class QueryResponse(BaseModel): success: bool response: str sources: Optional[List[str]] = [] metadata: Optional[Dict[str, Any]] = {} error: Optional[str] = None class Web3CoPilotService: def __init__(self): try: logger.info("🚀 Initializing Web3CoPilotService...") logger.info(f"📋 GEMINI_API_KEY configured: {'Yes' if config.GEMINI_API_KEY else 'No'}") if config.GEMINI_API_KEY: logger.info("🤖 Initializing AI agent...") self.agent = Web3ResearchAgent() logger.info("✅ AI agent initialized successfully") else: logger.warning("⚠️ GEMINI_API_KEY not found - AI features disabled") self.agent = None logger.info("🔗 Initializing AIRAA integration...") self.airaa = AIRAAIntegration() logger.info(f"🔗 AIRAA integration: {'Enabled' if self.airaa.enabled else 'Disabled'}") self.enabled = bool(config.GEMINI_API_KEY) logger.info(f"🎯 Web3CoPilotService initialized successfully (AI enabled: {self.enabled})") except Exception as e: logger.error(f"❌ Service initialization failed: {e}") self.agent = None self.airaa = None self.enabled = False async def process_query(self, query: str) -> QueryResponse: logger.info(f"🔍 Processing query: {query[:50]}{'...' if len(query) > 50 else ''}") if not query.strip(): logger.warning("⚠️ Empty query received") return QueryResponse(success=False, response="Please enter a query.", error="Empty query") try: if not self.enabled: logger.info("🔧 AI disabled - providing limited response") response = """⚠️ **AI Agent Disabled**: GEMINI_API_KEY not configured. **Limited Data Available:** - CoinGecko API (basic crypto data) - DeFiLlama API (DeFi protocols) - Etherscan API (gas prices) Please configure GEMINI_API_KEY for full AI analysis.""" return QueryResponse(success=True, response=response, sources=["Configuration"]) logger.info("🤖 Sending query to AI agent...") result = await self.agent.research_query(query) logger.info(f"✅ AI agent responded: {result.get('success', False)}") if result.get("success"): response = result.get("result", "No response generated") sources = result.get("sources", []) metadata = result.get("metadata", {}) # Send to AIRAA if enabled if self.airaa and self.airaa.enabled: try: logger.info("🔗 Sending data to AIRAA...") await self.airaa.send_research_data(query, response) logger.info("✅ Data sent to AIRAA successfully") except Exception as e: logger.warning(f"⚠️ AIRAA integration failed: {e}") logger.info("✅ Query processed successfully") return QueryResponse(success=True, response=response, sources=sources, metadata=metadata) else: error_msg = result.get("error", "Research failed. Please try again.") logger.error(f"❌ AI agent failed: {error_msg}") return QueryResponse(success=False, response=error_msg, error=error_msg) except Exception as e: logger.error(f"❌ Query processing error: {e}") error_msg = f"Error processing query: {str(e)}" return QueryResponse(success=False, response=error_msg, error=error_msg) # Initialize service logger.info("🚀 Starting Web3 Research Co-Pilot...") service = Web3CoPilotService() # API Routes @app.get("/", response_class=HTMLResponse) async def get_homepage(request: Request): logger.info("📄 Serving homepage") html_content = """ Web3 Research Co-Pilot

🚀 Web3 Research Co-Pilot

AI-powered cryptocurrency research assistant

🔄 Checking system status...
🤖 AI Research Assistant
👋 Welcome! I'm your Web3 Research Co-Pilot. Ask me anything about cryptocurrency markets, DeFi protocols, blockchain analysis, or trading insights.
📈 Bitcoin Analysis
🏦 DeFi Overview
🔥 Trending Coins
⛽ Gas Tracker
🌾 Yield Farming
⚖️ Ecosystem Compare

Powered by AI • Real-time Web3 data • Built with ❤️

""" return HTMLResponse(content=html_content) @app.get("/status") async def get_status(): logger.info("📊 Status endpoint called") status = { "enabled": service.enabled, "gemini_configured": bool(config.GEMINI_API_KEY), "tools_available": ["CoinGecko", "DeFiLlama", "Etherscan"], "airaa_enabled": service.airaa.enabled if service.airaa else False, "timestamp": datetime.now().isoformat() } logger.info(f"📊 Status response: {status}") return status @app.post("/query", response_model=QueryResponse) async def process_query(request: QueryRequest): logger.info(f"📥 Query endpoint called: {request.query[:50]}{'...' if len(request.query) > 50 else ''}") result = await service.process_query(request.query) logger.info(f"📤 Query response: success={result.success}") return result @app.get("/health") async def health_check(): logger.info("❤️ Health check endpoint called") return { "status": "healthy", "timestamp": datetime.now().isoformat(), "service_enabled": service.enabled, "version": "1.0.0" } if __name__ == "__main__": import uvicorn logger.info("🌟 Starting FastAPI server...") uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info")