valaq/test_mcp_client.py

#!/usr/bin/env python3
"""
Simple test script to verify VAPI resource provider functionality.
This script sends basic MCP requests to test the VAPI resource provider.
"""

import json
import subprocess
import sys
import time

def send_mcp_request(method, params=None):
    """Send an MCP request to the server and return the response."""
    request = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": method
    }
    if params:
        request["params"] = params
    
    # Start the MCP server process
    process = subprocess.Popen(
        ["./builddir/valaq", "--mcp"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    
    try:
        # Send the request
        request_json = json.dumps(request) + "\n"
        stdout, stderr = process.communicate(input=request_json, timeout=5)
        
        if stderr:
            print(f"STDERR: {stderr}")
        
        if stdout:
            try:
                response = json.loads(stdout.strip())
                return response
            except json.JSONDecodeError as e:
                print(f"Failed to parse JSON response: {e}")
                print(f"Raw response: {stdout}")
                return None
        else:
            print("No response from server")
            return None
            
    except subprocess.TimeoutExpired:
        process.kill()
        print("Server timed out")
        return None
    except Exception as e:
        print(f"Error communicating with server: {e}")
        return None

def test_list_resources():
    """Test listing VAPI resources."""
    print("Testing list_resources...")
    response = send_mcp_request("resources/list")
    
    if response and "result" in response:
        resources = response["result"].get("resources", [])
        print(f"Found {len(resources)} resources:")
        for resource in resources:
            print(f"  - {resource['uri']}: {resource.get('description', 'No description')}")
        return True
    else:
        print(f"Failed to list resources: {response}")
        return False

def test_read_resource():
    """Test reading a VAPI resource."""
    print("\nTesting read_resource for test_namespace_methods.vapi...")
    response = send_mcp_request("resources/read", {"uri": "vapi://test_namespace_methods"})
    
    if response and "result" in response:
        contents = response["result"].get("contents", [])
        if contents:
            print(f"Resource content (first 200 chars): {contents[0]['text'][:200]}...")
            return True
        else:
            print("No content in response")
            return False
    else:
        print(f"Failed to read resource: {response}")
        return False

def test_read_symbol_path():
    """Test reading a VAPI symbol path."""
    print("\nTesting read_resource for vapi://test_namespace_methods/Invercargill...")
    response = send_mcp_request("resources/read", {"uri": "vapi://test_namespace_methods/Invercargill"})
    
    if response and "result" in response:
        contents = response["result"].get("contents", [])
        if contents:
            print(f"Symbol path content (first 200 chars): {contents[0]['text'][:200]}...")
            return True
        else:
            print("No content in response")
            return False
    else:
        print(f"Failed to read symbol path: {response}")
        return False

def test_list_templates():
    """Test listing resource templates."""
    print("\nTesting resources/templates/list...")
    response = send_mcp_request("resources/templates/list")
    
    if response and "result" in response:
        templates = response["result"].get("resourceTemplates", [])
        print(f"Found {len(templates)} templates:")
        for template in templates:
            print(f"  - {template['name']}: {template.get('uriTemplate', 'No URI template')}")
            print(f"    Description: {template.get('description', 'No description')}")
        return True
    else:
        print(f"Failed to list templates: {response}")
        return False

def main():
    """Run all tests."""
    print("Testing VAPI Resource Provider MCP Implementation")
    print("=" * 50)
    
    tests = [
        test_list_resources,
        test_read_resource,
        test_read_symbol_path,
        test_list_templates
    ]
    
    passed = 0
    total = len(tests)
    
    for test in tests:
        if test():
            passed += 1
        time.sleep(1)  # Small delay between tests
    
    print(f"\nTest Results: {passed}/{total} tests passed")
    
    if passed == total:
        print("All tests passed! VAPI resource provider is working correctly.")
        return 0
    else:
        print("Some tests failed. Please check the implementation.")
        return 1

if __name__ == "__main__":
    sys.exit(main())