mcp-vala/debug_prompt_serialization.py

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