MCP Wikipedia Server

""" MCP Protocol Compliance Tests for Wikipedia Server. This module verifies that the server properly implements the Model Context Protocol specification and can communicate correctly with MCP clients. """ import asyncio import json import subprocess import time import sys import os from typing import Dict, Any, List, Optional # Add src to path for imports sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) try: from mcp_server.mcp_server import WikipediaServer except ImportError: sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src', 'mcp_server')) from mcp_server import WikipediaServer class MCPProtocolTester: """Test MCP protocol compliance for the Wikipedia server.""" def __init__(self): self.server_process = None self.test_results = [] async def start_server_process(self) -> bool: """Start the MCP server as a subprocess.""" try: server_path = os.path.join( os.path.dirname(__file__), '..', 'src', 'mcp_server', 'mcp_server.py' ) self.server_process = subprocess.Popen( [sys.executable, server_path], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=0 ) # Wait a moment for server to start await asyncio.sleep(1) if self.server_process.poll() is None: print("✅ Server process started successfully") return True else: print("❌ Server process failed to start") return False except Exception as e: print(f"❌ Failed to start server process: {e}") return False def stop_server_process(self): """Stop the server process if it's running.""" if self.server_process: try: self.server_process.terminate() self.server_process.wait(timeout=5) print("✅ Server process stopped") except subprocess.TimeoutExpired: self.server_process.kill() print("⚠️ Server process killed (timeout)") except Exception as e: print(f"⚠️ Error stopping server process: {e}") def send_mcp_message(self, message: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Send an MCP message to the server and get response.""" if not self.server_process: return None try: # Send JSON-RPC message message_json = json.dumps(message) + '\n' self.server_process.stdin.write(message_json) self.server_process.stdin.flush() # Read response (with timeout) response_line = self.server_process.stdout.readline() if response_line: return json.loads(response_line.strip()) else: return None except Exception as e: print(f"Error sending MCP message: {e}") return None def test_initialization(self) -> bool: """Test MCP initialization handshake.""" print("\n🔧 Testing MCP initialization...") # Send initialize request init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": { "roots": { "listChanged": True }, "sampling": {} }, "clientInfo": { "name": "test-client", "version": "1.0.0" } } } response = self.send_mcp_message(init_request) if response and response.get("id") == 1: print(" ✅ Initialize request successful") # Send initialized notification initialized_notification = { "jsonrpc": "2.0", "method": "notifications/initialized" } self.send_mcp_message(initialized_notification) print(" ✅ Initialized notification sent") return True else: print(f" ❌ Initialize failed: {response}") return False def test_list_tools(self) -> bool: """Test listing available tools.""" print("\n🛠️ Testing tools/list...") list_tools_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/list" } response = self.send_mcp_message(list_tools_request) if response and response.get("id") == 2: tools = response.get("result", {}).get("tools", []) expected_tools = [ "fetch_wikipedia_info", "list_wikipedia_sections", "get_section_content" ] found_tools = [tool["name"] for tool in tools] print(f" Found tools: {found_tools}") all_found = all(tool in found_tools for tool in expected_tools) if all_found: print(" ✅ All expected tools found") return True else: missing = [tool for tool in expected_tools if tool not in found_tools] print(f" ❌ Missing tools: {missing}") return False else: print(f" ❌ List tools failed: {response}") return False def test_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> bool: """Test calling a specific tool.""" print(f"\n🔧 Testing {tool_name} tool call...") tool_call_request = { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": tool_name, "arguments": arguments } } response = self.send_mcp_message(tool_call_request) if response and response.get("id") == 3: result = response.get("result") if result and "content" in result: content = result["content"] if content and len(content) > 0: # Try to parse the JSON content try: tool_result = json.loads(content[0]["text"]) if "success" in tool_result and "metadata" in tool_result: print(f" ✅ {tool_name} call successful") print(f" Success: {tool_result['success']}") return True else: print(f" ❌ Invalid tool result format: {tool_result}") return False except json.JSONDecodeError as e: print(f" ❌ Failed to parse tool result JSON: {e}") return False else: print(" ❌ Empty content in tool response") return False else: print(f" ❌ Invalid tool response format: {result}") return False else: print(f" ❌ Tool call failed: {response}") return False def test_error_handling(self) -> bool: """Test error handling for invalid requests.""" print("\n🚨 Testing error handling...") # Test invalid method invalid_request = { "jsonrpc": "2.0", "id": 4, "method": "invalid/method" } response = self.send_mcp_message(invalid_request) if response and response.get("error"): print(" ✅ Invalid method properly rejected") # Test invalid tool call invalid_tool_request = { "jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": { "name": "nonexistent_tool", "arguments": {} } } response = self.send_mcp_message(invalid_tool_request) if response and response.get("error"): print(" ✅ Invalid tool call properly rejected") return True else: print(" ❌ Invalid tool call not properly rejected") return False else: print(" ❌ Invalid method not properly rejected") return False def test_message_format_compliance(self) -> bool: """Test JSON-RPC 2.0 message format compliance.""" print("\n📋 Testing JSON-RPC 2.0 compliance...") # Test missing required fields test_cases = [ { "name": "missing jsonrpc", "message": {"id": 1, "method": "tools/list"}, "should_error": True }, { "name": "wrong jsonrpc version", "message": {"jsonrpc": "1.0", "id": 1, "method": "tools/list"}, "should_error": True }, { "name": "missing id", "message": {"jsonrpc": "2.0", "method": "tools/list"}, "should_error": False # Notifications don't need id } ] all_passed = True for case in test_cases: response = self.send_mcp_message(case["message"]) has_error = response and "error" in response if case["should_error"] and not has_error: print(f" ❌ {case['name']}: Should have errored but didn't") all_passed = False elif not case["should_error"] and has_error: print(f" ❌ {case['name']}: Should not have errored but did") all_passed = False else: print(f" ✅ {case['name']}: Handled correctly") return all_passed async def run_full_compliance_test(self) -> Dict[str, Any]: """Run complete MCP protocol compliance test suite.""" print("🚀 Starting MCP Protocol Compliance Tests") print("="*50) results = { "server_start": False, "initialization": False, "list_tools": False, "tool_calls": {}, "error_handling": False, "message_format": False, "overall_success": False } try: # Start server if not await self.start_server_process(): return results results["server_start"] = True # Test initialization results["initialization"] = self.test_initialization() if results["initialization"]: # Test listing tools results["list_tools"] = self.test_list_tools() # Test individual tool calls tool_tests = [ ("fetch_wikipedia_info", {"query": "Python programming"}), ("list_wikipedia_sections", {"topic": "Machine Learning"}), ("get_section_content", {"topic": "Python", "section_title": "History"}) ] for tool_name, arguments in tool_tests: results["tool_calls"][tool_name] = self.test_tool_call(tool_name, arguments) # Test error handling results["error_handling"] = self.test_error_handling() # Test message format compliance results["message_format"] = self.test_message_format_compliance() # Determine overall success tool_calls_success = all(results["tool_calls"].values()) results["overall_success"] = all([ results["server_start"], results["initialization"], results["list_tools"], tool_calls_success, results["error_handling"], results["message_format"] ]) except Exception as e: print(f"\n❌ Compliance test failed with exception: {e}") finally: self.stop_server_process() return results def print_compliance_summary(self, results: Dict[str, Any]): """Print a summary of compliance test results.""" print("\n" + "="*50) print("📊 MCP PROTOCOL COMPLIANCE SUMMARY") print("="*50) status_symbol = lambda success: "✅" if success else "❌" print(f"{status_symbol(results['server_start'])} Server Startup") print(f"{status_symbol(results['initialization'])} MCP Initialization") print(f"{status_symbol(results['list_tools'])} Tools Listing") print("\n🛠️ Tool Call Tests:") for tool_name, success in results["tool_calls"].items(): print(f" {status_symbol(success)} {tool_name}") print(f"\n{status_symbol(results['error_handling'])} Error Handling") print(f"{status_symbol(results['message_format'])} Message Format Compliance") print(f"\n🎯 Overall Success: {status_symbol(results['overall_success'])}") if results["overall_success"]: print("\n🎉 All MCP protocol compliance tests passed!") else: print("\n⚠️ Some compliance tests failed. Review the details above.") async def run_mcp_compliance_tests(): """Main function to run MCP compliance tests.""" tester = MCPProtocolTester() results = await tester.run_full_compliance_test() tester.print_compliance_summary(results) return results if __name__ == "__main__": # Run compliance tests when script is executed directly asyncio.run(run_mcp_compliance_tests())