#!/usr/bin/env python3
"""
Simple API endpoint tester for all MCP services
Tests all endpoints across HTTP bridge (8000) and Smart Search API (8002)
"""
import requests
import json
import time
import sys
def test_endpoint(service_url, endpoint, method='GET', data=None, timeout=10):
"""Test a single endpoint"""
url = f"{service_url}{endpoint}"
start_time = time.time()
try:
if method == 'GET':
response = requests.get(url, timeout=timeout)
elif method == 'POST':
response = requests.post(url, json=data, timeout=timeout)
else:
response = requests.request(method, url, json=data, timeout=timeout)
response_time = time.time() - start_time
try:
response_data = response.json()
except:
response_data = {"text": response.text}
return {
'status_code': response.status_code,
'response_time': response_time,
'data': response_data,
'success': response.status_code == 200
}
except Exception as e:
return {
'status_code': None,
'response_time': time.time() - start_time,
'data': None,
'success': False,
'error': str(e)
}
def main():
"""Test all API endpoints"""
print("MCP API Endpoint Comprehensive Tester")
print("=" * 50)
# Service URLs
services = {
'HTTP Bridge': 'http://localhost:8000',
'Smart Search API': 'http://localhost:8002'
}
# Test data
test_question = {
"question": "What tables are available in the database?",
"database": "db3",
"include_sql": True,
"include_semantic": True,
"include_schema": True
}
test_chat = {
"message": "Hello, what can you help me with?",
"conversation_id": "test-123"
}
# Endpoints to test
endpoints = [
# HTTP Bridge endpoints (based on OpenAPI spec)
('HTTP Bridge', '/health', 'GET'),
('HTTP Bridge', '/api/echo', 'POST', {"text": "test"}),
('HTTP Bridge', '/api/chat', 'POST', {"message": "Hello"}),
('HTTP Bridge', '/api/conversation/chat', 'POST', test_chat),
('HTTP Bridge', '/api/conversation/chat/stream', 'POST', test_chat),
('HTTP Bridge', '/api/test/stream', 'GET'),
('HTTP Bridge', '/api/chat/stream', 'POST', {"message": "Hello"}),
('HTTP Bridge', '/api/memory/store', 'POST', {"conversation_id": "test", "content": "test memory"}),
('HTTP Bridge', '/api/memory/get', 'POST', {"conversation_id": "test"}),
('HTTP Bridge', '/api/capabilities', 'GET'),
('HTTP Bridge', '/api/models', 'GET'),
('HTTP Bridge', '/api/memory/search?query=test&limit=5', 'GET'),
('HTTP Bridge', '/api/answer', 'POST', test_question),
# Smart Search API endpoints
('Smart Search API', '/health', 'GET'),
('Smart Search API', '/api/info', 'GET'),
('Smart Search API', '/api/databases', 'GET'),
('Smart Search API', '/api/models', 'GET'),
('Smart Search API', '/api/smart-search', 'POST', test_question),
]
# Test service availability first
print("Testing Service Availability")
print("-" * 40)
for name, url in services.items():
try:
response = requests.get(f"{url}/health", timeout=5)
if response.status_code == 200:
print(f"[OK] {name}: {url} - ONLINE")
else:
print(f"[WARN] {name}: {url} - HTTP {response.status_code}")
except Exception as e:
print(f"[FAIL] {name}: {url} - OFFLINE ({str(e)})")
print()
print("Testing Individual Endpoints")
print("-" * 40)
results = []
# Test all endpoints
for endpoint_data in endpoints:
service_name = endpoint_data[0]
endpoint = endpoint_data[1]
method = endpoint_data[2]
data = endpoint_data[3] if len(endpoint_data) > 3 else None
service_url = services[service_name]
result = test_endpoint(service_url, endpoint, method, data)
results.append((service_name, endpoint, method, result))
# Print result
status = "[PASS]" if result['success'] else "[FAIL]"
print(f"{status} {service_name}: {method} {endpoint}")
if result['success']:
print(f" Time: {result['response_time']:.2f}s | HTTP {result['status_code']}")
# Show interesting data
if result['data']:
if endpoint == '/api/models' and 'models' in result['data']:
models = result['data']['models']
if isinstance(models, list) and len(models) > 0:
model_preview = models[:3] if len(models) > 3 else models
suffix = '...' if len(models) > 3 else ''
print(f" Found {len(models)} models: {model_preview}{suffix}")
elif endpoint == '/api/databases' and 'databases' in result['data']:
dbs = result['data']['databases']
print(f" Available databases: {dbs}")
elif endpoint == '/api/smart-search' and 'strategy' in result['data']:
strategy = result['data']['strategy']
time_taken = result['data'].get('processing_time', 0)
print(f" Strategy: {strategy} | Processing: {time_taken:.2f}s")
else:
if 'error' in result:
print(f" Error: {result['error']}")
elif result['status_code']:
print(f" HTTP {result['status_code']}")
print()
# Print summary
print("Test Summary")
print("=" * 50)
total_tests = len(results)
passed = sum(1 for _, _, _, r in results if r['success'])
failed = total_tests - passed
print(f"Total Endpoints Tested: {total_tests}")
print(f"[OK] Passed: {passed}")
print(f"[FAIL] Failed: {failed}")
print(f"Success Rate: {(passed/total_tests*100):.1f}%")
print()
# Show failed tests
if failed > 0:
print("Failed Tests:")
for service, endpoint, method, result in results:
if not result['success']:
error_msg = result.get('error', f"HTTP {result.get('status_code', 'Unknown')}")
print(f" {method} {endpoint}: {error_msg}")
print()
# Performance summary
response_times = [r['response_time'] for _, _, _, r in results if r['response_time'] > 0]
if response_times:
avg_time = sum(response_times) / len(response_times)
max_time = max(response_times)
print(f"Performance:")
print(f" Average Response Time: {avg_time:.2f}s")
print(f" Slowest Response: {max_time:.2f}s")
print()
# Overall status
if failed == 0:
print("All endpoints are working perfectly!")
else:
print("Some endpoints have issues that need attention")
if __name__ == "__main__":
main()
