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