mcp-vala/diagnose_serialization.py

#!/usr/bin/env python3
"""
Diagnostic script to test content serialization and identify the exact issue
"""

import subprocess
import json
import sys
import os
import tempfile

def test_content_serialization():
    """Test content serialization by creating a simple test case"""
    
    # Create a test JSON that matches what the server should send
    test_prompt_response = {
        "jsonrpc": "2.0",
        "id": 2,
        "result": {
            "description": "Test prompt",
            "messages": [
                {
                    "role": "user",
                    "content": {
                        "type": "text",
                        "text": "Hello, world!"
                    }
                },
                {
                    "role": "assistant", 
                    "content": {
                        "type": "image",
                        "data": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==",
                        "mimeType": "image/png"
                    }
                },
                {
                    "role": "user",
                    "content": {
                        "type": "resource",
                        "resource": {
                            "uri": "file:///example.txt",
                            "text": "This is a test resource"
                        }
                    }
                }
            ]
        }
    }
    
    print("Testing JSON structure that should be valid:")
    print(json.dumps(test_prompt_response, indent=2))
    
    # Try to validate this against a simple schema
    try:
        # Basic validation of the structure
        result = test_prompt_response["result"]
        messages = result["messages"]
        
        for i, message in enumerate(messages):
            content = message["content"]
            content_type = content["type"]
            
            print(f"\nMessage {i} - Type: {content_type}")
            
            if content_type == "text":
                if "text" not in content:
                    print(f"  ✗ Missing 'text' field")
                else:
                    print(f"  ✓ Has 'text' field")
                    
            elif content_type in ["image", "audio"]:
                if "data" not in content:
                    print(f"  ✗ Missing 'data' field")
                else:
                    print(f"  ✓ Has 'data' field")
                    
                if "mimeType" not in content:
                    print(f"  ✗ Missing 'mimeType' field")
                else:
                    print(f"  ✓ Has 'mimeType' field: {content['mimeType']}")
                    
            elif content_type == "resource":
                if "resource" not in content:
                    print(f"  ✗ Missing 'resource' field")
                else:
                    resource = content["resource"]
                    print(f"  ✓ Has 'resource' field with keys: {list(resource.keys())}")
                    
                    if "uri" not in resource:
                        print(f"    ✗ Missing 'uri' in resource")
                    else:
                        print(f"    ✓ Has 'uri': {resource['uri']}")
                        
                    # Check if resource has either text or blob
                    has_text = "text" in resource
                    has_blob = "blob" in resource
                    
                    if not (has_text or has_blob):
                        print(f"    ✗ Resource must have either 'text' or 'blob'")
                    elif has_text:
                        print(f"    ✓ Has 'text' content")
                    elif has_blob:
                        print(f"    ✓ Has 'blob' content")
                        
            elif content_type == "resource_link":
                required_fields = ["name", "uri"]
                for field in required_fields:
                    if field not in content:
                        print(f"  ✗ Missing '{field}' field")
                    else:
                        print(f"  ✓ Has '{field}': {content[field]}")
                        
        print("\n✓ Basic structure validation passed")
        return True
        
    except Exception as e:
        print(f"\n✗ Validation failed: {e}")
        import traceback
        traceback.print_exc()
        return False

def test_server_response():
    """Test actual server response if available"""
    if len(sys.argv) != 2:
        print("Usage: python3 diagnose_serialization.py <server-executable>")
        return False
        
    server_path = sys.argv[1]
    if not os.path.exists(server_path):
        print(f"Error: Server executable '{server_path}' not found")
        return False
    
    print(f"\n{'='*60}")
    print(f"Testing server: {server_path}")
    print(f"{'='*60}")
    
    # Test the expected structure first
    if not test_content_serialization():
        print("✗ Basic structure validation failed")
        return False
    
    # Try to test the actual server
    try:
        # Create a simple test script
        test_script = '''
import json
import sys
import subprocess

# Start the server
proc = subprocess.Popen([sys.argv[1]], 
                      stdin=subprocess.PIPE, 
                      stdout=subprocess.PIPE, 
                      stderr=subprocess.PIPE,
                      text=False)

# Send initialize
init_msg = {
    "jsonrpc": "2.0",
    "method": "initialize", 
    "id": 0,
    "params": {
        "protocolVersion": "2025-11-25",
        "capabilities": {},
        "clientInfo": {"name": "test", "version": "1.0.0"}
    }
}

msg_str = json.dumps(init_msg)
proc.stdin.write(f"Content-Length: {len(msg_str)}\\r\\n\\r\\n{msg_str}".encode())
proc.stdin.flush()

# Read init response
line = proc.stdout.readline()
if line.startswith(b"Content-Length:"):
    length = int(line.decode().split(":")[1].strip())
    proc.stdout.readline()  # empty line
    data = proc.stdout.read(length)
    response = json.loads(data.decode())
    print("Init response:", "success" if "result" in response else "failed")

# List prompts
list_msg = {"jsonrpc": "2.0", "method": "prompts/list", "id": 1, "params": {}}
msg_str = json.dumps(list_msg)
proc.stdin.write(f"Content-Length: {len(msg_str)}\\r\\n\\r\\n{msg_str}".encode())
proc.stdin.flush()

# Read list response
line = proc.stdout.readline()
if line.startswith(b"Content-Length:"):
    length = int(line.decode().split(":")[1].strip())
    proc.stdout.readline()  # empty line
    data = proc.stdout.read(length)
    response = json.loads(data.decode())
    
    if "result" in response and "prompts" in response["result"]:
        prompts = response["result"]["prompts"]
        print(f"Found {len(prompts)} prompts")
        
        if prompts:
            # Get first prompt
            get_msg = {"jsonrpc": "2.0", "method": "prompts/get", "id": 2, 
                      "params": {"name": prompts[0]["name"]}}
            msg_str = json.dumps(get_msg)
            proc.stdin.write(f"Content-Length: {len(msg_str)}\\r\\n\\r\\n{msg_str}".encode())
            proc.stdin.flush()
            
            # Read get response
            line = proc.stdout.readline()
            if line.startswith(b"Content-Length:"):
                length = int(line.decode().split(":")[1].strip())
                proc.stdout.readline()  # empty line
                data = proc.stdout.read(length)
                response = json.loads(data.decode())
                
                if "result" in response:
                    print("✓ Got prompt result")
                    result = response["result"]
                    print(json.dumps(result, indent=2))
                elif "error" in response:
                    print("✗ Got error:")
                    print(json.dumps(response["error"], indent=2))
                else:
                    print("✗ No result or error in response")

proc.terminate()
'''
        
        # Write test script to temp file
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(test_script)
            temp_script = f.name
        
        # Run the test
        result = subprocess.run([sys.executable, temp_script, server_path], 
                              capture_output=True, text=True, timeout=10)
        
        print("Server test output:")
        print(result.stdout)
        if result.stderr:
            print("Server test errors:")
            print(result.stderr)
            
        # Clean up
        os.unlink(temp_script)
        
        return result.returncode == 0
        
    except subprocess.TimeoutExpired:
        print("✗ Server test timed out")
        return False
    except Exception as e:
        print(f"✗ Server test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

if __name__ == "__main__":
    if len(sys.argv) == 2:
        success = test_server_response()
    else:
        success = test_content_serialization()
    
    sys.exit(0 if success else 1)