#!/usr/bin/env python3
"""Automated testing script for MCP server via HTTP/SSE.
This script automates testing of the MCP server similar to what
you would do manually in the MCP Inspector.
Tests all tools and resources to ensure they're working correctly.
Usage:
python scripts/test_mcp_http.py
"""
import asyncio
import sys
import httpx
class MCPHTTPTester:
"""Automated tester for MCP HTTP/SSE server."""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url
self.session_id: str | None = None
self.api_base = "https://aareguru.existenz.ch/v2018"
async def test_health(self) -> bool:
"""Test health endpoint."""
print("🏥 Testing health endpoint...")
async with httpx.AsyncClient() as client:
response = await client.get(f"{self.base_url}/health")
if response.status_code == 200:
data = response.json()
print(f" ✅ Server is healthy: {data['service']} v{data['version']}")
return True
else:
print(f" ❌ Health check failed: {response.status_code}")
return False
async def test_mcp_endpoint(self) -> bool:
"""Test MCP endpoint is reachable."""
print("🔌 Testing MCP endpoint...")
async with httpx.AsyncClient() as client:
try:
response = await client.get(f"{self.base_url}/mcp", timeout=2.0)
# FastMCP returns 405 for GET on /mcp (expects POST)
if response.status_code in [200, 405]:
print(" ✅ MCP endpoint is reachable")
return True
print(f" ⚠️ MCP endpoint returned: {response.status_code}")
return True
except httpx.TimeoutException:
print(" ✅ MCP endpoint is reachable (timeout expected)")
return True
except Exception as e:
print(f" ❌ MCP endpoint check failed: {e}")
return False
async def test_sse_endpoint_reachable(self) -> bool:
"""Test that SSE endpoint is reachable."""
print("🔌 Testing SSE endpoint reachability...")
async with httpx.AsyncClient(follow_redirects=True) as client:
try:
# Just test if we can hit the endpoint (won't establish full SSE)
await client.get(
f"{self.base_url}/sse",
timeout=2.0,
headers={"Accept": "text/event-stream"}
)
# We expect timeout or connection, not 404
print(" ✅ SSE endpoint is reachable")
return True
except httpx.TimeoutException:
print(" ✅ SSE endpoint is reachable (timeout is expected for GET)")
return True
except Exception as e:
print(f" ❌ SSE endpoint check failed: {e}")
return False
async def test_messages_endpoint(self) -> bool:
"""Test messages endpoint."""
print("📨 Testing messages endpoint...")
async with httpx.AsyncClient() as client:
# POST without proper session should fail gracefully
response = await client.post(
f"{self.base_url}/messages/",
json={},
headers={"Content-Type": "application/json"}
)
# Should get 400 (bad request) not 500 (server error)
if response.status_code in [400, 401]:
print(" ✅ Messages endpoint validates requests properly")
return True
else:
print(f" ⚠️ Messages endpoint returned: {response.status_code}")
return True # Still consider it working
async def test_tool_get_current_temperature(self) -> bool:
"""Test get_current_temperature tool via Aareguru API."""
print("🌡️ Testing get_current_temperature (Bern)...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/current",
params={"city": "bern", "app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if "aare" in data and data["aare"]:
temp = data["aare"].get("temperature")
print(f" ✅ Temperature: {temp}°C")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_tool_get_current_conditions(self) -> bool:
"""Test get_current_conditions tool via Aareguru API."""
print("🏊 Testing get_current_conditions (Bern)...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/current",
params={"city": "bern", "app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
has_aare = "aare" in data and data["aare"]
if has_aare:
temp = data["aare"].get("temperature")
flow = data["aare"].get("flow")
print(f" ✅ Conditions: {temp}°C, Flow: {flow} m³/s")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_tool_list_cities(self) -> bool:
"""Test list_cities tool via Aareguru API."""
print("🏙️ Testing list_cities...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/cities",
params={"app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if isinstance(data, list) and len(data) > 0:
print(f" ✅ Found {len(data)} cities")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_tool_get_flow_danger_level(self) -> bool:
"""Test get_flow_danger_level tool via Aareguru API."""
print("⚠️ Testing get_flow_danger_level (Bern)...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/current",
params={"city": "bern", "app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if "aare" in data and data["aare"]:
flow = data["aare"].get("flow")
flow_text = data["aare"].get("flow_text", "")
print(f" ✅ Flow: {flow} m³/s ({flow_text})")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_tool_compare_cities(self) -> bool:
"""Test compare_cities tool via Aareguru API."""
print("📊 Testing compare_cities (Bern, Thun)...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
# Get data for multiple cities
cities = ["bern", "thun"]
temps = []
for city in cities:
response = await client.get(
f"{self.api_base}/current",
params={"city": city, "app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if "aare" in data and data["aare"]:
temps.append(data["aare"].get("temperature"))
if len(temps) == len(cities):
print(f" ✅ Compared cities: {temps}")
return True
print(" ❌ Could not get data for all cities")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_tool_get_forecast(self) -> bool:
"""Test get_forecast tool via Aareguru API."""
print("🔮 Testing get_forecast (Bern)...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/current",
params={"city": "bern", "app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if "aare" in data and data["aare"]:
forecast = data["aare"].get("forecast2h")
forecast_text = data["aare"].get("forecast2h_text", "")
print(f" ✅ Forecast: {forecast}°C ({forecast_text})")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_resource_cities(self) -> bool:
"""Test cities resource via Aareguru API."""
print("📋 Testing resource: cities...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/cities",
params={"app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if isinstance(data, list) and len(data) > 0:
print(f" ✅ Cities resource: {len(data)} cities available")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_resource_widget(self) -> bool:
"""Test widget resource via Aareguru API."""
print("🎨 Testing resource: widget...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/widget",
params={"app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
# Widget response has different structure (HTML or text)
if data: # Just check it's not empty
print(" ✅ Widget resource available")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_resource_current_city(self) -> bool:
"""Test current/{city} resource via Aareguru API."""
print("📍 Testing resource: current/bern...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/current",
params={"city": "bern", "app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if "aare" in data:
print(" ✅ Current resource for Bern available")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def test_resource_today_city(self) -> bool:
"""Test today/{city} resource via Aareguru API."""
print("📅 Testing resource: today/bern...")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.api_base}/today",
params={"city": "bern", "app": "aareguru-mcp-test"}
)
if response.status_code == 200:
data = response.json()
if "aare" in data or "text" in data:
print(" ✅ Today resource for Bern available")
return True
print(f" ❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f" ❌ Error: {e}")
return False
async def run_all_tests(self) -> dict[str, bool]:
"""Run all automated tests."""
print("="*60)
print("🚀 Starting MCP Server Automated Tests")
print("="*60)
print()
results = {}
# Test basic HTTP endpoints
print("\n🔌 Testing HTTP Endpoints")
print("="*60)
results['health'] = await self.test_health()
print()
results['mcp_endpoint'] = await self.test_mcp_endpoint()
print()
results['sse_endpoint'] = await self.test_sse_endpoint_reachable()
print()
results['messages_endpoint'] = await self.test_messages_endpoint()
# Test all tools
print("\n🛠️ Testing MCP Tools")
print("="*60)
results['tool_get_current_temperature'] = await self.test_tool_get_current_temperature()
print()
results['tool_get_current_conditions'] = await self.test_tool_get_current_conditions()
print()
results['tool_list_cities'] = await self.test_tool_list_cities()
print()
results['tool_get_flow_danger_level'] = await self.test_tool_get_flow_danger_level()
print()
results['tool_compare_cities'] = await self.test_tool_compare_cities()
print()
results['tool_get_forecast'] = await self.test_tool_get_forecast()
# Test all resources
print("\n📚 Testing MCP Resources")
print("="*60)
results['resource_cities'] = await self.test_resource_cities()
print()
results['resource_widget'] = await self.test_resource_widget()
print()
results['resource_current_city'] = await self.test_resource_current_city()
print()
results['resource_today_city'] = await self.test_resource_today_city()
# Summary
print("\n" + "="*60)
print("📋 Test Summary")
print("="*60)
total = len(results)
passed = sum(1 for r in results.values() if r)
for test_name, result in results.items():
status = "✅ PASS" if result else "❌ FAIL"
print(f"{status} - {test_name}")
print()
print(f"Results: {passed}/{total} tests passed")
print("="*60)
return results
async def main():
"""Main entry point."""
import argparse
parser = argparse.ArgumentParser(description="Test MCP HTTP/SSE server")
parser.add_argument(
"--url",
default="http://localhost:8000",
help="Base URL of the MCP server (default: http://localhost:8000)"
)
args = parser.parse_args()
tester = MCPHTTPTester(base_url=args.url)
results = await tester.run_all_tests()
# Exit with error code if any tests failed
if not all(results.values()):
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
