File size: 7,526 Bytes
38980bb 330c7d5 38980bb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 | from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
import logging
import threading
import time
import asyncio
from typing import Dict, Optional
import os
# Import core components
from components.search.search import perform_search, get_all_events, initialize_embeddings, refresh_embeddings
from components.Event_ai.crew import EventContentCrew
# Import MCP server
from mcp_server import SimpleMCPServer
# -----------------------------
# Logging
# -----------------------------
logger = logging.getLogger(__name__)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# -----------------------------
# FastAPI app
# -----------------------------
app = FastAPI(
title="EventHub API",
description="Semantic Event Search, Event Creation, and AI-assisted tools API",
version="1.1.0",
docs_url="/docs"
)
# -----------------------------
# Middleware
# -----------------------------
app.add_middleware(
CORSMiddleware,
allow_origins=["https://eventhu.vercel.app/","https://event-hub0.vercel.app","http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
_mcp_server_instance: Optional[SimpleMCPServer] = None
_mcp_server_started = False
# -----------------------------
# Models
# -----------------------------
class SearchRequest(BaseModel):
query: str
user_id: str = "default"
class EventInput(BaseModel):
description: str
title: str
# -----------------------------
# Global state tracking
# -----------------------------
_mcp_server_started = False
_mcp_server_instance: Optional[SimpleMCPServer] = None
# -----------------------------
# Background task for periodic refresh
# -----------------------------
def periodic_refresh():
"""Refresh embeddings every 30 minutes"""
logger.info("→ Starting periodic refresh thread")
while True:
try:
time.sleep(1800) # 30 minutes
logger.info("→ Starting periodic embedding refresh...")
refresh_embeddings()
logger.info("✓ Embeddings refreshed successfully")
except Exception as e:
logger.error(f"✗ Error during periodic refresh: {e}")
# -----------------------------
# Startup
# -----------------------------
@app.on_event("startup")
async def startup_event():
global _mcp_server_started, _mcp_server_instance
logger.info("→ Starting EventHub application")
# Initialize embeddings asynchronously
try:
logger.info("→ Initializing embeddings...")
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, initialize_embeddings)
logger.info("✓ Embeddings initialized successfully")
except Exception as e:
logger.error(f"✗ Error initializing embeddings: {e}")
logger.warning("→ Continuing without embeddings (search functionality may be limited)")
# Start background refresh thread
try:
refresh_thread = threading.Thread(target=periodic_refresh, daemon=True)
refresh_thread.start()
logger.info("✓ Background refresh thread started")
except Exception as e:
logger.error(f"✗ Failed to start refresh thread: {e}")
# Launch MCP server instance
try:
if not _mcp_server_started:
logger.info("→ Starting MCP server instance...")
_mcp_server_instance = SimpleMCPServer()
_mcp_server_started = True
logger.info("✓ MCP server instance ready")
except Exception as e:
logger.error(f"✗ Failed to start MCP server: {e}")
# -----------------------------
# Routes
# -----------------------------
@app.get("/")
async def health_check():
return {"status": "healthy", "message": "EventHub API is running"}
@app.post("/search")
async def search_endpoint(request: SearchRequest):
try:
results = perform_search(request.query, request.user_id)
if not results:
return {"results": [], "message": "No matching events found"}
return {"results": results}
except Exception as e:
logger.error(f"Search endpoint error: {e}")
raise HTTPException(status_code=500, detail="Search failed")
@app.post("/generate-event")
def generate_event(event: EventInput):
try:
crew = EventContentCrew()
result = crew.eventcrew().kickoff(
inputs={"event_description": event.description, "title": event.title}
)
# Access tasks_output safely
tasks = getattr(result, "tasks_output", [])
if tasks:
last_task = tasks[-1]
# Use vars() to convert TaskOutput object to dict
last_task_dict = vars(last_task) if hasattr(last_task, "__dict__") else {}
proposals = last_task_dict.get("json_dict", {}).get("proposals", [])
else:
proposals = getattr(result, "proposals", [])
return {"proposals": proposals}
except Exception as e:
logger.error(f"Generate-event error: {e}")
raise HTTPException(status_code=500, detail="Event generation failed")
@app.post("/refresh-embeddings")
async def manual_refresh():
try:
collection = refresh_embeddings()
return {"status": "success", "message": "Embeddings refreshed successfully","collection": collection}
except Exception as e:
logger.error(f"Manual refresh error: {e}")
raise HTTPException(status_code=500, detail="Embedding refresh failed")
# -----------------------------
# MCP HTTP Bridge Routes
# -----------------------------
@app.post("/mcp/tools/call")
async def mcp_tools_call(request: Dict):
global _mcp_server_instance
if not _mcp_server_instance:
logger.error("✗ MCP server instance not available")
raise HTTPException(status_code=503, detail="MCP server not initialized")
logger.info(f"→ MCP bridge request: {request.get('method', 'unknown')}")
try:
response = await _mcp_server_instance.handle_request(request)
logger.info("✓ MCP bridge request completed")
return response
except Exception as e:
logger.error(f"✗ MCP bridge error: {e}")
raise HTTPException(status_code=500, detail=f"MCP request failed: {str(e)}")
@app.post("/mcp/tools/list")
async def mcp_tools_list():
global _mcp_server_instance
if not _mcp_server_instance:
logger.error("✗ MCP server instance not available")
raise HTTPException(status_code=503, detail="MCP server not initialized")
logger.info("→ MCP tools list request")
try:
request = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
response = await _mcp_server_instance.handle_request(request)
logger.info("✓ MCP tools list completed")
return response
except Exception as e:
logger.error(f"✗ MCP tools list error: {e}")
raise HTTPException(status_code=500, detail=f"MCP tools list failed: {str(e)}")
if __name__ == "__main__":
logger.info("→ Starting EventHub server on http://localhost:8000")
logger.info("→ API documentation available at http://localhost:8000/docs")
try:
port = int(os.getenv("PORT", 8000))
uvicorn.run(app, host="0.0.0.0", port=port, reload=True)
except Exception as e:
logger.error(f"✗ Failed to start server: {e}")
raise
|