MCP Memory Service

#!/usr/bin/env python3 """Enhanced diagnostic script to debug server initialization and Cloudflare backend issues""" import asyncio import os import sys import traceback from pathlib import Path import logging # Setup logging to see detailed information logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Setup path sys.path.insert(0, str(Path(__file__).parent / "src")) os.chdir(Path(__file__).parent) # Load environment try: from mcp_memory_service import env_loader except ImportError: # env_loader might not be available in newer versions pass from mcp_memory_service.config import STORAGE_BACKEND, CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID print("=" * 80) print("ENHANCED MCP MEMORY SERVICE CLOUDFLARE BACKEND DIAGNOSTIC") print("=" * 80) print(f"\n📋 Configuration Check:") print(f" Storage Backend: {STORAGE_BACKEND}") print(f" API Token: {'SET' if CLOUDFLARE_API_TOKEN else 'NOT SET'}") print(f" Account ID: {'SET' if CLOUDFLARE_ACCOUNT_ID else 'NOT SET'}") async def test_server_initialization(): """Test the actual server initialization flow""" print(f"\n🚀 Testing Server Initialization Flow:") try: from mcp_memory_service.server import MemoryServer print(" ✅ MemoryServer import successful") server = MemoryServer() print(" ✅ MemoryServer instance created") # Test the eager initialization directly print(f"\n⚡ Testing Eager Storage Initialization:") try: success = await server._initialize_storage_with_timeout() print(f" 📊 Eager init result: {'SUCCESS' if success else 'FAILED'}") if success and hasattr(server, 'storage') and server.storage: storage_type = server.storage.__class__.__name__ print(f" 📦 Storage type: {storage_type}") # Test storage initialization if hasattr(server.storage, 'initialize'): print(f" 🔧 Testing storage.initialize()...") await server.storage.initialize() print(f" ✅ Storage initialization complete") else: print(f" ❌ No storage object created or eager init failed") except Exception as eager_error: print(f" ❌ Eager initialization error: {str(eager_error)}") print(f" 📝 Traceback:") traceback.print_exc() # Test the lazy initialization path print(f"\n🔄 Testing Lazy Storage Initialization:") # Reset state to test lazy initialization server.storage = None server._storage_initialized = False try: storage = await server._ensure_storage_initialized() if storage: storage_type = storage.__class__.__name__ print(f" ✅ Lazy init successful, storage type: {storage_type}") else: print(f" ❌ Lazy init returned None") except Exception as lazy_error: print(f" ❌ Lazy initialization error: {str(lazy_error)}") print(f" 📝 Traceback:") traceback.print_exc() # Test health check print(f"\n🏥 Testing Health Check:") try: result = await server.handle_check_database_health({}) health_text = result[0].text if result and len(result) > 0 else "No result" print(f" 📊 Health check result:") # Parse and pretty print the health check result import json try: health_data = json.loads(health_text.replace("Database Health Check Results:\n", "")) backend = health_data.get("statistics", {}).get("backend", "unknown") status = health_data.get("validation", {}).get("status", "unknown") print(f" Backend: {backend}") print(f" Status: {status}") if "error" in health_data.get("statistics", {}): print(f" Error: {health_data['statistics']['error']}") except json.JSONDecodeError: print(f" Raw result: {health_text[:200]}...") except Exception as health_error: print(f" ❌ Health check error: {str(health_error)}") print(f" 📝 Traceback:") traceback.print_exc() except Exception as server_error: print(f"❌ Server creation error: {str(server_error)}") print(f"📝 Traceback:") traceback.print_exc() async def test_cloudflare_storage_directly(): """Test Cloudflare storage initialization directly""" print(f"\n☁️ Testing Cloudflare Storage Directly:") try: from mcp_memory_service.storage.cloudflare import CloudflareStorage from mcp_memory_service.config import ( CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_VECTORIZE_INDEX, CLOUDFLARE_D1_DATABASE_ID, CLOUDFLARE_R2_BUCKET, CLOUDFLARE_EMBEDDING_MODEL, CLOUDFLARE_LARGE_CONTENT_THRESHOLD, CLOUDFLARE_MAX_RETRIES, CLOUDFLARE_BASE_DELAY ) print(f" 📋 Creating CloudflareStorage instance...") storage = CloudflareStorage( api_token=CLOUDFLARE_API_TOKEN, account_id=CLOUDFLARE_ACCOUNT_ID, vectorize_index=CLOUDFLARE_VECTORIZE_INDEX, d1_database_id=CLOUDFLARE_D1_DATABASE_ID, r2_bucket=CLOUDFLARE_R2_BUCKET, embedding_model=CLOUDFLARE_EMBEDDING_MODEL, large_content_threshold=CLOUDFLARE_LARGE_CONTENT_THRESHOLD, max_retries=CLOUDFLARE_MAX_RETRIES, base_delay=CLOUDFLARE_BASE_DELAY ) print(f" ✅ CloudflareStorage instance created") print(f" 🔧 Testing initialize() method...") await storage.initialize() print(f" ✅ CloudflareStorage.initialize() completed") print(f" 📊 Testing get_stats() method...") stats = await storage.get_stats() print(f" ✅ Statistics retrieved: {stats}") except Exception as direct_error: print(f" ❌ Direct Cloudflare storage error: {str(direct_error)}") print(f" 📝 Traceback:") traceback.print_exc() async def main(): """Run all diagnostic tests""" await test_cloudflare_storage_directly() await test_server_initialization() print(f"\n" + "=" * 80) print("DIAGNOSTIC COMPLETE") print("=" * 80) if __name__ == "__main__": asyncio.run(main())