NSAF MCP Server

#!/usr/bin/env python3 """ NSAF MCP Server Test Suite =========================== Author: Bolorerdene Bundgaa Contact: bolor@ariunbolor.org Website: https://bolor.me Comprehensive test suite for the NSAF MCP Server. """ import asyncio import json from nsaf_mcp_server import NSAFMCPServer async def test_framework_management(): """Test framework management tools.""" print("🔧 Testing Framework Management") print("-" * 40) server = NSAFMCPServer() # Test initialization print("1. Testing framework initialization...") result = await server.handle_tool_call("initialize_nsaf_framework", { "config_path": "config/config.yaml" }) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") if result.get('success'): status = result['result']['status'] print(f" Framework Status: {status}") # Test status check print("\n2. Testing status retrieval...") result = await server.handle_tool_call("get_nsaf_status", { "include_details": True }) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") if result.get('success'): metrics = result['result']['metrics'] print(f" Active Agents: {metrics['active_agents']}") print(f" Task Clusters: {metrics['task_clusters']}") print(f" Active Tasks: {metrics['active_tasks']}") # Test shutdown print("\n3. Testing framework shutdown...") result = await server.handle_tool_call("shutdown_nsaf_framework", {}) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") return True async def test_task_processing(): """Test task processing capabilities.""" print("\n🧠 Testing Task Processing") print("-" * 40) server = NSAFMCPServer() # Initialize framework first await server.handle_tool_call("initialize_nsaf_framework", {}) # Test complex task processing print("1. Testing complex task processing...") complex_task = { "description": "Design an intelligent recommendation system", "goals": [ {"type": "accuracy", "target": 0.92, "priority": 1.0, "complexity": 0.8, "progress": 0.0}, {"type": "latency", "target": 100, "priority": 0.8, "complexity": 0.6, "progress": 0.0} ], "constraints": [ {"type": "memory", "limit": "8GB", "importance": 0.9, "strictness": 0.7} ], "requirements": { "frameworks": ["pytorch", "scikit-learn"], "data_sources": ["user_behavior", "item_features"], "deployment": "cloud" }, "tasks": [ {"name": "data_preprocessing", "type": "pipeline", "priority": 1.0, "dependencies": []}, {"name": "model_training", "type": "ml", "priority": 0.9, "dependencies": ["data_preprocessing"]}, {"name": "deployment", "type": "deployment", "priority": 0.7, "dependencies": ["model_training"]} ], "complexity": 0.8 } result = await server.handle_tool_call("process_complex_task", complex_task) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") task_id = None if result.get('success'): task_id = result['result']['task_id'] summary = result['result']['processing_summary'] print(f" Task ID: {task_id}") print(f" Task Clusters: {summary['task_clusters']}") print(f" Agents Created: {summary['agents_created']}") print(f" Status: {result['result']['status']}") else: print(f" Error: {result.get('error', 'Unknown error')}") # Test task status retrieval if task_id: print(f"\n2. Testing task status retrieval...") result = await server.handle_tool_call("get_task_status", { "task_id": task_id }) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") if result.get('success'): task_status = result['result'] print(f" Task Status: {task_status['status']}") print(f" Progress: {task_status['progress']}") await server.handle_tool_call("shutdown_nsaf_framework", {}) return True async def test_component_tools(): """Test individual component tools.""" print("\n⚛️ Testing Component Tools") print("-" * 40) server = NSAFMCPServer() await server.handle_tool_call("initialize_nsaf_framework", {}) # Test quantum clustering print("1. Testing quantum task clustering...") result = await server.handle_tool_call("cluster_tasks_quantum", { "problem_description": "Optimize machine learning pipeline", "tasks": [ {"name": "data_collection", "description": "Gather training data"}, {"name": "preprocessing", "description": "Clean and prepare data"}, {"name": "training", "description": "Train ML models"}, {"name": "evaluation", "description": "Evaluate model performance"} ], "max_clusters": 3 }) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") if result.get('success'): clusters = result['result']['clusters_created'] print(f" Clusters Created: {clusters}") else: print(f" Error: {result.get('error', 'Unknown error')}") # Test intent projection print("\n2. Testing recursive intent projection...") result = await server.handle_tool_call("project_intent_recursive", { "intent_description": "Create an automated code review system", "goals": [ {"type": "accuracy", "target": 0.9, "priority": 0.8} ], "projection_depth": 3, "confidence_threshold": 0.6 }) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") if result.get('success'): projections = result['result']['projections_generated'] print(f" Projections Generated: {projections}") if result['result']['projections']: first_proj = result['result']['projections'][0] print(f" First Projection Confidence: {first_proj['confidence']}") # Test memory operations print("\n3. Testing memory system...") # Add memory add_result = await server.handle_tool_call("add_memory", { "content": "Machine learning optimization techniques are crucial for performance", "memory_type": "knowledge", "importance": 0.8 }) print(f" Add Memory: {'✅ SUCCESS' if add_result.get('success') else '❌ FAILED'}") # Query memory query_result = await server.handle_tool_call("query_memory", { "query": "machine learning optimization", "max_results": 5 }) print(f" Query Memory: {'✅ SUCCESS' if query_result.get('success') else '❌ FAILED'}") if query_result.get('success'): memories = query_result['result']['results_found'] print(f" Memories Found: {memories}") # Get memory metrics metrics_result = await server.handle_tool_call("get_memory_metrics", {}) print(f" Memory Metrics: {'✅ SUCCESS' if metrics_result.get('success') else '❌ FAILED'}") if metrics_result.get('success'): metrics = metrics_result['result']['metrics'] print(f" Total Memories: {metrics['total_memories']}") print(f" Connections: {metrics['total_connections']}") # Test agent evolution print("\n4. Testing agent evolution...") result = await server.handle_tool_call("evolve_agents_scma", { "task_description": "Optimize neural network architectures", "population_size": 10, "generations": 5, "architecture_complexity": "medium" }) print(f" Status: {'✅ SUCCESS' if result.get('success') else '❌ FAILED'}") if result.get('success'): agents = result['result']['agents_created'] fitness = result['result']['best_fitness'] print(f" Agents Created: {agents}") print(f" Best Fitness: {fitness}") await server.handle_tool_call("shutdown_nsaf_framework", {}) return True async def test_tool_discovery(): """Test MCP tool discovery.""" print("\n🔍 Testing Tool Discovery") print("-" * 40) server = NSAFMCPServer() print(f"Total Tools Available: {len(server.tools)}") print("\nTool Categories:") categories = { "Framework Management": ["initialize_nsaf_framework", "get_nsaf_status", "shutdown_nsaf_framework"], "Task Processing": ["process_complex_task", "get_task_status", "update_task_state"], "Quantum-Symbolic": ["cluster_tasks_quantum"], "Meta-Agents": ["evolve_agents_scma", "get_active_agents"], "Memory System": ["add_memory", "query_memory", "get_memory_metrics"], "Intent Projection": ["project_intent_recursive"], "Human-AI Synergy": ["synchronize_cognitive_state"], "Foundation Models": ["generate_with_foundation_models", "get_embeddings"], "System Analytics": ["analyze_system_performance", "update_configuration", "get_configuration"] } for category, tools in categories.items(): available_tools = [tool.name for tool in server.tools if tool.name in tools] print(f" {category}: {len(available_tools)}/{len(tools)} tools") for tool in available_tools: print(f" ✅ {tool}") return True async def demonstrate_mcp_capabilities(): """Demonstrate key MCP capabilities with practical examples.""" print("\n🚀 MCP Capability Demonstration") print("=" * 50) # Example 1: End-to-end task processing print("\n📋 Example 1: E2E Task Processing") print(" Task: Build AI-powered customer service system") server = NSAFMCPServer() await server.handle_tool_call("initialize_nsaf_framework", {}) task_result = await server.handle_tool_call("process_complex_task", { "description": "Build AI-powered customer service system", "goals": [ {"type": "response_accuracy", "target": 0.95, "priority": 1.0, "complexity": 0.8, "progress": 0.0}, {"type": "response_time", "target": 2.0, "priority": 0.9, "complexity": 0.6, "progress": 0.0}, {"type": "customer_satisfaction", "target": 0.9, "priority": 0.8, "complexity": 0.7, "progress": 0.0} ], "constraints": [ {"type": "budget", "limit": 100000, "importance": 0.9, "strictness": 0.8}, {"type": "timeline", "limit": "3 months", "importance": 0.8, "strictness": 0.7} ], "tasks": [ {"name": "intent_classification", "type": "ml", "priority": 1.0, "dependencies": []}, {"name": "response_generation", "type": "nlp", "priority": 0.9, "dependencies": ["intent_classification"]}, {"name": "sentiment_analysis", "type": "ml", "priority": 0.8, "dependencies": []}, {"name": "escalation_detection", "type": "ml", "priority": 0.7, "dependencies": ["sentiment_analysis"]}, {"name": "integration_layer", "type": "api", "priority": 0.6, "dependencies": ["response_generation", "escalation_detection"]} ], "complexity": 0.85 }) if task_result.get('success'): print(f" ✅ Task processed successfully") print(f" 📊 Summary: {task_result['result']['processing_summary']}") else: print(f" ❌ Task processing failed: {task_result.get('error')}") # Example 2: Intent projection for planning print("\n🔮 Example 2: Strategic Planning with Intent Projection") print(" Intent: Develop autonomous research capabilities") intent_result = await server.handle_tool_call("project_intent_recursive", { "intent_description": "Develop autonomous research capabilities for scientific discovery", "goals": [ {"type": "research_quality", "target": 0.9, "priority": 1.0}, {"type": "automation_level", "target": 0.8, "priority": 0.9} ], "projection_depth": 5, "confidence_threshold": 0.7 }) if intent_result.get('success'): projections = intent_result['result']['projections'] print(f" ✅ Generated {len(projections)} projection steps") for i, proj in enumerate(projections[:3]): # Show first 3 print(f" Step {i+1}: Confidence {proj['confidence']:.2f}") # Example 3: Adaptive agent creation print("\n🤖 Example 3: Adaptive Agent Evolution") print(" Goal: Create specialized code analysis agents") agent_result = await server.handle_tool_call("evolve_agents_scma", { "task_description": "Automated code quality analysis and improvement", "population_size": 15, "generations": 8, "fitness_criteria": [ {"name": "code_quality_detection", "weight": 0.4, "target": 0.9}, {"name": "performance_optimization", "weight": 0.3, "target": 0.8}, {"name": "security_analysis", "weight": 0.3, "target": 0.85} ], "architecture_complexity": "complex" }) if agent_result.get('success'): agents = agent_result['result']['agents_created'] fitness = agent_result['result']['best_fitness'] print(f" ✅ Evolved {agents} specialized agents") print(f" 🏆 Best fitness achieved: {fitness}") await server.handle_tool_call("shutdown_nsaf_framework", {}) print(f"\n🎯 MCP Server Capability Summary:") print(f" • Complete NSAF framework exposed as {len(server.tools)} MCP tools") print(f" • Multi-objective task processing with quantum enhancement") print(f" • Neural intent projection for strategic planning") print(f" • Evolutionary agent optimization for specialized tasks") print(f" • Symbolic memory system for knowledge management") print(f" • Human-AI synergy for collaborative intelligence") print(f" • Foundation model integration across providers") print(f" • Real-time monitoring and system analytics") async def main(): """Run comprehensive MCP server test suite.""" print("🧪 NSAF MCP Server - Comprehensive Test Suite") print("=" * 60) print("Author: Bolorerdene Bundgaa (https://bolor.me)") print() try: # Run test suites await test_tool_discovery() await test_framework_management() await test_task_processing() await test_component_tools() await demonstrate_mcp_capabilities() print(f"\n🎉 All MCP Server Tests Completed Successfully!") print(f"\n📚 Usage Instructions:") print(f" 1. Start MCP server: python3 nsaf_mcp_server.py") print(f" 2. Connect AI assistant via MCP protocol") print(f" 3. Use any of the {len(NSAFMCPServer().tools)} available tools") print(f" 4. Access full NSAF capabilities through natural language") print(f"\n🔧 Available Tool Categories:") print(f" • Framework Management (3 tools)") print(f" • Task Processing (3 tools)") print(f" • Quantum-Symbolic Computing (1 tool)") print(f" • Meta-Agent Evolution (2 tools)") print(f" • Memory Management (3 tools)") print(f" • Intent Projection (1 tool)") print(f" • Human-AI Synergy (1 tool)") print(f" • Foundation Models (2 tools)") print(f" • System Analytics (3 tools)") except Exception as e: print(f"❌ Test suite failed: {str(e)}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main())