#!/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 ") 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()