simple_print_server.py•6.33 kB
#!/usr/bin/env python3
"""
Simple MCP Server that can print messages
"""
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__)
# Global storage for messages
messages = []
class SimplePrintServer(MCPServer):
"""Simple MCP server with print functionality."""
def __init__(self, host="localhost", port=8001, debug=True):
super().__init__(host=host, port=port, debug=debug)
self.messages = []
self._setup_print_tools()
def _setup_print_tools(self):
"""Setup print-related tools."""
# Add print tool
print_tool = MCPTool(
name="print_message",
description="Print a message that will be displayed on the web interface",
input_schema={
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "The message to print"
},
"level": {
"type": "string",
"enum": ["info", "warning", "error", "success"],
"default": "info",
"description": "Message level"
}
},
"required": ["message"]
},
output_schema={
"type": "object",
"properties": {
"success": {"type": "boolean"},
"message": {"type": "string"},
"timestamp": {"type": "string"}
}
},
handler=self._print_message
)
# Add list messages tool
list_tool = MCPTool(
name="list_messages",
description="List all printed messages",
input_schema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"default": 10,
"description": "Maximum number of messages to return"
}
}
},
output_schema={
"type": "object",
"properties": {
"messages": {
"type": "array",
"items": {
"type": "object",
"properties": {
"message": {"type": "string"},
"level": {"type": "string"},
"timestamp": {"type": "string"}
}
}
}
}
},
handler=self._list_messages
)
# Add clear messages tool
clear_tool = MCPTool(
name="clear_messages",
description="Clear all stored messages",
input_schema={},
output_schema={
"type": "object",
"properties": {
"success": {"type": "boolean"},
"count": {"type": "integer"}
}
},
handler=self._clear_messages
)
# Register tools
asyncio.create_task(self.registry.register_tool(print_tool))
asyncio.create_task(self.registry.register_tool(list_tool))
asyncio.create_task(self.registry.register_tool(clear_tool))
async def _print_message(self, arguments):
"""Handle print message tool."""
message = arguments.get("message", "")
level = arguments.get("level", "info")
if not message:
return {
"success": False,
"error": "Message is required"
}
# Add message to storage
msg_data = {
"message": message,
"level": level,
"timestamp": datetime.now().isoformat()
}
self.messages.append(msg_data)
# Keep only last 100 messages
if len(self.messages) > 100:
self.messages = self.messages[-100:]
logger.info(f"PRINT [{level.upper()}]: {message}")
return {
"success": True,
"message": f"Message printed: {message}",
"timestamp": msg_data["timestamp"]
}
async def _list_messages(self, arguments):
"""Handle list messages tool."""
limit = arguments.get("limit", 10)
# Get recent messages
recent_messages = self.messages[-limit:] if self.messages else []
return {
"messages": recent_messages
}
async def _clear_messages(self, arguments):
"""Handle clear messages tool."""
count = len(self.messages)
self.messages.clear()
return {
"success": True,
"count": count
}
async def main():
"""Start the simple print server."""
print("🚀 Starting Simple Print MCP Server")
print("=" * 40)
try:
# Create server
server = SimplePrintServer(host="localhost", port=8001, debug=True)
print(f"📡 Server will start on localhost:8001")
print("🔄 Starting server...")
# Start the server
await server.start()
except KeyboardInterrupt:
print("\n🛑 Server stopped by user")
except Exception as e:
print(f"❌ Error starting server: {e}")
sys.exit(1)
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n🛑 Server stopped")
except Exception as e:
print(f"❌ Fatal error: {e}")
sys.exit(1)