mcp-vala/debug_resources.py

#!/usr/bin/env python3
"""
Debug resource handling to see actual response format
"""

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 with Content-Length parsing"""
        buffer = b""
        while self.server_process and self.server_process.poll() is None:
            try:
                data = self.server_process.stdout.read(1)
                if not data:
                    break
                buffer += data
                
                if b'\r\n\r\n' in buffer:
                    headers, body = buffer.split(b'\r\n\r\n', 1)
                    content_length = None
                    
                    for line in headers.split(b'\r\n'):
                        if line.lower().startswith(b'content-length:'):
                            content_length = int(line.split(b':')[1].strip())
                            break
                    
                    if content_length is not None and len(body) >= content_length:
                        message = body[:content_length]
                        buffer = body[content_length:]
                        
                        try:
                            parsed = json.loads(message.decode())
                            self.message_queue.put(parsed)
                        except json.JSONDecodeError as e:
                            print(f"JSON DECODE ERROR: {e}")
                            print(f"RAW MESSAGE: {message}")
            except:
                break
                        
    def send_jsonrpc_request(self, method, params=None, id=1):
        """Send a JSON-RPC request with proper 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}"
        
        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 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_resources.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)
        
        # Initialize first
        init_params = {
            "protocolVersion": "2025-11-25",
            "capabilities": {},
            "clientInfo": {
                "name": "test-client",
                "version": "1.0.0"
            }
        }
        init_response = tester.send_jsonrpc_request("initialize", init_params, id=0)
        
        if init_response:
            print("✓ Server initialized successfully")
            
            # List resources
            list_response = tester.send_jsonrpc_request("resources/list", id=1)
            if list_response:
                print(f"Resources list response: {json.dumps(list_response, indent=2)}")
                
                # Get first resource URI
                if "result" in list_response and "resources" in list_response["result"]:
                    resources = list_response["result"]["resources"]
                    if resources:
                        first_uri = resources[0]["uri"]
                        print(f"Trying to read resource: {first_uri}")
                        
                        # Read resource
                        read_response = tester.send_jsonrpc_request("resources/read", {"uri": first_uri}, id=2)
                        if read_response:
                            print(f"Resource read response: {json.dumps(read_response, indent=2)}")
                        else:
                            print("Failed to read resource")
                    else:
                        print("No resources found")
                else:
                    print("No resources in response")
            else:
                print("Failed to list resources")
        else:
            print("Failed to initialize server")
            
    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()