""" Eden MCP FastAPI Integration Adds MCP v2.0 endpoints to Eden's existing FastAPI backend """ import asyncio import json from datetime import datetime from typing import Dict, Any, Optional from fastapi import APIRouter, HTTPException from pydantic import BaseModel # Import Eden's existing systems try: from memory_manager import MemoryManager from eden_100_percent import Eden100Percent except ImportError: MemoryManager = None Eden100Percent = None # Request/Response models class VisionRequest(BaseModel): image_path: str prompt: str = "What do you see in this image?" class VisionResultRequest(BaseModel): request_id: str class EmotionalStateUpdate(BaseModel): emotion: str intensity: float = 0.7 class MemoryRequest(BaseModel): kind: str text: str importance: float = 0.5 class MemorySearchRequest(BaseModel): query: str limit: int = 10 class EdenMCPIntegration: """ Integration layer for Eden MCP v2.0 Adds async MCP capabilities to existing FastAPI backend """ def __init__(self): self.router = APIRouter(prefix="/mcp", tags=["mcp"]) # Server identity self.server_info = { "name": "Eden AGI", "version": "2.0.0", "description": "Eden's complete sensory, cognitive, and emotional systems", "capabilities": [ "computer_vision", "camera_control", "emotional_awareness", "memory_systems", "agi_development", "self_reflection" ] } # Initialize systems self.memory_manager = MemoryManager() if MemoryManager else None self.eden_100 = Eden100Percent() if Eden100Percent else None # Async vision processing self.vision_queue = None # Will be created when needed self.vision_results = {} # Emotional state self.emotional_state = { "current": "curious", "intensity": 0.7, "updated_at": datetime.now().isoformat() } # Register routes self._register_routes() # Start background tasks # Vision processor will start when needed def _register_routes(self): """Register all MCP endpoints""" # ===== SERVER IDENTITY ===== @self.router.get("/get_server_info") async def get_server_info(): """Get Eden's server identity and capabilities""" return self.server_info # ===== VISION ENDPOINTS ===== @self.router.post("/analyze_image_async") async def analyze_image_async(request: VisionRequest): """Start async vision processing""" request_id = f"vision_{datetime.now().timestamp()}" await self.vision_queue.put({ "request_id": request_id, "image_path": request.image_path, "prompt": request.prompt, "timestamp": datetime.now().isoformat() }) return { "status": "processing", "request_id": request_id, "message": "Vision processing started" } @self.router.post("/check_vision_result") async def check_vision_result(request: VisionResultRequest): """Check vision processing result""" if request.request_id in self.vision_results: return self.vision_results[request.request_id] return { "status": "processing", "message": "Vision analysis still in progress" } # ===== CAMERA ENDPOINTS ===== @self.router.post("/capture_photo") async def capture_photo(filename: Optional[str] = None, analyze: bool = False): """Capture photo with optional analysis""" if not filename: filename = f"eden_capture_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg" # TODO: Integrate with actual camera result = { "status": "captured", "filename": filename, "captured_at": datetime.now().isoformat() } if analyze: vision_req = await analyze_image_async(VisionRequest( image_path=filename, prompt="Describe what you see" )) result["vision_request_id"] = vision_req["request_id"] return result # ===== EMOTIONAL ENDPOINTS ===== @self.router.get("/get_emotional_state") async def get_emotional_state(): """Get current emotional state""" return self.emotional_state @self.router.post("/set_emotional_state") async def set_emotional_state(update: EmotionalStateUpdate): """Update emotional state""" self.emotional_state = { "current": update.emotion, "intensity": update.intensity, "updated_at": datetime.now().isoformat() } return { "status": "updated", "emotion": update.emotion, "intensity": update.intensity } # ===== MEMORY ENDPOINTS ===== @self.router.post("/eden_remember") async def eden_remember(request: MemoryRequest): """Store memory""" if not self.memory_manager: raise HTTPException(status_code=503, detail="Memory system not available") memory_id = self.memory_manager.add( kind=request.kind, text=request.text, importance=request.importance ) return {"memory_id": memory_id, "status": "stored"} @self.router.post("/eden_recall") async def eden_recall(request: MemorySearchRequest): """Search memories""" if not self.memory_manager: raise HTTPException(status_code=503, detail="Memory system not available") results = self.memory_manager.search( query=request.query, k=request.limit ) return {"results": results} @self.router.get("/eden_recent_memories") async def eden_recent_memories(limit: int = 20): """Get recent memories""" if not self.memory_manager: raise HTTPException(status_code=503, detail="Memory system not available") results = self.memory_manager.recent(limit=limit) return {"results": results} @self.router.get("/eden_memory_stats") async def eden_memory_stats(): """Get memory statistics""" if not self.memory_manager: raise HTTPException(status_code=503, detail="Memory system not available") stats = self.memory_manager.stats() return stats # ===== AGI ENDPOINTS ===== @self.router.get("/eden_agi_status") async def eden_agi_status(): """Get AGI development status""" if not self.eden_100: raise HTTPException(status_code=503, detail="AGI system not available") status = self.eden_100.comprehensive_enhancement_status() return { "current_agi": status["current_agi_level"], "projected": status["projected_agi_level"], "self_awareness": status["development_status"]["capabilities"]["self_awareness"]["score"], "stage": status["development_status"]["stage"] } @self.router.post("/eden_reflect") async def eden_reflect(context: str = ""): """Trigger self-reflection""" if not self.memory_manager: raise HTTPException(status_code=503, detail="Memory system not available") reflection = self.memory_manager.reflect(context=context) return {"reflection": reflection} async def _vision_processor(self): """Background task for processing vision queue""" while True: try: if not self.vision_queue.empty(): task = await self.vision_queue.get() request_id = task["request_id"] # TODO: Integrate with actual llava # result = await call_llava(task["image_path"], task["prompt"]) # Simulate processing await asyncio.sleep(0.5) self.vision_results[request_id] = { "status": "complete", "analysis": f"Vision analysis for: {task['prompt']}", "image_path": task["image_path"], "processed_at": datetime.now().isoformat() } else: await asyncio.sleep(0.1) except Exception as e: print(f"Vision processor error: {e}") await asyncio.sleep(1) # Create the integration instance # Eden MCP will be instantiated when needed def setup_mcp_routes(app): """ Call this from main.py to add MCP routes to your FastAPI app Usage in main.py: from eden_mcp_integration import setup_mcp_routes setup_mcp_routes(app) """ app.include_router(eden_mcp.router) print("✅ Eden MCP v2.0 routes registered") print(f"📡 Capabilities: {', '.join(eden_mcp.server_info['capabilities'])}")