mcp-vala/test_calculator_debug.py

#!/usr/bin/env python3
"""
Debug script for calculator-server to trace JSON-RPC request handling
"""

import subprocess
import time
import json
import sys
import os
import threading
import queue

class MCPServerTester:
    def __init__(self, server_path):
        self.server_path = server_path
        self.server_process = None
        self.message_queue = queue.Queue()
        
    def start_server(self):
        """Start MCP server process"""
        print(f"Starting MCP server: {self.server_path}")
        self.server_process = subprocess.Popen(
            [self.server_path],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=False,
            bufsize=0
        )
        
        # Start thread to read stderr
        stderr_thread = threading.Thread(target=self._read_stderr)
        stderr_thread.daemon = True
        stderr_thread.start()
        
        # Start thread to read stdout
        stdout_thread = threading.Thread(target=self._read_stdout)
        stdout_thread.daemon = True
        stdout_thread.start()
        
        time.sleep(1)
        
    def _read_stderr(self):
        """Read stderr output from server"""
        while self.server_process and self.server_process.poll() is None:
            try:
                line = self.server_process.stderr.readline()
                if line:
                    print(f"SERVER STDERR: {line.decode().strip()}")
            except:
                break
                
    def _read_stdout(self):
        """Read stdout output from server with proper Content-Length parsing"""
        buffer = b""
        while self.server_process and self.server_process.poll() is None:
            try:
                # Read line by line
                line = self.server_process.stdout.readline()
                if not line:
                    break
                    
                buffer += line
                
                # Check if we have a complete message
                if b'\n' in buffer:
                    # Split on newline to get individual messages
                    lines = buffer.split(b'\n')
                    for i, msg_line in enumerate(lines):
                        if msg_line.strip():
                            try:
                                # Try to parse as JSON
                                message = json.loads(msg_line.decode())
                                self.message_queue.put(message)
                                print(f"RECEIVED: {json.dumps(message, indent=2)}")
                            except json.JSONDecodeError as e:
                                print(f"JSON DECODE ERROR: {e}")
                                print(f"RAW: {msg_line}")
                    # Keep the last incomplete line if any
                    if lines and not lines[-1].strip():
                        buffer = lines[-1]
                    else:
                        buffer = b""
            except:
                break
                        
    def send_jsonrpc_request(self, method, params=None, id=1):
        """Send a JSON-RPC request with Content-Length header"""
        request = {
            "jsonrpc": "2.0",
            "id": id,
            "method": method,
            "params": params or {}
        }
        
        request_str = json.dumps(request)
        # Use the format from test_resources_fixed.py which works
        message = f"{request_str}\n"
        
        print(f"SENDING: {json.dumps(request, indent=2)}")
        print(f"RAW: {message.encode()}")
        
        self.server_process.stdin.write(message.encode())
        self.server_process.stdin.flush()
        
        try:
            response = self.message_queue.get(timeout=10)
            return response
        except queue.Empty:
            print("TIMEOUT: No response received")
            return None
            
    def test_calculator_with_debug(self):
        """Test calculator with the exact request that's failing"""
        print("\n=== Testing Calculator with Debug ===")
        
        # Test with the exact request from the issue
        test_request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "basic_calculator",
                "arguments": {
                    "operation": "add",
                    "operand1": 2,
                    "operand2": 3
                },
                "_meta": {
                    "progressToken": 0
                }
            }
        }
        
        print(f"\nTesting with request:")
        print(json.dumps(test_request, indent=2))
        
        response = self.send_jsonrpc_request("tools/call", test_request["params"], id=1)
        
        if response:
            print(f"\nResponse received:")
            print(json.dumps(response, indent=2))
            
            if "result" in response:
                result = response["result"]
                if "content" in result and len(result["content"]) > 0:
                    content = result["content"][0]
                    if "text" in content:
                        print(f"\nResult text: {content['text']}")
                    if "structured_content" in result:
                        print(f"Structured content: {result['structured_content']}")
                        
                        # Check if operands are correctly parsed
                        if isinstance(result['structured_content'], dict):
                            if 'operand1' in result['structured_content']:
                                print(f"operand1 in result: {result['structured_content']['operand1']}")
                            if 'operand2' in result['structured_content']:
                                print(f"operand2 in result: {result['structured_content']['operand2']}")
                            if 'result' in result['structured_content']:
                                print(f"calculation result: {result['structured_content']['result']}")
                else:
                    print("No content in response")
            elif "error" in response:
                print(f"Error in response: {response['error']}")
        else:
            print("No response received")
            
    def test_simple_request(self):
        """Test with a simpler request"""
        print("\n=== Testing Simple Request ===")
        
        # Test tools/list first
        print("\nTesting tools/list...")
        list_response = self.send_jsonrpc_request("tools/list", id=2)
        
        if list_response:
            print(f"Tools list response:")
            print(json.dumps(list_response, indent=2))
        else:
            print("No response for tools/list")
            
    def stop_server(self):
        """Stop MCP server process"""
        if self.server_process:
            print("\nStopping server...")
            self.server_process.terminate()
            try:
                self.server_process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.server_process.kill()
            self.server_process = None

def main():
    if len(sys.argv) != 2:
        print("Usage: python3 test_calculator_debug.py <server-executable>")
        sys.exit(1)
        
    server_path = sys.argv[1]
    if not os.path.exists(server_path):
        print(f"Error: Server executable '{server_path}' not found")
        sys.exit(1)
        
    tester = MCPServerTester(server_path)
    
    try:
        tester.start_server()
        time.sleep(2)
        
        # Initialize first
        print("\nInitializing server...")
        init_params = {
            "protocolVersion": "2025-11-25",
            "capabilities": {},
            "clientInfo": {
                "name": "test-client",
                "version": "1.0.0"
            }
        }
        init_response = tester.send_jsonrpc_request("initialize", init_params, id=0)
        
        if init_response:
            print("✓ Server initialized successfully")
            
            # Test simple request first
            tester.test_simple_request()
            
            # Test calculator with debug
            tester.test_calculator_with_debug()
        else:
            print("✗ Failed to initialize server")
            
    except KeyboardInterrupt:
        print("\nTest interrupted by user")
    except Exception as e:
        print(f"\nError during test: {e}")
        import traceback
        traceback.print_exc()
    finally:
        tester.stop_server()

if __name__ == "__main__":
    main()