| |
| """ |
| Test script for the Python MCP server |
| """ |
|
|
| import asyncio |
| import json |
| import requests |
| import sys |
| from pathlib import 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() |
| |
| |
| 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}") |
| |
| |
| 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}") |
| |
| |
| 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" |
| |
| |
| 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 |
| |
| |
| 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}") |
| |
| |
| 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) |
| |
| |
| 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]}...") |
| |
| |
| 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") |
| |
| |
| asyncio.run(test_mcp_server_direct()) |
| |
| |
| test_http_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") |
|
|
