File size: 6,177 Bytes
2be6245 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | #!/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")
|