#!/usr/bin/env python3
"""Test client for 42crunch MCP Server.
This client tests the MCP server by sending JSON-RPC requests and validating responses.
"""
import json
import subprocess
import sys
import time
from typing import Dict, Any, Optional
class MCPClient:
"""Client for communicating with MCP server via JSON-RPC over stdio.
Can either start a new server process or connect to an existing running server
via a socket/pipe connection.
"""
def __init__(self, server_command: list[str] = None, connect_to_existing: bool = False, socket_path: str = None):
"""Initialize the MCP client.
Args:
server_command: Command to start the server (default: python main.py)
connect_to_existing: If True, connect to existing server instead of starting new one
socket_path: Path to socket/pipe for connecting to existing server (optional)
"""
self.server_command = server_command or ["python", "main.py"]
self.process: Optional[subprocess.Popen] = None
self.request_id = 1
self.connect_to_existing = connect_to_existing
self.socket_path = socket_path
self._stdin = None
self._stdout = None
def _connect_to_systemd_service(self) -> bool:
"""Try to connect to a systemd service running the MCP server.
For systemd services, we use systemctl to communicate via the service's stdin/stdout.
This uses systemd's socket activation or journal communication.
Returns:
True if connection successful, False otherwise
"""
try:
import subprocess
# Check if service is running
result = subprocess.run(
["systemctl", "is-active", "42crunch-mcp.service"],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0 or result.stdout.strip() != "active":
print(f"Service status: {result.stdout.strip()}")
return False
# For systemd services running stdio servers, we need to use journal communication
# or connect via a socket. The most practical approach is to use systemd's
# socket activation or communicate via journalctl.
# However, for MCP stdio, we typically need direct stdin/stdout access.
# Option: Use systemd-run to execute a command that connects to the service
# This is a workaround - ideally the service should expose a socket
print("⚠️ Systemd service detected, but direct stdio connection not available.")
print(" Consider using the HTTP server (http_main.py) for systemd services.")
print(" Or configure the service to use socket activation.")
return False
except FileNotFoundError:
# systemctl not available (not on Linux or systemd not installed)
return False
except Exception as e:
print(f"Error checking systemd service: {e}")
return False
def _connect_to_existing_server(self) -> bool:
"""Try to connect to an existing server.
Returns:
True if connection successful, False otherwise
"""
# Option 1: Try connecting via named pipe/socket if socket_path provided
if self.socket_path:
try:
import socket
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(self.socket_path)
# Convert socket to file-like objects
self._stdin = sock.makefile('w')
self._stdout = sock.makefile('r')
print(f"✅ Connected to existing server via socket: {self.socket_path}")
return True
except Exception as e:
print(f"❌ Failed to connect via socket: {e}")
return False
# Option 2: Try connecting to systemd service
if self._connect_to_systemd_service():
return True
# Option 3: Try connecting via stdin/stdout if server is running in parent process
try:
import sys
if not sys.stdin.isatty(): # Not a TTY, might be piped
self._stdin = sys.stdin
self._stdout = sys.stdout
print("✅ Connected to existing server via stdin/stdout")
return True
except Exception:
pass
# Option 4: Try to find and connect to a running server process via PID file
try:
pid_file = "/Users/kbarvind/Documents/workspace/GK/mcp/42crunch-mcp.pid"
import os
if os.path.exists(pid_file):
with open(pid_file, 'r') as f:
pid = int(f.read().strip())
# Check if process is still running
try:
os.kill(pid, 0) # Signal 0 just checks if process exists
print(f"⚠️ Found server process (PID: {pid}), but cannot connect via stdio.")
print(" For systemd services, use HTTP server or configure socket activation.")
return False
except ProcessLookupError:
# Process doesn't exist
return False
except Exception:
pass
return False
def start(self) -> None:
"""Start the MCP server process or connect to existing one."""
if self.process or self._stdin:
raise RuntimeError("Already connected to server")
if self.connect_to_existing:
if self._connect_to_existing_server():
return
else:
raise RuntimeError(
"Failed to connect to existing server. "
"Make sure server is running or provide socket_path."
)
# Start new server process
print(f"Starting MCP server: {' '.join(self.server_command)}")
self.process = subprocess.Popen(
self.server_command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
print(f"Server started (PID: {self.process.pid})")
time.sleep(0.5) # Give server time to initialize
def stop(self) -> None:
"""Stop the MCP server process or disconnect."""
if self.process:
print("Stopping server...")
self.process.terminate()
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait()
self.process = None
print("Server stopped")
elif self._stdin:
print("Disconnecting from server...")
try:
self._stdin.close()
self._stdout.close()
except Exception:
pass
self._stdin = None
self._stdout = None
print("Disconnected")
def send_request(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""Send a JSON-RPC request to the server.
Args:
method: JSON-RPC method name
params: Method parameters
Returns:
JSON-RPC response
"""
# Determine which input/output to use
stdin = self._stdin if self._stdin else (self.process.stdin if self.process else None)
stdout = self._stdout if self._stdout else (self.process.stdout if self.process else None)
if not stdin or not stdout:
raise RuntimeError("Server is not running or connected. Call start() first.")
request = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
}
if params:
request["params"] = params
self.request_id += 1
# Send request
request_json = json.dumps(request) + "\n"
print(f"\n📤 Request: {json.dumps(request, indent=2)}")
try:
stdin.write(request_json)
stdin.flush()
except BrokenPipeError:
raise RuntimeError("Server connection terminated unexpectedly")
except OSError as e:
raise RuntimeError(f"Failed to send request: {e}")
# Read response
response_line = stdout.readline()
if not response_line:
raise RuntimeError("No response from server")
try:
response = json.loads(response_line.strip())
print(f"📥 Response: {json.dumps(response, indent=2)}")
return response
except json.JSONDecodeError as e:
raise RuntimeError(f"Invalid JSON response: {response_line.strip()}")
def initialize(self) -> Dict[str, Any]:
"""Initialize the MCP connection."""
return self.send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
})
def list_tools(self) -> Dict[str, Any]:
"""List all available tools."""
return self.send_request("tools/list")
def call_tool(self, name: str, arguments: Dict[str, Any] = None) -> Dict[str, Any]:
"""Call a tool by name.
Args:
name: Tool name
arguments: Tool arguments
Returns:
Tool result
"""
return self.send_request("tools/call", {
"name": name,
"arguments": arguments or {}
})
def __enter__(self):
"""Context manager entry."""
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
self.stop()
def test_list_collections(client: MCPClient) -> bool:
"""Test the list_collections tool."""
print("\n" + "="*60)
print("TEST: list_collections")
print("="*60)
try:
response = client.call_tool("list_collections", {
"page": 1,
"per_page": 10
})
if "error" in response:
print(f"❌ Error: {response['error']}")
return False
result = response.get("result", {})
if result.get("success"):
print("✅ list_collections succeeded")
print(f" Data keys: {list(result.get('data', {}).keys())}")
return True
else:
print(f"❌ Tool returned error: {result.get('error')}")
return False
except Exception as e:
print(f"❌ Exception: {e}")
return False
def test_get_collection_apis(client: MCPClient, collection_id: str = None) -> bool:
"""Test the get_collection_apis tool."""
print("\n" + "="*60)
print("TEST: get_collection_apis")
print("="*60)
# If no collection_id provided, try to get one from list_collections first
if not collection_id:
print("No collection_id provided, trying to get one from list_collections...")
list_response = client.call_tool("list_collections", {"per_page": 1})
if "result" in list_response:
result = list_response["result"]
if result.get("success") and "data" in result:
# Try to extract a collection ID (this depends on the API response structure)
data = result["data"]
if "collections" in data and len(data["collections"]) > 0:
collection_id = data["collections"][0].get("id")
print(f"Using collection_id: {collection_id}")
if not collection_id:
print("⚠️ Skipping test - no collection_id available")
return True # Not a failure, just skip
try:
response = client.call_tool("get_collection_apis", {
"collection_id": collection_id,
"with_tags": True
})
if "error" in response:
print(f"❌ Error: {response['error']}")
return False
result = response.get("result", {})
if result.get("success"):
print("✅ get_collection_apis succeeded")
print(f" Data keys: {list(result.get('data', {}).keys())}")
return True
else:
print(f"❌ Tool returned error: {result.get('error')}")
return False
except Exception as e:
print(f"❌ Exception: {e}")
return False
def test_get_api_details(client: MCPClient, api_id: str = None) -> bool:
"""Test the get_api_details tool."""
print("\n" + "="*60)
print("TEST: get_api_details")
print("="*60)
if not api_id:
print("⚠️ Skipping test - no api_id provided")
print(" To test this, provide an api_id from a collection")
return True # Not a failure, just skip
try:
response = client.call_tool("get_api_details", {
"api_id": api_id,
"branch": "main",
"include_definition": True,
"include_assessment": True,
"include_scan": True
})
if "error" in response:
print(f"❌ Error: {response['error']}")
return False
result = response.get("result", {})
if result.get("success"):
print("✅ get_api_details succeeded")
print(f" Data keys: {list(result.get('data', {}).keys())}")
return True
else:
print(f"❌ Tool returned error: {result.get('error')}")
return False
except Exception as e:
print(f"❌ Exception: {e}")
return False
def main():
"""Run all tests."""
import argparse
parser = argparse.ArgumentParser(
description="Test client for 42crunch MCP Server",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Run all tests
python test_client.py
# Test with specific collection ID
python test_client.py --collection-id <uuid>
# Test with specific API ID
python test_client.py --api-id <uuid>
# Use custom server command
python test_client.py --server "python -m src.server"
"""
)
parser.add_argument(
"--collection-id",
type=str,
help="Collection UUID for testing get_collection_apis"
)
parser.add_argument(
"--api-id",
type=str,
help="API UUID for testing get_api_details"
)
parser.add_argument(
"--server",
type=str,
default="python main.py",
help="Server command (default: 'python main.py')"
)
parser.add_argument(
"--skip-init",
action="store_true",
help="Skip initialization step"
)
parser.add_argument(
"--systemd-service",
action="store_true",
help="Connect to systemd service (will use HTTP if available)"
)
args = parser.parse_args()
server_command = args.server.split()
print("="*60)
print("42crunch MCP Server Test Client")
print("="*60)
if args.systemd_service:
print("\n⚠️ Mode: Connecting to systemd service")
print(" Note: For systemd services, HTTP server is recommended.")
print(" Use test_client_systemd.py --http for HTTP connection.")
print(" Or use test_http_client.py if HTTP server is running.")
sys.exit(0)
elif args.connect_existing:
print("\n⚠️ Mode: Connecting to existing server")
if args.socket_path:
print(f" Socket path: {args.socket_path}")
else:
print("\n⚠️ Mode: Starting new server process")
results = []
try:
client = MCPClient(
server_command=server_command,
connect_to_existing=args.connect_existing,
socket_path=args.socket_path
)
with client:
# Initialize connection
if not args.skip_init:
print("\n" + "="*60)
print("INITIALIZE")
print("="*60)
try:
init_response = client.initialize()
print(f"✅ Initialization: {json.dumps(init_response, indent=2)}")
except Exception as e:
print(f"⚠️ Initialization failed: {e}")
print(" Continuing anyway...")
# List available tools
print("\n" + "="*60)
print("LIST TOOLS")
print("="*60)
try:
tools_response = client.list_tools()
if "result" in tools_response:
tools = tools_response["result"].get("tools", [])
print(f"✅ Found {len(tools)} tools:")
for tool in tools:
print(f" - {tool.get('name')}: {tool.get('description', 'No description')}")
else:
print(f"⚠️ Unexpected response: {tools_response}")
except Exception as e:
print(f"⚠️ Failed to list tools: {e}")
# Run tests
results.append(("list_collections", test_list_collections(client)))
results.append(("get_collection_apis", test_get_collection_apis(client, args.collection_id)))
results.append(("get_api_details", test_get_api_details(client, args.api_id)))
except KeyboardInterrupt:
print("\n\n⚠️ Tests interrupted by user")
sys.exit(1)
except Exception as e:
print(f"\n\n❌ Fatal error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
# Print summary
print("\n" + "="*60)
print("TEST SUMMARY")
print("="*60)
passed = sum(1 for _, result in results if result)
total = len(results)
for test_name, result in results:
status = "✅ PASS" if result else "❌ FAIL"
print(f"{status}: {test_name}")
print(f"\nTotal: {passed}/{total} tests passed")
if passed == total:
print("🎉 All tests passed!")
sys.exit(0)
else:
print("⚠️ Some tests failed")
sys.exit(1)
if __name__ == "__main__":
main()
