test_mcp_stdio.py•13.5 kB
#!/usr/bin/env python3
"""
Comprehensive MCP stdio server test
Tests all MCP capabilities: Resources, Tools, and Prompts via stdio transport
"""
import asyncio
import json
import subprocess
import sys
from pathlib import Path
class MCPStdioTester:
"""Test MCP server stdio transport"""
def __init__(self):
self.server_process = None
self.test_results = []
async def start_mcp_server(self):
"""Start the MCP server in stdio mode"""
print("Starting MCP stdio server...")
# Start the hybrid server in stdio mode
self.server_process = subprocess.Popen(
[sys.executable, "mcp_server_hybrid.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=Path(__file__).parent
)
# Give it a moment to start
await asyncio.sleep(1)
print("MCP stdio server started")
def send_request(self, request: dict):
"""Send a JSON-RPC request to the MCP server"""
if not self.server_process:
raise RuntimeError("MCP server not started")
# Send the request
request_line = json.dumps(request) + '\n'
self.server_process.stdin.write(request_line)
self.server_process.stdin.flush()
# Read the response
response_line = self.server_process.stdout.readline()
if not response_line:
return None
return json.loads(response_line.strip())
def test_initialize(self):
"""Test MCP server initialization"""
print("\n[INIT] Testing MCP initialization...")
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"resources": {"subscribe": False},
"tools": {},
"prompts": {}
},
"clientInfo": {
"name": "mcp-stdio-test",
"version": "1.0.0"
}
}
}
try:
response = self.send_request(request)
if response and "result" in response:
capabilities = response["result"]["capabilities"]
server_info = response["result"]["serverInfo"]
print(f"[OK] Server initialized: {server_info['name']} v{server_info['version']}")
print(f" Capabilities: {list(capabilities.keys())}")
# Check for all expected capabilities
expected_caps = ["resources", "tools", "prompts"]
for cap in expected_caps:
if cap in capabilities:
print(f" [OK] {cap.title()} capability available")
else:
print(f" [FAIL] {cap.title()} capability missing")
# Send initialized notification as required by MCP protocol
initialized_request = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
self.server_process.stdin.write(json.dumps(initialized_request) + '\n')
self.server_process.stdin.flush()
self.test_results.append(("Initialize", True, "Server initialized successfully"))
return True
else:
print(f"[FAIL] Initialization failed: {response}")
self.test_results.append(("Initialize", False, f"Failed: {response}"))
return False
except Exception as e:
print(f"[ERROR] Initialization error: {e}")
self.test_results.append(("Initialize", False, f"Error: {e}"))
return False
def test_resources(self):
"""Test MCP resources functionality"""
print("\n[RESOURCES] Testing MCP resources...")
# Test list resources
request = {
"jsonrpc": "2.0",
"id": 2,
"method": "resources/list",
"params": {}
}
try:
response = self.send_request(request)
if response and "result" in response:
resources = response["result"]["resources"]
print(f"[OK] Listed {len(resources)} resources")
# Test reading a specific resource
if resources:
# Test database://tables resource
read_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "resources/read",
"params": {
"uri": "database://tables"
}
}
read_response = self.send_request(read_request)
if read_response and "result" in read_response:
content = read_response["result"]["contents"][0]["text"]
data = json.loads(content)
print(f" [OK] Read resource: found {data.get('count', 0)} tables")
self.test_results.append(("Resources", True, f"Listed {len(resources)} resources, read sample successfully"))
else:
print(f" [FAIL] Failed to read resource: {read_response}")
self.test_results.append(("Resources", False, "Failed to read sample resource"))
else:
print(" [WARN] No resources available to test")
self.test_results.append(("Resources", True, "No resources to test"))
else:
print(f"[FAIL] Failed to list resources: {response}")
self.test_results.append(("Resources", False, f"Failed to list: {response}"))
except Exception as e:
print(f"[ERROR] Resources test error: {e}")
self.test_results.append(("Resources", False, f"Error: {e}"))
def test_tools(self):
"""Test MCP tools functionality"""
print("\n[TOOLS] Testing MCP tools...")
# Test list tools
request = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/list",
"params": {}
}
try:
response = self.send_request(request)
if response and "result" in response:
tools = response["result"]["tools"]
print(f"[OK] Listed {len(tools)} tools")
# Test calling a tool (validate SQL syntax)
if tools:
# Find a safe tool to test
test_tool = None
for tool in tools:
if tool["name"] == "validate_sql_syntax":
test_tool = tool
break
if test_tool:
call_request = {
"jsonrpc": "2.0",
"id": 5,
"method": "tools/call",
"params": {
"name": "validate_sql_syntax",
"arguments": {
"sql_query": "SELECT 1"
}
}
}
call_response = self.send_request(call_request)
if call_response and "result" in call_response:
print(f" [OK] Called tool successfully")
self.test_results.append(("Tools", True, f"Listed {len(tools)} tools, called sample successfully"))
else:
print(f" [FAIL] Failed to call tool: {call_response}")
self.test_results.append(("Tools", False, "Failed to call sample tool"))
else:
print(" [WARN] No safe tools available to test")
self.test_results.append(("Tools", True, f"Listed {len(tools)} tools (no safe test tool)"))
else:
print(" [WARN] No tools available to test")
self.test_results.append(("Tools", True, "No tools to test"))
else:
print(f"[FAIL] Failed to list tools: {response}")
self.test_results.append(("Tools", False, f"Failed to list: {response}"))
except Exception as e:
print(f"[ERROR] Tools test error: {e}")
self.test_results.append(("Tools", False, f"Error: {e}"))
def test_prompts(self):
"""Test MCP prompts functionality"""
print("\n[PROMPTS] Testing MCP prompts...")
# Test list prompts
request = {
"jsonrpc": "2.0",
"id": 6,
"method": "prompts/list",
"params": {}
}
try:
response = self.send_request(request)
if response and "result" in response:
prompts = response["result"]["prompts"]
print(f"[OK] Listed {len(prompts)} prompts")
# Test getting a specific prompt
if prompts:
# Find a simple prompt to test
test_prompt = None
for prompt in prompts:
if prompt["name"] == "check_database_health":
test_prompt = prompt
break
if test_prompt:
get_request = {
"jsonrpc": "2.0",
"id": 7,
"method": "prompts/get",
"params": {
"name": "check_database_health",
"arguments": {
"database": "db3"
}
}
}
get_response = self.send_request(get_request)
if get_response and "result" in get_response:
messages = get_response["result"]["messages"]
print(f" [OK] Got prompt with {len(messages)} messages")
self.test_results.append(("Prompts", True, f"Listed {len(prompts)} prompts, got sample successfully"))
else:
print(f" [FAIL] Failed to get prompt: {get_response}")
self.test_results.append(("Prompts", False, "Failed to get sample prompt"))
else:
print(" [WARN] No suitable prompts available to test")
self.test_results.append(("Prompts", True, f"Listed {len(prompts)} prompts (no test prompt)"))
else:
print(" [WARN] No prompts available to test")
self.test_results.append(("Prompts", True, "No prompts to test"))
else:
print(f"[FAIL] Failed to list prompts: {response}")
self.test_results.append(("Prompts", False, f"Failed to list: {response}"))
except Exception as e:
print(f"[ERROR] Prompts test error: {e}")
self.test_results.append(("Prompts", False, f"Error: {e}"))
def cleanup(self):
"""Clean up the MCP server process"""
if self.server_process:
print("\n[CLEANUP] Cleaning up MCP server...")
self.server_process.terminate()
try:
self.server_process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.server_process.kill()
self.server_process.wait()
print("[OK] MCP server stopped")
def print_summary(self):
"""Print test results summary"""
print("\n" + "="*60)
print("[SUMMARY] MCP STDIO TEST SUMMARY")
print("="*60)
passed = sum(1 for _, success, _ in self.test_results if success)
total = len(self.test_results)
print(f"Tests passed: {passed}/{total}")
print()
for test_name, success, message in self.test_results:
status = "[PASS]" if success else "[FAIL]"
print(f"{status:10} {test_name:<15} {message}")
print("\n" + "="*60)
if passed == total:
print("[SUCCESS] ALL TESTS PASSED! MCP stdio server is fully functional.")
else:
print(f"[WARNING] {total - passed} test(s) failed. Check the details above.")
print("="*60)
async def main():
"""Main test runner"""
print("MCP STDIO COMPREHENSIVE TEST")
print("Testing Resources, Tools, and Prompts via stdio transport")
print("="*60)
tester = MCPStdioTester()
try:
# Start the MCP server
await tester.start_mcp_server()
# Run all tests
tester.test_initialize()
tester.test_resources()
tester.test_tools()
tester.test_prompts()
except KeyboardInterrupt:
print("\n[INTERRUPT] Test interrupted by user")
except Exception as e:
print(f"\n[ERROR] Test runner error: {e}")
finally:
# Always cleanup
tester.cleanup()
tester.print_summary()
if __name__ == "__main__":
asyncio.run(main())