#!/usr/bin/env python3 """ Debug script to test prompt serialization and identify the exact issue """ 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 Content-Length parsing""" while self.server_process and self.server_process.poll() is None: try: # Read Content-Length header line = self.server_process.stdout.readline() if not line: break line_str = line.decode().strip() if line_str.startswith("Content-Length:"): content_length = int(line_str.split(":")[1].strip()) # Read empty line self.server_process.stdout.readline() # Read JSON content json_data = self.server_process.stdout.read(content_length) if json_data: try: parsed = json.loads(json_data.decode()) self.message_queue.put(parsed) print(f"SERVER MESSAGE: {json.dumps(parsed, indent=2)}") except json.JSONDecodeError as e: print(f"JSON DECODE ERROR: {e}") print(f"RAW DATA: {json_data}") 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", "method": method, "id": id, "params": params or {} } request_str = json.dumps(request) message = f"Content-Length: {len(request_str)}\r\n\r\n{request_str}" print(f"SENDING: {repr(message)}") 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_prompt_get(self): """Test prompts/get to trigger the serialization issue""" print("\n=== Testing Prompt Get ===") # Initialize first init_params = { "protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": { "name": "test-client", "version": "1.0.0" } } init_response = self.send_jsonrpc_request("initialize", init_params, id=0) if init_response and "result" in init_response: print("✓ Server initialized successfully") else: print("✗ Failed to initialize server") return False # List prompts first print("\nListing prompts...") list_response = self.send_jsonrpc_request("prompts/list", id=1) if list_response and "result" in list_response: prompts = list_response["result"].get("prompts", []) print(f"✓ Found {len(prompts)} prompts") if prompts: # Try to get the first prompt prompt_name = prompts[0]["name"] print(f"\nGetting prompt: {prompt_name}") get_response = self.send_jsonrpc_request("prompts/get", { "name": prompt_name }, id=2) if get_response: if "result" in get_response: print("✓ Prompt retrieved successfully") result = get_response["result"] # Check the structure of the response if "messages" in result: messages = result["messages"] print(f"✓ Found {len(messages)} messages") for i, message in enumerate(messages): print(f"\nMessage {i}:") print(f" Role: {message.get('role')}") if "content" in message: content = message["content"] print(f" Content type: {content.get('type')}") print(f" Content keys: {list(content.keys())}") # Check for specific issues if content.get("type") in ["image", "audio"]: if "mimeType" in content: print(f" ✓ Has mimeType: {content['mimeType']}") else: print(" ✗ Missing mimeType field") if "mime_type" in content: print(f" ✗ Has mime_type instead of mimeType: {content['mime_type']}") if "data" in content: print(f" ✓ Has data field") else: print(" ✗ Missing data field") elif content.get("type") == "resource_link": if "name" in content: print(f" ✓ Has name: {content['name']}") else: print(" ✗ Missing name field") if "uri" in content: print(f" ✓ Has uri: {content['uri']}") else: print(" ✗ Missing uri field") elif content.get("type") == "resource": if "resource" in content: print(f" ✓ Has resource field") resource = content["resource"] print(f" Resource keys: {list(resource.keys())}") else: print(" ✗ Missing resource field") else: print("✗ No messages field in response") else: print("✗ No result field in response") if "error" in get_response: error = get_response["error"] print(f"ERROR: {json.dumps(error, indent=2)}") else: print("✗ Failed to get prompt") else: print("✗ No prompts found") else: print("✗ Failed to list prompts") return False 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 debug_prompt_serialization.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) # Test prompt get tester.test_prompt_get() 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()