#!/usr/bin/env python3
"""
Complete test of MCP server in Docker (Corrected Version)
"""
import json
import subprocess
import time
import select
def test_docker_mcp():
"""
Test complete MCP communication with Docker
"""
print("=== MCP SERVER TEST IN DOCKER ===\n")
# Start process
cmd = ["python", "mcp_server.py"]
process = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
# Test 1: Server initialization
print("1. Initialization test...")
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
response = send_request(process, init_request)
if response and "result" in response:
print(" OK - Server initialized correctly")
print(f" Server: {response['result'].get('serverInfo', {}).get('name', 'Unknown')}")
else:
print(f" ERROR - Initialization failed: {response}")
process.terminate()
return
# Test 2: Tools notification
print("2. Tools notification test...")
notification_request = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
# Notifications don't require response
send_request(process, notification_request, expect_response=False)
print(f"OK - Notification sent")
# Test 3: List tools
print("\n3. Tools listing test...")
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
response = send_request(process, tools_request)
if response and "result" in response and "tools" in response["result"]:
tools = response["result"]["tools"]
print(f"OK - Found {len(tools)} tools:")
for tool in tools: # First 3
print(f" - {tool['name']}")
else:
print(f"ERROR - Could not list tools: {response}")
# Test 4: Call list_schemas
print("\n4. Schemas listing test...")
list_schemas_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "get_validation_info",
"arguments": {}
}
}
response = send_request(process, list_schemas_request)
if response and "result" in response:
content = response["result"].get("content", [])
if content:
text = content[0].get("text", "")
print(f"OK - Schema list obtained: {text}")
# Search for schema count
if "Total de schemas:" in text:
lines = text.split('\n')
for line in lines:
if "Total de schemas:" in line:
print(f" {line.strip()}")
elif "• " in line:
print(f" Schema: {line.strip()}")
else:
print(" OK - No content in response")
else:
print(f" ERROR - Failed to list schemas: {response}")
# Close process
process.stdin.close()
process.terminate()
print("\n=== TEST COMPLETED ===")
def send_request(process, request, expect_response=True, timeout=5):
"""
Sends a request and reads the response from the process
"""
try:
# Send request
request_str = json.dumps(request) + "\n"
process.stdin.write(request_str)
process.stdin.flush()
if not expect_response:
return None
# Read response
response_lines = []
start_time = time.time()
while time.time() - start_time < timeout:
# Check if data is available
line = process.stdout.readline().strip()
if line:
response_lines.append(line)
try:
# Try parsing as JSON
return json.loads(line)
except json.JSONDecodeError:
continue
# If no valid response was obtained
if response_lines:
print(f" Invalid response: {response_lines}")
else:
print(" Timeout waiting for response")
return None
except Exception as e:
print(f" Communication error: {e}")
return None
if __name__ == "__main__":
test_docker_mcp()
