Building an Echo Server

"""
echo_server.py - A JSON-RPC style echo server.
This is a precursor to our MCP server.
"""

import json
from http.server import HTTPServer, BaseHTTPRequestHandler


class EchoHandler(BaseHTTPRequestHandler):
    """Handle JSON-RPC style requests by echoing them back."""

    def log_message(self, format, *args):
        """Suppress default logging."""
        pass

    def read_json(self):
        """Read and parse JSON from request body."""
        content_length = int(self.headers.get("Content-Length", 0))
        if content_length == 0:
            return None
        body = self.rfile.read(content_length)
        return json.loads(body.decode("utf-8"))

    def send_json(self, data, status=200):
        """Send a JSON response."""
        body = json.dumps(data, indent=2).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        """Handle POST requests."""
        # Validate content type
        content_type = self.headers.get("Content-Type", "")
        if "application/json" not in content_type:
            self.send_json(
                {"error": "Content-Type must be application/json"},
                status=415
            )
            return

        # Parse JSON
        try:
            request = self.read_json()
        except json.JSONDecodeError as e:
            self.send_json(
                {"error": f"Invalid JSON: {e}"},
                status=400
            )
            return

        if request is None:
            self.send_json({"error": "Empty body"}, status=400)
            return

        # Validate JSON-RPC structure
        if not isinstance(request, dict):
            self.send_json(
                {"error": "Expected JSON object"},
                status=400
            )
            return

        if request.get("jsonrpc") != "2.0":
            self.send_json(
                {"error": "jsonrpc must be '2.0'"},
                status=400
            )
            return

        if "method" not in request:
            self.send_json(
                {"error": "Missing 'method' field"},
                status=400
            )
            return

        # Echo the request back as the result
        response = {
            "jsonrpc": "2.0",
            "id": request.get("id"),
            "result": {
                "echo": {
                    "method": request.get("method"),
                    "params": request.get("params", {})
                }
            }
        }

        print(f"[Echo] {request.get('method')}")
        self.send_json(response)


def run_server(host="localhost", port=8080):
    """Start the echo server."""
    server = HTTPServer((host, port), EchoHandler)
    print(f"Echo server running on http://{host}:{port}")
    print("Press Ctrl+C to stop")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down...")
        server.shutdown()


if __name__ == "__main__":
    run_server()

python echo_server.py

# Valid request
curl -X POST http://localhost:8080 \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {}
  }'

# Response:
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "echo": {
      "method": "tools/list",
      "params": {}
    }
  }
}