Personal RAG MCP Server

"""Simple test to verify MCP server components.""" import asyncio import os import sys async def test_basic(): print("=" * 60) print("Personal RAG MCP Server - Basic Component Test") print("=" * 60) print("\n[1/4] Testing imports...") try: from personal_rag_mcp.storage.sqlite_store import SQLiteStore from personal_rag_mcp.storage.qdrant_store import QdrantStore from personal_rag_mcp.utils.embeddings import EmbeddingClient print("✓ All imports successful") except Exception as e: print(f"✗ Import failed: {e}") return False # Test SQLite print("\n[2/4] Testing SQLite...") try: sqlite = SQLiteStore("/tmp/test.db") await sqlite.initialize() print("✓ SQLite initialized") except Exception as e: print(f"✗ SQLite failed: {e}") import traceback traceback.print_exc() return False # Test Qdrant connection print("\n[3/4] Testing Qdrant connection...") try: qdrant_url = os.getenv("QDRANT_URL", "http://qdrant:6333") print(f" Connecting to: {qdrant_url}") qdrant = QdrantStore(qdrant_url, collection_name="test_collection") await qdrant.initialize() print("✓ Qdrant connected and collection created") except Exception as e: print(f"✗ Qdrant failed: {e}") import traceback traceback.print_exc() return False # Test Ollama connection print("\n[4/4] Testing Ollama embeddings...") try: ollama_url = os.getenv("OLLAMA_URL", "http://ollama:11434") print(f" Connecting to: {ollama_url}") embeddings = EmbeddingClient(ollama_url) embedding = await embeddings.embed_text("test") print(f"✓ Ollama embedding generated ({len(embedding)} dimensions)") except Exception as e: print(f"✗ Ollama failed: {e}") import traceback traceback.print_exc() return False print("\n" + "=" * 60) print("✓✓✓ All basic tests passed!") print("=" * 60) return True if __name__ == "__main__": result = asyncio.run(test_basic()) sys.exit(0 if result else 1)