test_mcp.py

#!/usr/bin/env python3
"""
Test script for the Python MCP server
"""

import asyncio
import json
import requests
import sys
from pathlib import Path

# Add the current directory to path
sys.path.insert(0, str(Path(__file__).parent))

from server import PrestaShopMCPServer

async def test_mcp_server_direct():
    """Test the MCP server directly (without HTTP wrapper)"""
    print("Testing MCP Server Direct Interface...")
    print("=" * 50)
    
    server = PrestaShopMCPServer()
    
    # Test get_product_details
    print("\n🧪 Testing get_product_details...")
    try:
        result = await server._get_product_details({"product_id": 123})
        if result:
            print(f"✅ Success: {result[0].text[:200]}...")
        else:
            print("❌ No result returned")
    except Exception as e:
        print(f"❌ Error: {e}")
    
    # Test get_product_features  
    print("\n🧪 Testing get_product_features...")
    try:
        result = await server._get_product_features({"product_id": 123})
        if result:
            print(f"✅ Success: {result[0].text[:200]}...")
        else:
            print("❌ No result returned")
    except Exception as e:
        print(f"❌ Error: {e}")
    
    # Test search_products
    print("\n🧪 Testing search_products...")
    try:
        result = await server._search_products({"query": "laptop", "limit": 5})
        if result:
            print(f"✅ Success: {result[0].text[:200]}...")
        else:
            print("❌ No result returned")
    except Exception as e:
        print(f"❌ Error: {e}")

def test_http_server():
    """Test the HTTP wrapper server"""
    print("\n\nTesting MCP HTTP Server...")
    print("=" * 50)
    
    base_url = "http://localhost:8080"
    
    # Test health check
    print("\n🧪 Testing health check...")
    try:
        response = requests.get(f"{base_url}/health", timeout=5)
        if response.status_code == 200:
            print(f"✅ Health check: {response.json()}")
        else:
            print(f"❌ Health check failed: {response.status_code}")
    except Exception as e:
        print(f"❌ Health check error: {e}")
        print("💡 Make sure to start the HTTP server with: python -m mcp_server.http_server")
        return
    
    # Test tools/list
    print("\n🧪 Testing tools/list...")
    try:
        payload = {
            "jsonrpc": "2.0",
            "method": "tools/list",
            "params": {},
            "id": "test-1"
        }
        response = requests.post(f"{base_url}/", json=payload, timeout=10)
        if response.status_code == 200:
            result = response.json()
            tools = result.get("result", {}).get("tools", [])
            print(f"✅ Found {len(tools)} tools:")
            for tool in tools:
                print(f"   - {tool['name']}: {tool['description']}")
        else:
            print(f"❌ Tools list failed: {response.status_code} - {response.text}")
    except Exception as e:
        print(f"❌ Tools list error: {e}")
    
    # Test tools/call
    print("\n🧪 Testing tools/call - get_product_details...")
    try:
        payload = {
            "jsonrpc": "2.0",
            "method": "tools/call",
            "params": {
                "name": "get_product_details",
                "arguments": {"product_id": 123}
            },
            "id": "test-2"
        }
        response = requests.post(f"{base_url}/", json=payload, timeout=30)
        if response.status_code == 200:
            result = response.json()
            content = result.get("result", {}).get("content", [])
            if content:
                text = content[0].get("text", "")
                print(f"✅ Tool call success: {text[:200]}...")
            else:
                print("❌ No content in response")
        else:
            print(f"❌ Tool call failed: {response.status_code} - {response.text}")
    except Exception as e:
        print(f"❌ Tool call error: {e}")

def test_integration_with_ai_server():
    """Test integration with the main AI server"""
    print("\n\nTesting Integration with AI Server...")
    print("=" * 50)
    
    # This simulates how the main AI server calls the MCP server
    ai_server_url = "http://localhost:8000"
    
    print("\n🧪 Testing AI server product query...")
    try:
        payload = {
            "inputs": "What are the features of product 123?",
            "parameters": {"max_new_tokens": 150}
        }
        response = requests.post(ai_server_url, json=payload, timeout=30)
        if response.status_code == 200:
            result = response.json()
            if isinstance(result, list) and len(result) > 0:
                generated_text = result[0].get("generated_text", "")
                print(f"✅ AI server response: {generated_text[:200]}...")
                
                # Check if it contains tool call
                if "tool" in generated_text and "product_id" in generated_text:
                    print("✅ AI correctly detected product query and returned tool call")
                else:
                    print("ℹ️ AI returned direct response (no tool call detected)")
            else:
                print("❌ Unexpected response format")
        else:
            print(f"❌ AI server failed: {response.status_code} - {response.text}")
    except Exception as e:
        print(f"❌ AI server error: {e}")
        print("💡 Make sure to start the AI server with: python app_api.py")

if __name__ == "__main__":
    print("🚀 PrestaShop MCP Server Test Suite")
    print("Testing the refactored Python MCP server")
    
    # Test 1: Direct MCP server
    asyncio.run(test_mcp_server_direct())
    
    # Test 2: HTTP wrapper (requires server to be running)
    test_http_server()
    
    # Test 3: Integration with AI server
    test_integration_with_ai_server()
    
    print("\n" + "=" * 50)
    print("✅ Test suite completed!")
    print("\n📋 To run the servers:")
    print("1. MCP HTTP Server: python -m mcp_server.http_server")
    print("2. AI Server: python app_api.py")
    print("3. Then run this test again to verify integration")