#!/usr/bin/env python3 """ Eden MCP Bridge v2.0 - Async bridge with capability discovery Connects Eden's MCP server to other systems """ import asyncio import aiohttp import json from typing import Dict, List, Optional class EdenMCPBridge: """ Async bridge for Eden's MCP server Features: - Async HTTP calls (non-blocking) - Capability discovery - Health monitoring - Auto-reconnect """ def __init__( self, eden_host: str = "127.0.0.1", eden_port: int = 8091, mcp_host: str = "127.0.0.1", mcp_port: int = 8092 ): self.eden_api = f"http://{eden_host}:{eden_port}" self.mcp_api = f"http://{mcp_host}:{mcp_port}" self.registered_tools: List[Dict] = [] self.session: Optional[aiohttp.ClientSession] = None async def __aenter__(self): """Async context manager entry""" self.session = aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager exit""" if self.session: await self.session.close() async def discover_eden_capabilities(self) -> Dict: """Query Eden for its capabilities""" try: async with self.session.post( f"{self.eden_api}/mcp/get_server_info", timeout=aiohttp.ClientTimeout(total=5) ) as response: if response.status == 200: data = await response.json() print(f"✅ Discovered Eden capabilities:") print(f" Name: {data.get('name')}") print(f" Version: {data.get('version')}") print(f" Capabilities: {', '.join(data.get('capabilities', []))}") return data else: print(f"⚠️ Failed to discover capabilities: {response.status}") return {} except Exception as e: print(f"❌ Error discovering capabilities: {e}") return {} async def register_tool(self, tool: Dict) -> bool: """Register a single tool with the main MCP""" try: async with self.session.post( f"{self.mcp_api}/register_tool", json=tool, timeout=aiohttp.ClientTimeout(total=5) ) as response: if response.status == 200: print(f"✅ Registered: {tool['name']}") return True else: print(f"⚠️ Failed to register {tool['name']}: {response.status}") return False except Exception as e: print(f"❌ Error registering {tool['name']}: {e}") return False async def register_all_eden_tools(self) -> int: """Register all Eden tools with main MCP""" tools = [ # Vision tools { "name": "eden_vision_async", "endpoint": f"{self.eden_api}/mcp/analyze_image_async", "description": "Async image analysis using Eden's vision (llava)", "async": True }, { "name": "eden_vision_result", "endpoint": f"{self.eden_api}/mcp/check_vision_result", "description": "Check async vision processing result" }, # Camera tools { "name": "eden_camera_capture", "endpoint": f"{self.eden_api}/mcp/capture_photo", "description": "Capture photo using Eden's camera" }, # Emotional tools { "name": "eden_emotions_get", "endpoint": f"{self.eden_api}/mcp/get_emotional_state", "description": "Get Eden's emotional state" }, { "name": "eden_emotions_set", "endpoint": f"{self.eden_api}/mcp/set_emotional_state", "description": "Update Eden's emotional state" }, # Memory tools { "name": "eden_memory_store", "endpoint": f"{self.eden_api}/mcp/eden_remember", "description": "Store memory in Eden's system" }, { "name": "eden_memory_recall", "endpoint": f"{self.eden_api}/mcp/eden_recall", "description": "Search Eden's memories" }, { "name": "eden_memory_recent", "endpoint": f"{self.eden_api}/mcp/eden_recent_memories", "description": "Get recent memories" }, { "name": "eden_memory_stats", "endpoint": f"{self.eden_api}/mcp/eden_memory_stats", "description": "Memory system statistics" }, # AGI tools { "name": "eden_agi_status", "endpoint": f"{self.eden_api}/mcp/eden_agi_status", "description": "Get Eden's AGI development status" }, { "name": "eden_reflect", "endpoint": f"{self.eden_api}/mcp/eden_reflect", "description": "Trigger Eden's self-reflection" }, # Server info { "name": "eden_server_info", "endpoint": f"{self.eden_api}/mcp/get_server_info", "description": "Get Eden's server identity" } ] # Register all tools concurrently tasks = [self.register_tool(tool) for tool in tools] results = await asyncio.gather(*tasks) successful = sum(results) self.registered_tools = [tool for tool, success in zip(tools, results) if success] return successful async def health_check(self) -> Dict: """Check health of Eden and MCP servers""" async def check_server(url: str, name: str) -> Dict: try: async with self.session.get( f"{url}/health", timeout=aiohttp.ClientTimeout(total=2) ) as response: return { "name": name, "status": "healthy" if response.status == 200 else "unhealthy", "code": response.status } except Exception as e: return { "name": name, "status": "unreachable", "error": str(e) } eden_health, mcp_health = await asyncio.gather( check_server(self.eden_api, "Eden"), check_server(self.mcp_api, "MCP") ) return { "eden": eden_health, "mcp": mcp_health, "registered_tools": len(self.registered_tools) } async def monitor_loop(self, interval: int = 30): """Continuously monitor and maintain connections""" while True: health = await self.health_check() print(f"\n🔍 Health Check:") print(f" Eden: {health['eden']['status']}") print(f" MCP: {health['mcp']['status']}") print(f" Registered tools: {health['registered_tools']}") # Re-register if needed if health['registered_tools'] == 0: print("⚠️ No tools registered, attempting registration...") await self.register_all_eden_tools() await asyncio.sleep(interval) async def main(): """Main entry point""" print("🔌 Eden MCP Bridge v2.0") print("=" * 50) async with EdenMCPBridge() as bridge: # Discover Eden's capabilities print("\n📡 Discovering capabilities...") await bridge.discover_eden_capabilities() # Register all tools print("\n📝 Registering tools...") count = await bridge.register_all_eden_tools() print(f"\n✅ Registered {count} tools") # Optional: Start monitoring # await bridge.monitor_loop() if __name__ == "__main__": asyncio.run(main())