test_mcp_protocol.py•5.34 kB
#!/usr/bin/env python3
"""
Test MCP protocol communication with the Wikipedia server.
"""
import asyncio
import json
import subprocess
import sys
async def test_mcp_server():
"""Test the MCP server with proper protocol initialization."""
print("🧪 Testing MCP Protocol Communication")
print("=" * 50)
# Start the server process
process = await asyncio.create_subprocess_exec(
sys.executable, "-m", "src.wikipedia_mcp_server",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd="/Users/ishita/wiki-mcp2"
)
try:
# Send initialization request
init_request = {
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"roots": {
"listChanged": True
},
"sampling": {}
},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
},
"id": 1
}
print("📤 Sending initialization request...")
request_str = json.dumps(init_request) + "\n"
process.stdin.write(request_str.encode())
await process.stdin.drain()
# Read response
response_line = await process.stdout.readline()
if response_line:
try:
response = json.loads(response_line.decode().strip())
print("📥 Initialization response:")
print(json.dumps(response, indent=2))
if "result" in response:
print("✅ Server initialized successfully!")
# Send initialized notification (required by MCP protocol)
initialized_notification = {
"jsonrpc": "2.0",
"method": "notifications/initialized",
"params": {}
}
print("\n📤 Sending initialized notification...")
request_str = json.dumps(initialized_notification) + "\n"
process.stdin.write(request_str.encode())
await process.stdin.drain()
# Wait a moment for the notification to be processed
await asyncio.sleep(0.1)
# Now test tools/list
tools_request = {
"jsonrpc": "2.0",
"method": "tools/list",
"params": {},
"id": 2
}
print("📤 Requesting tools list...")
request_str = json.dumps(tools_request) + "\n"
process.stdin.write(request_str.encode())
await process.stdin.drain()
# Read tools response
tools_response_line = await process.stdout.readline()
if tools_response_line:
tools_response = json.loads(tools_response_line.decode().strip())
print("📥 Tools list response:")
print(json.dumps(tools_response, indent=2))
if "result" in tools_response and "tools" in tools_response["result"]:
tools = tools_response["result"]["tools"]
print(f"\n🔧 Found {len(tools)} tools:")
for tool in tools:
print(f" - {tool['name']}: {tool['description']}")
print("\n✅ MCP server is working correctly!")
return True
else:
print("❌ No tools found in response")
return False
else:
print("❌ No response to tools/list request")
return False
else:
print("❌ Initialization failed:")
print(json.dumps(response, indent=2))
return False
except json.JSONDecodeError as e:
print(f"❌ Failed to parse response: {e}")
print(f"Raw response: {response_line.decode()}")
return False
else:
print("❌ No response from server")
return False
except Exception as e:
print(f"❌ Error during testing: {e}")
# Print stderr for debugging
stderr_data = await process.stderr.read()
if stderr_data:
print(f"Server stderr: {stderr_data.decode()}")
return False
finally:
# Clean up
process.terminate()
await process.wait()
if __name__ == "__main__":
success = asyncio.run(test_mcp_server())
if not success:
print("\n🔧 Debug suggestion: Check if there are any import errors or missing dependencies")
sys.exit(1)