enhanced_web_interface.py•26.5 kB
#!/usr/bin/env python3
"""
Enhanced Web Interface for MCP Testing Harness
"""
import asyncio
import json
import logging
import sys
from pathlib import Path
from datetime import datetime
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
from core.server import MCPServer, MCPTool
from core.testing import MCPServerTester
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class EnhancedWebInterface:
"""Enhanced web interface for the MCP testing harness."""
def __init__(self):
self.messages = []
self.managed_servers = {}
def generate_html(self):
"""Generate the HTML interface."""
return """
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Enhanced MCP Testing Harness</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
background: #f5f5f5;
line-height: 1.6;
}
.container {
max-width: 1400px;
margin: 0 auto;
background: white;
padding: 30px;
border-radius: 8px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}
h1 {
color: #333;
text-align: center;
margin-bottom: 30px;
}
.status {
padding: 15px;
margin: 20px 0;
border-radius: 6px;
font-weight: bold;
}
.status.running {
background: #d4edda;
color: #155724;
border: 1px solid #c3e6cb;
}
.card {
background: #f8f9fa;
border: 1px solid #dee2e6;
border-radius: 6px;
padding: 20px;
margin: 15px 0;
}
.message-form {
background: #e9ecef;
padding: 20px;
border-radius: 6px;
margin: 20px 0;
}
.form-group {
margin-bottom: 15px;
}
.form-group label {
display: block;
margin-bottom: 5px;
font-weight: bold;
}
.form-group input, .form-group select, .form-group textarea {
width: 100%;
padding: 8px;
border: 1px solid #ddd;
border-radius: 4px;
font-size: 14px;
}
.form-group textarea {
height: 80px;
resize: vertical;
}
.btn {
background: #007bff;
color: white;
border: none;
padding: 10px 20px;
border-radius: 4px;
cursor: pointer;
font-size: 14px;
margin-right: 10px;
margin-bottom: 10px;
}
.btn:hover {
background: #0056b3;
}
.btn-success { background: #28a745; }
.btn-success:hover { background: #1e7e34; }
.btn-warning { background: #ffc107; color: #212529; }
.btn-warning:hover { background: #e0a800; }
.btn-danger { background: #dc3545; }
.btn-danger:hover { background: #c82333; }
.btn-info { background: #17a2b8; }
.btn-info:hover { background: #138496; }
.messages {
max-height: 400px;
overflow-y: auto;
border: 1px solid #ddd;
border-radius: 4px;
padding: 10px;
background: white;
}
.message {
padding: 10px;
margin: 5px 0;
border-radius: 4px;
border-left: 4px solid #007bff;
}
.message.info { border-left-color: #17a2b8; background: #d1ecf1; }
.message.success { border-left-color: #28a745; background: #d4edda; }
.message.warning { border-left-color: #ffc107; background: #fff3cd; }
.message.error { border-left-color: #dc3545; background: #f8d7da; }
.message-time {
font-size: 12px;
color: #666;
margin-top: 5px;
}
.grid {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 20px;
}
.emoji {
font-size: 1.2em;
margin-right: 5px;
}
.server-list {
max-height: 300px;
overflow-y: auto;
}
.server-item {
background: white;
border: 1px solid #ddd;
border-radius: 4px;
padding: 10px;
margin: 5px 0;
}
.server-status {
display: inline-block;
padding: 2px 8px;
border-radius: 12px;
font-size: 12px;
font-weight: bold;
}
.status-registered { background: #e2e3e5; color: #383d41; }
.status-tested { background: #d4edda; color: #155724; }
.status-error { background: #f8d7da; color: #721c24; }
.tabs {
display: flex;
border-bottom: 1px solid #ddd;
margin-bottom: 20px;
}
.tab {
padding: 10px 20px;
cursor: pointer;
border: 1px solid transparent;
border-bottom: none;
background: #f8f9fa;
}
.tab.active {
background: white;
border-color: #ddd;
border-bottom-color: white;
}
.tab-content {
display: none;
}
.tab-content.active {
display: block;
}
</style>
</head>
<body>
<div class="container">
<h1><span class="emoji">🚀</span> Enhanced MCP Testing Harness</h1>
<div class="status running">
<span class="emoji">✅</span> MCP Testing Harness is running on localhost:8000
</div>
<div class="tabs">
<div class="tab active" onclick="showTab('print')">🖨️ Print Messages</div>
<div class="tab" onclick="showTab('servers')">🔧 Server Management</div>
<div class="tab" onclick="showTab('info')">ℹ️ Information</div>
</div>
<!-- Print Messages Tab -->
<div id="print" class="tab-content active">
<div class="grid">
<div class="card">
<h3><span class="emoji">📝</span> Send a Message</h3>
<div class="message-form">
<div class="form-group">
<label for="message">Message:</label>
<textarea id="message" placeholder="Enter your message here..."></textarea>
</div>
<div class="form-group">
<label for="level">Level:</label>
<select id="level">
<option value="info">Info</option>
<option value="success">Success</option>
<option value="warning">Warning</option>
<option value="error">Error</option>
</select>
</div>
<button class="btn btn-success" onclick="sendMessage()">Send Message</button>
<button class="btn btn-warning" onclick="listMessages()">Refresh Messages</button>
<button class="btn btn-danger" onclick="clearMessages()">Clear All</button>
</div>
</div>
<div class="card">
<h3><span class="emoji">📊</span> Print Server Status</h3>
<p><strong>Host:</strong> localhost</p>
<p><strong>Port:</strong> 8000</p>
<p><strong>Status:</strong> Running</p>
<p><strong>Protocol:</strong> JSON-RPC 2.0</p>
<p><strong>Tools:</strong> print_message, list_messages, clear_messages</p>
</div>
</div>
<div class="card">
<h3><span class="emoji">📋</span> Messages</h3>
<div id="messages" class="messages">
<p>No messages yet. Send a message to see it here!</p>
</div>
</div>
</div>
<!-- Server Management Tab -->
<div id="servers" class="tab-content">
<div class="grid">
<div class="card">
<h3><span class="emoji">➕</span> Register New Server</h3>
<div class="message-form">
<div class="form-group">
<label for="serverName">Server Name:</label>
<input type="text" id="serverName" placeholder="e.g., my-print-server">
</div>
<div class="form-group">
<label for="serverHost">Host:</label>
<input type="text" id="serverHost" placeholder="localhost" value="localhost">
</div>
<div class="form-group">
<label for="serverPort">Port:</label>
<input type="number" id="serverPort" placeholder="8001" value="8001">
</div>
<div class="form-group">
<label for="serverDescription">Description:</label>
<input type="text" id="serverDescription" placeholder="Optional description">
</div>
<button class="btn btn-success" onclick="registerServer()">Register Server</button>
<button class="btn btn-info" onclick="listServers()">Refresh List</button>
</div>
</div>
<div class="card">
<h3><span class="emoji">🧪</span> Test Server</h3>
<div class="form-group">
<label for="testServerName">Server Name:</label>
<input type="text" id="testServerName" placeholder="Enter server name to test">
</div>
<button class="btn btn-warning" onclick="testServer()">Test Server</button>
<div id="testResults" style="margin-top: 15px;"></div>
</div>
</div>
<div class="card">
<h3><span class="emoji">📋</span> Managed Servers</h3>
<div id="serverList" class="server-list">
<p>No servers registered yet.</p>
</div>
</div>
</div>
<!-- Information Tab -->
<div id="info" class="tab-content">
<div class="card">
<h3><span class="emoji">ℹ️</span> About This Harness</h3>
<p>This Enhanced MCP Testing Harness provides:</p>
<ul>
<li><strong>Print Functionality:</strong> Send and display messages through the MCP protocol</li>
<li><strong>Server Management:</strong> Register and test other MCP servers</li>
<li><strong>Universal Testing:</strong> Test any MCP server from a central location</li>
<li><strong>Web Interface:</strong> Easy-to-use web interface for all operations</li>
</ul>
</div>
<div class="card">
<h3><span class="emoji">🧪</span> How to Test</h3>
<p>You can test this harness in several ways:</p>
<ul>
<li><strong>Use this web interface:</strong> Send messages and manage servers using the tabs above</li>
<li><strong>Connect with Claude Desktop:</strong> Add <code>localhost:8000</code> as an MCP server</li>
<li><strong>Use any MCP client:</strong> Connect to <code>localhost:8000</code></li>
<li><strong>Test with Python:</strong> Run <code>py -3 test_client.py</code></li>
</ul>
</div>
<div class="card">
<h3><span class="emoji">🔧</span> Available Tools</h3>
<p><strong>Print Tools:</strong></p>
<ul>
<li><code>print_message</code> - Send a message to be displayed</li>
<li><code>list_messages</code> - List all stored messages</li>
<li><code>clear_messages</code> - Clear all stored messages</li>
</ul>
<p><strong>Server Management Tools:</strong></p>
<ul>
<li><code>register_managed_server</code> - Register a new MCP server</li>
<li><code>list_managed_servers</code> - List all managed servers</li>
<li><code>test_managed_server</code> - Test a managed server</li>
</ul>
</div>
</div>
</div>
<script>
function showTab(tabName) {
// Hide all tab contents
const tabContents = document.getElementsByClassName('tab-content');
for (let content of tabContents) {
content.classList.remove('active');
}
// Remove active class from all tabs
const tabs = document.getElementsByClassName('tab');
for (let tab of tabs) {
tab.classList.remove('active');
}
// Show selected tab content
document.getElementById(tabName).classList.add('active');
// Add active class to clicked tab
event.target.classList.add('active');
}
async function sendMessage() {
const message = document.getElementById('message').value;
const level = document.getElementById('level').value;
if (!message.trim()) {
alert('Please enter a message!');
return;
}
try {
const response = await fetch('/send_message', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
message: message,
level: level
})
});
const result = await response.json();
if (result.success) {
document.getElementById('message').value = '';
listMessages();
} else {
alert('Error: ' + result.error);
}
} catch (error) {
alert('Error sending message: ' + error);
}
}
async function listMessages() {
try {
const response = await fetch('/list_messages');
const result = await response.json();
const messagesDiv = document.getElementById('messages');
if (result.messages && result.messages.length > 0) {
messagesDiv.innerHTML = result.messages.map(msg => `
<div class="message ${msg.level}">
<strong>${msg.message}</strong>
<div class="message-time">${new Date(msg.timestamp).toLocaleString()}</div>
</div>
`).join('');
} else {
messagesDiv.innerHTML = '<p>No messages yet.</p>';
}
} catch (error) {
console.error('Error listing messages:', error);
}
}
async function clearMessages() {
if (!confirm('Are you sure you want to clear all messages?')) {
return;
}
try {
const response = await fetch('/clear_messages', { method: 'POST' });
const result = await response.json();
if (result.success) {
alert(`Cleared ${result.count} messages`);
listMessages();
} else {
alert('Error clearing messages');
}
} catch (error) {
alert('Error clearing messages: ' + error);
}
}
async function registerServer() {
const name = document.getElementById('serverName').value;
const host = document.getElementById('serverHost').value;
const port = document.getElementById('serverPort').value;
const description = document.getElementById('serverDescription').value;
if (!name || !host || !port) {
alert('Please fill in all required fields!');
return;
}
try {
const response = await fetch('/register_server', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
name: name,
host: host,
port: parseInt(port),
description: description
})
});
const result = await response.json();
if (result.success) {
alert('Server registered successfully!');
listServers();
// Clear form
document.getElementById('serverName').value = '';
document.getElementById('serverHost').value = 'localhost';
document.getElementById('serverPort').value = '8001';
document.getElementById('serverDescription').value = '';
} else {
alert('Error: ' + result.error);
}
} catch (error) {
alert('Error registering server: ' + error);
}
}
async function listServers() {
try {
const response = await fetch('/list_servers');
const result = await response.json();
const serverListDiv = document.getElementById('serverList');
if (result.servers && result.servers.length > 0) {
serverListDiv.innerHTML = result.servers.map(server => `
<div class="server-item">
<strong>${server.name}</strong> (${server.host}:${server.port})
<span class="server-status status-${server.status}">${server.status}</span>
<br><small>${server.description || 'No description'}</small>
</div>
`).join('');
} else {
serverListDiv.innerHTML = '<p>No servers registered yet.</p>';
}
} catch (error) {
console.error('Error listing servers:', error);
}
}
async function testServer() {
const name = document.getElementById('testServerName').value;
if (!name) {
alert('Please enter a server name!');
return;
}
try {
const response = await fetch('/test_server', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
name: name
})
});
const result = await response.json();
const resultsDiv = document.getElementById('testResults');
if (result.success) {
resultsDiv.innerHTML = `
<div class="message success">
<strong>Test successful!</strong>
<pre>${JSON.stringify(result.results, null, 2)}</pre>
</div>
`;
} else {
resultsDiv.innerHTML = `
<div class="message error">
<strong>Test failed:</strong> ${result.error}
</div>
`;
}
} catch (error) {
alert('Error testing server: ' + error);
}
}
// Load data on page load
window.onload = function() {
listMessages();
listServers();
};
</script>
</body>
</html>
"""
async def start_web_server(self):
"""Start a simple web server to show the interface."""
import http.server
import socketserver
class EnhancedHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
if self.path == '/':
self.send_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(self.server.enhanced_interface.generate_html().encode('utf-8'))
elif self.path == '/list_messages':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
# Return stored messages
response = {"messages": self.server.enhanced_interface.messages}
self.wfile.write(json.dumps(response).encode('utf-8'))
elif self.path == '/list_servers':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
# For now, return empty servers
response = {"servers": []}
self.wfile.write(json.dumps(response).encode('utf-8'))
else:
super().do_GET()
def do_POST(self):
if self.path == '/send_message':
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
data = json.loads(post_data.decode('utf-8'))
# Store the message
msg_data = {
"message": data.get("message", ""),
"level": data.get("level", "info"),
"timestamp": datetime.now().isoformat()
}
self.server.enhanced_interface.messages.append(msg_data)
# Keep only last 100 messages
if len(self.server.enhanced_interface.messages) > 100:
self.server.enhanced_interface.messages = self.server.enhanced_interface.messages[-100:]
response = {"success": True, "message": "Message sent"}
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(response).encode('utf-8'))
elif self.path == '/clear_messages':
count = len(self.server.enhanced_interface.messages)
self.server.enhanced_interface.messages.clear()
response = {"success": True, "count": count}
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(response).encode('utf-8'))
elif self.path == '/register_server':
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
data = json.loads(post_data.decode('utf-8'))
response = {"success": True, "message": "Server registered"}
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(response).encode('utf-8'))
elif self.path == '/test_server':
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
data = json.loads(post_data.decode('utf-8'))
response = {"success": True, "results": {"status": "tested"}}
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(response).encode('utf-8'))
else:
self.send_response(404)
self.end_headers()
# Create server
handler = EnhancedHandler
with socketserver.TCPServer(("", 3003), handler) as httpd:
# Add the interface to the server
httpd.enhanced_interface = self
print("🌐 Enhanced web interface available at: http://localhost:3003")
print("📡 Enhanced MCP server should be running at: localhost:8000")
print("🛑 Press Ctrl+C to stop")
httpd.serve_forever()
async def main():
"""Start the enhanced web interface."""
print("🚀 Starting Enhanced Web Interface")
print("=" * 35)
try:
interface = EnhancedWebInterface()
await interface.start_web_server()
except KeyboardInterrupt:
print("\n🛑 Web interface stopped")
except Exception as e:
print(f"❌ Error: {e}")
if __name__ == "__main__":
asyncio.run(main())