test_print_client.py•5.61 kB
#!/usr/bin/env python3
"""
Test client for the Simple Print MCP Server
"""
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_print_server():
"""Test the print server functionality."""
print("🧪 Testing Simple Print MCP Server")
print("=" * 40)
try:
# Connect to the print server
print("📡 Connecting to localhost:8001...")
reader, writer = await asyncio.open_connection("localhost", 8001)
print("✅ Connected successfully!")
# Send initialization request
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "Print 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
data = await reader.read(1024)
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")
else:
print("⚠️ No response from tools/list")
# Test print message
await asyncio.sleep(0.1)
print_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "print_message",
"arguments": {
"message": "Hello from the test client!",
"level": "info"
}
}
}
print("\n📤 Testing print_message tool...")
writer.write(json.dumps(print_request).encode())
await writer.drain()
# Read response
data = await reader.read(1024)
if data:
response = json.loads(data.decode())
print("📥 Received print response:")
print(json.dumps(response, indent=2))
else:
print("⚠️ No response from print_message")
# Test list messages
await asyncio.sleep(0.1)
list_request = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "list_messages",
"arguments": {
"limit": 5
}
}
}
print("\n📤 Testing list_messages tool...")
writer.write(json.dumps(list_request).encode())
await writer.drain()
# Read response
data = await reader.read(1024)
if data:
response = json.loads(data.decode())
print("📥 Received list_messages response:")
print(json.dumps(response, indent=2))
else:
print("⚠️ No response from list_messages")
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 print server is running on localhost:8001")
print("💡 Start the print server with: py -3 simple_print_server.py")
except Exception as e:
print(f"❌ Error during test: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(test_print_server())