Multi Agent Orchestrator MCP

#!/usr/bin/env python3 """ Test script to verify the Smithery deployment fix Simulates Smithery's MCP scanning process with proper JSON-RPC protocol """ import os import sys import json import time import requests import subprocess from typing import Dict, Any, Optional def test_mcp_json_rpc_endpoints(base_url: str = "http://localhost:8080") -> Dict[str, Any]: """Test MCP JSON-RPC endpoints that Smithery uses for scanning""" results = { "base_url": base_url, "tests": {}, "overall_success": True, "smithery_compatible": True } mcp_endpoint = f"{base_url}/mcp" print(f"🔍 Testing MCP JSON-RPC Protocol for Smithery compatibility...") print(f"🌐 MCP Endpoint: {mcp_endpoint}") print() # Standard MCP headers required by the protocol headers = { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "MCP-Protocol-Version": "2025-06-18" } # Critical MCP JSON-RPC requests that Smithery uses mcp_requests = { "initialize": { "description": "MCP initialization - CRITICAL for Smithery", "payload": { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": { "name": "smithery-test-client", "version": "1.0.0" } } }, "critical": True }, "tools_list": { "description": "Tool discovery - CRITICAL for Smithery", "payload": { "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }, "critical": True }, "resources_list": { "description": "Resource discovery - Optional", "payload": { "jsonrpc": "2.0", "id": 3, "method": "resources/list", "params": {} }, "critical": False }, "prompts_list": { "description": "Prompt discovery - Optional", "payload": { "jsonrpc": "2.0", "id": 4, "method": "prompts/list", "params": {} }, "critical": False } } for test_name, test_info in mcp_requests.items(): try: print(f"📡 Testing {test_name} - {test_info['description']}") # Make JSON-RPC request without authentication (as Smithery would during discovery) response = requests.post( mcp_endpoint, headers=headers, json=test_info["payload"], timeout=10 ) test_result = { "endpoint": mcp_endpoint, "description": test_info["description"], "request": test_info["payload"], "status_code": response.status_code, "success": False, "response_data": None, "error": None, "smithery_critical": test_info["critical"] } # Check if response is successful (200-299 range) if 200 <= response.status_code < 300: test_result["success"] = True try: response_data = response.json() test_result["response_data"] = response_data # Validate JSON-RPC response if "jsonrpc" in response_data and "id" in response_data: if "result" in response_data: print(f" ✅ SUCCESS (HTTP {response.status_code}) - Valid JSON-RPC response") elif "error" in response_data: print(f" ⚠️ JSON-RPC ERROR: {response_data['error']}") if test_info["critical"]: results["smithery_compatible"] = False else: print(f" ⚠️ Invalid JSON-RPC response format") else: print(f" ⚠️ Non-JSON-RPC response: {str(response_data)[:100]}") except: test_result["response_data"] = response.text[:200] + "..." if len(response.text) > 200 else response.text print(f" ✅ SUCCESS (HTTP {response.status_code}) - Non-JSON response") elif response.status_code == 401: print(f" ❌ AUTHENTICATION REQUIRED (HTTP 401) - Discovery should be open!") test_result["error"] = f"HTTP 401 - Authentication required for discovery" results["smithery_compatible"] = False if test_info["critical"]: results["overall_success"] = False elif response.status_code == 403: print(f" ❌ FORBIDDEN (HTTP 403) - Discovery should be accessible!") test_result["error"] = f"HTTP 403 - Forbidden" results["smithery_compatible"] = False if test_info["critical"]: results["overall_success"] = False elif response.status_code == 404: print(f" ❌ NOT FOUND (HTTP 404) - MCP endpoint not available") test_result["error"] = f"HTTP 404 - MCP endpoint not found" results["smithery_compatible"] = False if test_info["critical"]: results["overall_success"] = False elif response.status_code == 405: print(f" ❌ METHOD NOT ALLOWED (HTTP 405) - POST should be supported") test_result["error"] = f"HTTP 405 - Method not allowed" results["smithery_compatible"] = False if test_info["critical"]: results["overall_success"] = False elif response.status_code == 406: print(f" ❌ NOT ACCEPTABLE (HTTP 406) - Headers issue") test_result["error"] = f"HTTP 406 - Not acceptable" results["smithery_compatible"] = False if test_info["critical"]: results["overall_success"] = False else: print(f" ❌ FAILED (HTTP {response.status_code})") test_result["error"] = f"HTTP {response.status_code}" if test_info["critical"]: results["smithery_compatible"] = False results["overall_success"] = False results["tests"][test_name] = test_result except requests.exceptions.RequestException as e: print(f" ❌ CONNECTION ERROR: {str(e)}") results["tests"][test_name] = { "endpoint": mcp_endpoint, "description": test_info["description"], "status_code": None, "success": False, "error": str(e), "smithery_critical": test_info["critical"] } results["overall_success"] = False if test_info["critical"]: results["smithery_compatible"] = False print() return results def test_http_endpoints(base_url: str = "http://localhost:8080") -> Dict[str, Any]: """Test basic HTTP endpoints for server health""" results = { "base_url": base_url, "tests": {}, "overall_success": True } print(f"🔍 Testing Basic HTTP Endpoints...") print(f"🌐 Base URL: {base_url}") print() # Basic HTTP endpoints endpoints = { "/health": "Server health check" } for endpoint, description in endpoints.items(): test_name = endpoint.replace("/", "_").strip("_") or "root" url = f"{base_url}{endpoint}" try: print(f"📡 Testing {endpoint} - {description}") response = requests.get(url, timeout=10) test_result = { "url": url, "description": description, "status_code": response.status_code, "success": False, "response_data": None, "error": None } if 200 <= response.status_code < 300: test_result["success"] = True try: test_result["response_data"] = response.json() except: test_result["response_data"] = response.text[:200] + "..." if len(response.text) > 200 else response.text print(f" ✅ SUCCESS (HTTP {response.status_code})") else: print(f" ❌ FAILED (HTTP {response.status_code})") test_result["error"] = f"HTTP {response.status_code}" results["overall_success"] = False results["tests"][test_name] = test_result except requests.exceptions.RequestException as e: print(f" ❌ CONNECTION ERROR: {str(e)}") results["tests"][test_name] = { "url": url, "description": description, "status_code": None, "success": False, "error": str(e) } results["overall_success"] = False print() return results def test_production_simulation(): """Test with simulated production environment variables (as Smithery would set them)""" print("🧪 TESTING PRODUCTION SIMULATION") print("=" * 50) # Backup original environment original_env = { "DESCOPE_PROJECT_ID": os.environ.get("DESCOPE_PROJECT_ID"), "DESCOPE_MANAGEMENT_KEY": os.environ.get("DESCOPE_MANAGEMENT_KEY") } try: # Set simulated Smithery production credentials os.environ["DESCOPE_PROJECT_ID"] = "P2l8xGh4KQf3wXr9mN5yZ1vB" # Fake production-like project ID os.environ["DESCOPE_MANAGEMENT_KEY"] = "K1:fake_management_key_for_testing_purposes_only" # Fake management key print(f"🔧 Set simulated production credentials:") print(f" DESCOPE_PROJECT_ID: P2l8xGh4KQf3wXr9mN5yZ1vB") print(f" DESCOPE_MANAGEMENT_KEY: K1:fake_management_key...") print() # Test server startup with production credentials print("🚀 Testing server startup with production credentials...") # Start server as subprocess server_process = subprocess.Popen([ sys.executable, "mcp_server.py" ], cwd=os.getcwd(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) # Wait for server to start time.sleep(3) if server_process.poll() is None: print("✅ Server started successfully with production credentials") # Test basic endpoints first http_results = test_http_endpoints() # Test MCP protocol endpoints mcp_results = test_mcp_json_rpc_endpoints() # Stop server server_process.terminate() server_process.wait(timeout=5) return { "server_startup": True, "http_tests": http_results, "mcp_tests": mcp_results, "smithery_compatible": mcp_results.get("smithery_compatible", False) } else: print("❌ Server failed to start with production credentials") output, _ = server_process.communicate() print(f"Server output: {output}") return { "server_startup": False, "error": "Server startup failed", "output": output, "smithery_compatible": False } finally: # Restore original environment for key, value in original_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value def main(): """Main test function""" print("🔥 SMITHERY DEPLOYMENT FIX VERIFICATION") print("=" * 60) print() # Test 1: Current demo mode print("📋 TEST 1: Demo Mode (Current Configuration)") print("-" * 40) try: http_results = test_http_endpoints() mcp_results = test_mcp_json_rpc_endpoints() demo_compatible = mcp_results.get("smithery_compatible", False) except Exception as e: print(f"❌ Demo mode test failed: {e}") demo_compatible = False print("\n📋 TEST 2: Production Mode Simulation") print("-" * 40) prod_results = test_production_simulation() prod_compatible = prod_results.get("smithery_compatible", False) if isinstance(prod_results, dict) else False # Summary print("\n📊 FINAL RESULTS") print("=" * 30) print(f"Demo Mode Smithery Compatible: {'✅ YES' if demo_compatible else '❌ NO'}") print(f"Production Mode Smithery Compatible: {'✅ YES' if prod_compatible else '❌ NO'}") overall_fix_success = demo_compatible and prod_compatible print(f"\n🎯 OVERALL FIX STATUS: {'✅ SUCCESS' if overall_fix_success else '❌ NEEDS MORE WORK'}") if overall_fix_success: print("\n🎉 Smithery deployment should now work successfully!") print(" • MCP JSON-RPC discovery endpoints are accessible without authentication") print(" • Server starts reliably with any credential configuration") print(" • Graceful degradation prevents startup failures") print(" • Proper MCP protocol support for tool discovery") else: print("\n⚠️ Issues still need to be resolved:") if not demo_compatible: print(" • Demo mode MCP discovery endpoints are not working") if not prod_compatible: print(" • Production mode has issues with credential handling") return overall_fix_success if __name__ == "__main__": success = main() sys.exit(0 if success else 1)