#!/usr/bin/env python3
"""Test client for HTTP-based 42crunch MCP Server."""
import requests
import json
import sys
from typing import Dict, Any, Optional
class HTTPMCPClient:
"""Client for communicating with MCP server via JSON-RPC over HTTP."""
def __init__(self, base_url: str = "http://localhost:8000"):
"""Initialize the HTTP MCP client.
Args:
base_url: Base URL of the HTTP server
"""
self.base_url = base_url.rstrip('/')
self.jsonrpc_url = f"{self.base_url}/jsonrpc"
self.request_id = 1
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
"""
request = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
}
if params:
request["params"] = params
self.request_id += 1
print(f"\n📤 Request: {json.dumps(request, indent=2)}")
try:
response = requests.post(
self.jsonrpc_url,
json=request,
headers={"Content-Type": "application/json"},
timeout=30
)
response.raise_for_status()
result = response.json()
print(f"📥 Response: {json.dumps(result, indent=2)}")
return result
except requests.exceptions.RequestException as e:
print(f"❌ HTTP Error: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f" Response: {e.response.text}")
raise
def initialize(self) -> Dict[str, Any]:
"""Initialize the MCP connection."""
return self.send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-http-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 ping(self) -> Dict[str, Any]:
"""Send a ping request."""
return self.send_request("ping")
def health_check(self) -> Dict[str, Any]:
"""Check server health."""
try:
response = requests.get(f"{self.base_url}/health", timeout=5)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"status": "unhealthy", "error": str(e)}
def test_list_collections(client: HTTPMCPClient) -> 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}")
import traceback
traceback.print_exc()
return False
def test_get_collection_apis(client: HTTPMCPClient, collection_id: str = None) -> bool:
"""Test the get_collection_apis tool."""
print("\n" + "="*60)
print("TEST: get_collection_apis")
print("="*60)
if not collection_id:
print("⚠️ Skipping test - no collection_id provided")
return True
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}")
import traceback
traceback.print_exc()
return False
def test_get_api_details(client: HTTPMCPClient, 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")
return True
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}")
import traceback
traceback.print_exc()
return False
def main():
"""Run all tests."""
import argparse
parser = argparse.ArgumentParser(
description="Test client for HTTP-based 42crunch MCP Server",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Run all tests (default: http://localhost:8000)
python test_http_client.py
# Test with custom server URL
python test_http_client.py --url http://localhost:8080
# Test with specific collection ID
python test_http_client.py --collection-id <uuid>
# Test with specific API ID
python test_http_client.py --api-id <uuid>
"""
)
parser.add_argument(
"--url",
type=str,
default="http://localhost:8000",
help="Base URL of the HTTP server (default: http://localhost:8000)"
)
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"
)
args = parser.parse_args()
print("="*60)
print("42crunch MCP HTTP Server Test Client")
print("="*60)
print(f"Server URL: {args.url}")
client = HTTPMCPClient(base_url=args.url)
# Health check
print("\n" + "="*60)
print("HEALTH CHECK")
print("="*60)
health = client.health_check()
print(f"Health: {json.dumps(health, indent=2)}")
if health.get("status") != "healthy":
print("⚠️ Server health check failed, but continuing...")
results = []
try:
# Initialize connection
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}")
# Ping
print("\n" + "="*60)
print("PING")
print("="*60)
try:
ping_response = client.ping()
print(f"✅ Ping: {json.dumps(ping_response, indent=2)}")
except Exception as e:
print(f"⚠️ Ping failed: {e}")
# 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()
