test_calculator.py•5.16 kB
#!/usr/bin/env python3
"""
Test script for the Calculator MCP Server
Tests the server by sending JSON-RPC requests via stdio
"""
import asyncio
import json
import sys
import subprocess
async def test_calculator_server():
"""Test the calculator MCP server with various operations."""
# Start the server process
process = await asyncio.create_subprocess_exec(
sys.executable, "calculator_server.py",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
test_cases = [
{
"method": "tools/list",
"params": {},
"description": "List available tools"
},
{
"method": "tools/call",
"params": {
"name": "add",
"arguments": {"numbers": [1, 2, 3, 4]}
},
"description": "Add: 1 + 2 + 3 + 4"
},
{
"method": "tools/call",
"params": {
"name": "multiply",
"arguments": {"numbers": [5, 6, 2]}
},
"description": "Multiply: 5 * 6 * 2"
},
{
"method": "tools/call",
"params": {
"name": "subtract",
"arguments": {"numbers": [100, 20, 5]}
},
"description": "Subtract: 100 - 20 - 5"
},
{
"method": "tools/call",
"params": {
"name": "divide",
"arguments": {"numbers": [100, 5, 2]}
},
"description": "Divide: 100 / 5 / 2"
},
{
"method": "tools/call",
"params": {
"name": "power",
"arguments": {"base": 2, "exponent": 8}
},
"description": "Power: 2^8"
},
{
"method": "tools/call",
"params": {
"name": "sqrt",
"arguments": {"number": 16}
},
"description": "Square root of 16"
},
{
"method": "tools/call",
"params": {
"name": "evaluate",
"arguments": {"expression": "2 + 2 * 3"}
},
"description": "Evaluate: 2 + 2 * 3"
}
]
try:
# Initialize: send initialize request
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
# Send initialize request
init_json = json.dumps(init_request) + "\n"
process.stdin.write(init_json.encode())
await process.stdin.drain()
# Wait for response
response_line = await asyncio.wait_for(process.stdout.readline(), timeout=5.0)
print("[OK] Initialize response received")
# Send initialized notification
initialized = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
init_notif = json.dumps(initialized) + "\n"
process.stdin.write(init_notif.encode())
await process.stdin.drain()
# Test each tool
request_id = 2
for test_case in test_cases:
print(f"\nTesting: {test_case['description']}")
request = {
"jsonrpc": "2.0",
"id": request_id,
"method": test_case["method"],
"params": test_case["params"]
}
request_json = json.dumps(request) + "\n"
process.stdin.write(request_json.encode())
await process.stdin.drain()
# Read response
response_line = await asyncio.wait_for(process.stdout.readline(), timeout=5.0)
if response_line:
response = json.loads(response_line.decode())
if "result" in response:
print(f" Result: {json.dumps(response['result'], indent=2)}")
elif "error" in response:
print(f" Error: {response['error']}")
else:
print(f" Response: {json.dumps(response, indent=2)}")
else:
print(" No response received")
request_id += 1
print("\n[OK] All tests completed!")
except asyncio.TimeoutError:
print("[ERROR] Timeout waiting for server response")
except Exception as e:
print(f"[ERROR] Error during testing: {e}")
stderr = await process.stderr.read()
if stderr:
print(f"Server stderr: {stderr.decode()}")
finally:
process.stdin.close()
await process.wait()
if __name__ == "__main__":
print("Testing Calculator MCP Server...")
print("=" * 50)
asyncio.run(test_calculator_server())