mcp-vala/debug_templates.py

#!/usr/bin/env python3
"""
Debug script to check if templates are being handled properly
"""

import subprocess
import time
import json
import sys
import os
import threading
import queue
import tempfile
import shutil

class MCPServerTester:
    def __init__(self, server_path):
        self.server_path = server_path
        self.server_process = None
        self.message_queue = queue.Queue()
        self.test_dir = None
        
    def start_server(self):
        """Start MCP server process"""
        # Create a temporary directory for testing
        self.test_dir = tempfile.mkdtemp(prefix="mcp_test_")
        
        # Create some test files
        with open(os.path.join(self.test_dir, "test.txt"), "w") as f:
            f.write("Hello, World!")
            
        print(f"Starting MCP server: {self.server_path} with test directory: {self.test_dir}")
        self.server_process = subprocess.Popen(
            [self.server_path, self.test_dir],
            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 without Content-Length parsing"""
        buffer = b""
        while self.server_process and self.server_process.poll() is None:
            try:
                data = self.server_process.stdout.readline()
                if not data:
                    break
                buffer += data
                
                # Try to parse JSON directly from each line
                try:
                    message = buffer.strip()
                    if message:
                        parsed = json.loads(message.decode())
                        self.message_queue.put(parsed)
                        print(f"SERVER MESSAGE: {json.dumps(parsed, indent=2)}")
                        buffer = b""
                except json.JSONDecodeError:
                    # If not valid JSON yet, continue reading
                    continue
            except:
                break
                        
    def send_jsonrpc_request(self, method, params=None, id=1):
        """Send a JSON-RPC request without Content-Length header"""
        request = {
            "jsonrpc": "2.0",
            "method": method,
            "id": id,
            "params": params or {}
        }
            
        request_str = json.dumps(request)
        message = f"{request_str}\n"
        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_templates(self):
        """Test resource template functionality"""
        print("\n=== Testing Resource Templates ===")
        
        # 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:
            print("✓ Server initialized successfully")
            print(f"Init response: {json.dumps(init_response, indent=2)}")
        else:
            print("✗ Failed to initialize server")
            return False
        
        # Test templates/list
        print("\nTesting resources/templates/list...")
        templates_response = self.send_jsonrpc_request("resources/templates/list", id=7)
        
        if templates_response:
            print(f"Templates response: {json.dumps(templates_response, indent=2)}")
            
            if "result" in templates_response:
                result = templates_response["result"]
                if "templates" in result:
                    templates = result["templates"]
                    print(f"✓ Found {len(templates)} templates")
                    return True
                else:
                    print("✗ No templates field in response")
                    return False
            else:
                print("✗ No result field in response")
                return False
        else:
            print("✗ Failed to list templates")
            return False
            
    def cleanup(self):
        """Clean up test environment"""
        if self.test_dir and os.path.exists(self.test_dir):
            shutil.rmtree(self.test_dir)
            
    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_templates.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 templates
        tester.test_templates()
            
    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()
        tester.cleanup()

if __name__ == "__main__":
    main()