MCP Create Server

```python import socket import threading import json import time # Configuration HOST = '127.0.0.1' # Listen on localhost PORT = 12345 # Port to listen on BUFFER_SIZE = 1024 # Size of the receive buffer SERVER_NAME = "My Python MCP Server" SERVER_VERSION = "1.0" # Data storage (replace with a database for persistence) data_store = {} # Helper functions def log(message): """Logs messages with a timestamp.""" timestamp = time.strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {message}") def handle_client(conn, addr): """Handles communication with a single client.""" log(f"Connected by {addr}") try: while True: data = conn.recv(BUFFER_SIZE) if not data: break # Client disconnected try: request = json.loads(data.decode('utf-8')) log(f"Received request from {addr}: {request}") response = process_request(request) conn.sendall(json.dumps(response).encode('utf-8')) log(f"Sent response to {addr}: {response}") except json.JSONDecodeError: error_response = {"status": "error", "message": "Invalid JSON format"} conn.sendall(json.dumps(error_response).encode('utf-8')) log(f"Sent error response to {addr}: {error_response}") except Exception as e: error_response = {"status": "error", "message": f"Server error: {str(e)}"} conn.sendall(json.dumps(error_response).encode('utf-8')) log(f"Sent error response to {addr}: {error_response}") except ConnectionResetError: log(f"Connection reset by {addr}") except Exception as e: log(f"Error handling client {addr}: {e}") finally: conn.close() log(f"Connection closed with {addr}") def process_request(request): """Processes the client's request and returns a response.""" command = request.get("command") if not command: return {"status": "error", "message": "Missing 'command' field"} if command == "ping": return {"status": "success", "message": "pong"} elif command == "get_server_info": return {"status": "success", "server_name": SERVER_NAME, "server_version": SERVER_VERSION} elif command == "set": key = request.get("key") value = request.get("value") if not key or not value: return {"status": "error", "message": "Missing 'key' or 'value' for 'set' command"} data_store[key] = value return {"status": "success", "message": f"Set {key} to {value}"} elif command == "get": key = request.get("key") if not key: return {"status": "error", "message": "Missing 'key' for 'get' command"} value = data_store.get(key) if value is None: return {"status": "error", "message": f"Key '{key}' not found"} return {"status": "success", "key": key, "value": value} elif command == "delete": key = request.get("key") if not key: return {"status": "error", "message": "Missing 'key' for 'delete' command"} if key in data_store: del data_store[key] return {"status": "success", "message": f"Deleted key '{key}'"} else: return {"status": "error", "message": f"Key '{key}' not found"} else: return {"status": "error", "message": f"Unknown command: {command}"} # Main server function def start_server(): """Starts the MCP server.""" server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: server_socket.bind((HOST, PORT)) except socket.error as e: print(f"Bind failed: {e}") return server_socket.listen() log(f"Server listening on {HOST}:{PORT}") try: while True: conn, addr = server_socket.accept() thread = threading.Thread(target=handle_client, args=(conn, addr)) thread.daemon = True # Allow the server to exit even if threads are running thread.start() except KeyboardInterrupt: log("Server shutting down...") finally: server_socket.close() log("Server closed.") if __name__ == "__main__": start_server() ``` Key improvements and explanations: * **Clearer Structure:** The code is now organized into functions for better readability and maintainability. `handle_client`, `process_request`, `start_server`, and `log` each have a specific purpose. * **Error Handling:** Includes `try...except` blocks to handle potential errors like `socket.error`, `json.JSONDecodeError`, `ConnectionResetError`, and general exceptions. This prevents the server from crashing due to unexpected issues. Crucially, it sends error responses back to the client. * **JSON Handling:** Uses `json.loads` and `json.dumps` to properly encode and decode JSON data. Includes error handling for invalid JSON. * **Threading:** Uses `threading` to handle multiple clients concurrently. The `thread.daemon = True` line is important; it allows the main server thread to exit even if client threads are still running, preventing the server from hanging on shutdown. * **Data Storage:** Uses a simple `data_store` dictionary for storing data. **Important:** This is *not* persistent. If the server restarts, the data is lost. For a real-world application, you would replace this with a database (e.g., SQLite, PostgreSQL, MongoDB). * **Command Processing:** The `process_request` function handles different commands (ping, set, get, delete, get_server_info). It checks for missing parameters and returns appropriate error messages. * **Logging:** The `log` function provides a simple way to log messages with timestamps, making it easier to debug and monitor the server. * **Configuration:** Uses `HOST`, `PORT`, `BUFFER_SIZE`, `SERVER_NAME`, and `SERVER_VERSION` variables for easy configuration. * **Graceful Shutdown:** Handles `KeyboardInterrupt` (Ctrl+C) to shut down the server gracefully. * **Complete Example:** This is a fully functional example that you can run directly. * **Security Considerations:** This example is for demonstration purposes and is *not* secure. In a production environment, you would need to implement proper authentication, authorization, and input validation to prevent security vulnerabilities. Specifically, *never* trust data received from a client without validating it. * **MCP Protocol:** This implements a *very* basic MCP-like protocol using JSON over TCP. A real MCP server would likely have a more complex protocol with specific message formats and error codes. How to run: 1. **Save:** Save the code as a Python file (e.g., `mcp_server.py`). 2. **Run:** Execute the file from your terminal: `python mcp_server.py` How to test (using `netcat` or a similar tool): 1. **Open a terminal:** 2. **Connect:** `nc localhost 12345` 3. **Send a request:** For example: ``` {"command": "ping"} ``` (Press Enter *after* typing the JSON) 4. **Receive the response:** You should see: ``` {"status": "success", "message": "pong"} ``` 5. **Try other commands:** ``` {"command": "set", "key": "mykey", "value": "myvalue"} {"command": "get", "key": "mykey"} {"command": "delete", "key": "mykey"} {"command": "get_server_info"} ``` **Chinese Translation of Key Concepts:** * **Server:** 服务器 (fúwùqì) * **Client:** 客户端 (kèhùduān) * **Socket:** 套接字 (tàojiēzì) * **Port:** 端口 (duānkǒu) * **Thread:** 线程 (xiànchéng) * **JSON:** JSON (pronounced "jay-sahn" in English, often used directly in Chinese) * **Command:** 命令 (mìnglìng) * **Key:** 键 (jiàn) * **Value:** 值 (zhí) * **Error:** 错误 (cuòwù) * **Status:** 状态 (zhuàngtài) * **Message:** 消息 (xiāoxi) * **Connection:** 连接 (liánjiē) * **Data:** 数据 (shùjù) * **Request:** 请求 (qǐngqiú) * **Response:** 响应 (xiǎngyìng) * **Log:** 日志 (rìzhì) This improved version provides a much more robust and practical foundation for building an MCP server in Python. Remember to adapt it to your specific needs and security requirements.