fixed_test_client.py•5.22 kB
#!/usr/bin/env python3
"""
Fixed test client for the MCP Testing Harness
"""
import asyncio
import json
import logging
import sys
from pathlib import Path
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def test_mcp_client():
"""Test connecting to our MCP server."""
print("🧪 Testing MCP Client Connection")
print("=" * 35)
try:
# Connect to the server
print("📡 Connecting to localhost:8000...")
reader, writer = await asyncio.open_connection("localhost", 8000)
print("✅ Connected successfully!")
# Send initialization request
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "MCP Test Client",
"version": "1.0.0"
}
}
}
print("📤 Sending initialization request...")
writer.write(json.dumps(init_request).encode())
await writer.drain()
# Read response
data = await reader.read(1024)
response = json.loads(data.decode())
print("📥 Received initialization response:")
print(json.dumps(response, indent=2))
if "result" in response:
print("✅ Initialization successful!")
# Wait a moment before sending next request
await asyncio.sleep(0.1)
# Test tools listing
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
print("\n📤 Sending tools/list request...")
writer.write(json.dumps(tools_request).encode())
await writer.drain()
# Read response with timeout
try:
data = await asyncio.wait_for(reader.read(1024), timeout=5.0)
if data:
response = json.loads(data.decode())
print("📥 Received tools response:")
print(json.dumps(response, indent=2))
if "result" in response and "tools" in response["result"]:
tools = response["result"]["tools"]
print(f"✅ Found {len(tools)} tools:")
for tool in tools:
print(f" - {tool.get('name', 'unknown')}: {tool.get('description', 'No description')}")
else:
print("⚠️ No tools found or unexpected response format")
else:
print("⚠️ No data received from tools/list request")
except asyncio.TimeoutError:
print("⚠️ Timeout waiting for tools/list response")
except json.JSONDecodeError as e:
print(f"⚠️ Invalid JSON response: {e}")
# Test a built-in tool
await asyncio.sleep(0.1)
list_servers_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "list_servers",
"arguments": {}
}
}
print("\n📤 Testing list_servers tool...")
writer.write(json.dumps(list_servers_request).encode())
await writer.drain()
# Read response with timeout
try:
data = await asyncio.wait_for(reader.read(1024), timeout=5.0)
if data:
response = json.loads(data.decode())
print("📥 Received list_servers response:")
print(json.dumps(response, indent=2))
else:
print("⚠️ No data received from list_servers request")
except asyncio.TimeoutError:
print("⚠️ Timeout waiting for list_servers response")
except json.JSONDecodeError as e:
print(f"⚠️ Invalid JSON response: {e}")
else:
print("❌ Initialization failed!")
print(f"Error: {response.get('error', 'Unknown error')}")
# Close connection
writer.close()
await writer.wait_closed()
print("\n✅ Test completed!")
except ConnectionRefusedError:
print("❌ Connection refused! Make sure the server is running on localhost:8000")
print("💡 Start the server with: py -3 start_server.py")
except Exception as e:
print(f"❌ Error during test: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(test_mcp_client())