Test MCP wire protocol compliance

Author: staradigm

config/includes.chroot/usr/local/bin/neuros-mcp

@@ -21,47 +21,105 @@ from http.server import HTTPServer, BaseHTTPRequestHandler
 
 DEFAULT_PORT = 11436
 OLLAMA_URL = "http://localhost:11434/api/generate"
+SUPPORTED_PROTOCOL_VERSION = "2024-11-05"
 
 
 class MCPServer(BaseHTTPRequestHandler):
     """MCP protocol server for NeurOS."""
 
+    def _send_error(self, code, message, req_id=None, data=None, status=200):
+        error = {"code": code, "message": message}
+        if data is not None:
+            error["data"] = data
+        self.send_json({"jsonrpc": "2.0", "error": error, "id": req_id}, status=status)
+
+    def _send_no_content(self):
+        self.send_response(202)
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
     def do_POST(self):
-        content_length = int(self.headers.get('Content-Length', 0))
+        content_type = self.headers.get("Content-Type", "")
+        accept = {value.strip().split(";", 1)[0].lower()
+                  for value in self.headers.get("Accept", "").split(",")}
+        media_type = content_type.split(";", 1)[0].strip().lower()
+        if media_type != "application/json":
+            self._send_error(-32600, "Content-Type must be application/json", status=400)
+            return
+        if not {"application/json", "text/event-stream"}.issubset(accept):
+            self._send_error(-32600,
+                             "Accept must include application/json and text/event-stream",
+                             status=400)
+            return
+
+        try:
+            content_length = int(self.headers.get("Content-Length", ""))
+        except ValueError:
+            self._send_error(-32600, "Content-Length must be a valid integer", status=400)
+            return
+        if content_length < 0:
+            self._send_error(-32600, "Content-Length must not be negative", status=400)
+            return
         body = self.rfile.read(content_length)
 
         try:
             request = json.loads(body)
         except json.JSONDecodeError:
-            self.send_error(400, "Invalid JSON")
+            self._send_error(-32700, "Parse error", status=400)
+            return
+
+        if not isinstance(request, dict):
+            self._send_error(-32600, "Invalid Request", status=400)
             return
 
-        method = request.get("method", "")
-        params = request.get("params", {})
         req_id = request.get("id")
+        if request.get("jsonrpc") != "2.0" or not isinstance(request.get("method"), str):
+            self._send_error(-32600, "Invalid Request", status=400)
+            return
+        if "id" in request and (req_id is None or not isinstance(req_id, (str, int))
+                                 or isinstance(req_id, bool)):
+            self._send_error(-32600, "Invalid Request", status=400)
+            return
+        params = request.get("params", {})
+        if not isinstance(params, dict):
+            self._send_error(-32602, "Invalid params", req_id=req_id)
+            return
+
+        method = request["method"]
+        is_notification = "id" not in request
 
         if method == "initialize":
-            result = self.handle_initialize(params)
+            result, error = self.handle_initialize(params)
         elif method == "tools/list":
-            result = self.handle_list_tools()
+            result, error = self.handle_list_tools(), None
         elif method == "tools/call":
-            result = self.handle_call_tool(params)
+            result, error = self.handle_call_tool(params)
         elif method == "resources/list":
-            result = self.handle_list_resources()
+            result, error = self.handle_list_resources(), None
         elif method == "resources/read":
-            result = self.handle_read_resource(params)
+            result, error = self.handle_read_resource(params)
         else:
-            self.send_json({"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": req_id})
+            error = {"code": -32601, "message": "Method not found"}
+            result = None
+
+        if is_notification:
+            self._send_no_content()
             return
 
-        self.send_json({"jsonrpc": "2.0", "result": result, "id": req_id})
+        if error is not None:
+            self.send_json({"jsonrpc": "2.0", "error": error, "id": req_id})
+        else:
+            self.send_json({"jsonrpc": "2.0", "result": result, "id": req_id})
 
-    def send_json(self, data):
-        self.send_response(200)
+    def send_json(self, data, status=200):
+        body = json.dumps(data).encode("utf-8")
+        self.send_response(status)
         self.send_header("Content-Type", "application/json")
         self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Content-Length", str(len(body)))
         self.end_headers()
-        self.wfile.write(json.dumps(data).encode('utf-8'))
+        self.wfile.write(body)
 
     def do_GET(self):
         if self.path == "/health":
@@ -81,8 +139,25 @@ class MCPServer(BaseHTTPRequestHandler):
         self.end_headers()
 
     def handle_initialize(self, params):
+        if not isinstance(params.get("protocolVersion"), str):
+            return None, {"code": -32602, "message": "Invalid params"}
+        client_info = params.get("clientInfo")
+        if not isinstance(params.get("capabilities"), dict) or not isinstance(client_info, dict):
+            return None, {"code": -32602, "message": "Invalid params"}
+        if not isinstance(client_info.get("name"), str) or not isinstance(client_info.get("version"), str):
+            return None, {"code": -32602, "message": "Invalid params"}
+        requested_version = params["protocolVersion"]
+        if requested_version != SUPPORTED_PROTOCOL_VERSION:
+            return None, {
+                "code": -32602,
+                "message": "Unsupported protocol version",
+                "data": {
+                    "supported": [SUPPORTED_PROTOCOL_VERSION],
+                    "requested": requested_version,
+                },
+            }
         return {
-            "protocolVersion": "2024-11-05",
+            "protocolVersion": SUPPORTED_PROTOCOL_VERSION,
             "serverInfo": {
                 "name": "NeurOS MCP Server",
                 "version": "1.0.0"
@@ -91,7 +166,7 @@ class MCPServer(BaseHTTPRequestHandler):
                 "tools": {},
                 "resources": {}
             }
-        }
+        }, None
 
     def handle_list_tools(self):
         return {
@@ -163,21 +238,23 @@ class MCPServer(BaseHTTPRequestHandler):
     def handle_call_tool(self, params):
         tool_name = params.get("name", "")
         arguments = params.get("arguments", {})
+        if not isinstance(tool_name, str) or not isinstance(arguments, dict):
+            return None, {"code": -32602, "message": "Invalid params"}
 
         if tool_name == "read_file":
-            return self.tool_read_file(arguments)
+            return self.tool_read_file(arguments), None
         elif tool_name == "list_directory":
-            return self.tool_list_directory(arguments)
+            return self.tool_list_directory(arguments), None
         elif tool_name == "run_command":
-            return self.tool_run_command(arguments)
+            return self.tool_run_command(arguments), None
         elif tool_name == "ask_llm":
-            return self.tool_ask_llm(arguments)
+            return self.tool_ask_llm(arguments), None
         elif tool_name == "get_system_info":
-            return self.tool_get_system_info(arguments)
+            return self.tool_get_system_info(arguments), None
         elif tool_name == "git_status":
-            return self.tool_git_status(arguments)
+            return self.tool_git_status(arguments), None
         else:
-            return {"error": f"Unknown tool: {tool_name}"}
+            return None, {"code": -32602, "message": f"Unknown tool: {tool_name}"}
 
     def tool_read_file(self, args):
         path = os.path.expanduser(args.get("path", ""))
@@ -316,15 +393,21 @@ class MCPServer(BaseHTTPRequestHandler):
 
     def handle_read_resource(self, params):
         uri = params.get("uri", "")
+        if not isinstance(uri, str) or not uri:
+            return None, {"code": -32602, "message": "Invalid params"}
 
         if uri == "neuros://system/info":
-            return self.tool_get_system_info({})
+            return self.tool_get_system_info({}), None
         elif uri == "neuros://ollama/models":
-            return self.tool_ollama_models()
+            return self.tool_ollama_models(), None
         elif uri == "neuros://git/status":
-            return self.tool_git_status({})
+            return self.tool_git_status({}), None
         else:
-            return {"contents": [{"uri": uri, "mimeType": "text/plain", "text": "Resource not found"}]}
+            return None, {
+                "code": -32002,
+                "message": "Resource not found",
+                "data": {"uri": uri},
+            }
 
     def tool_ollama_models(self):
         try:

tests/test_mcp.py

@@ -5,6 +5,9 @@
 """
 
 import os
+import http.client
+import json
+import threading
 import tempfile
 import unittest
 import importlib.util
@@ -66,7 +69,108 @@ def test_run_command_missing_binary(self):
 
     def test_call_tool_unknown_name(self):
         result = self.handler.handle_call_tool({"name": "no_such_tool", "arguments": {}})
-        self.assertIn("error", result)
+        value, error = result
+        self.assertIsNone(value)
+        self.assertEqual(error["code"], -32602)
+
+
+class TestMCPWireProtocol(unittest.TestCase):
+    def setUp(self):
+        self.mcp = load_neuros_mcp()
+        self.server = self.mcp.HTTPServer(("127.0.0.1", 0), self.mcp.MCPServer)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.host, self.port = self.server.server_address
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=2)
+
+    def post(self, message, *, accept="application/json, text/event-stream",
+             content_type="application/json"):
+        body = json.dumps(message).encode("utf-8") if not isinstance(message, bytes) else message
+        connection = http.client.HTTPConnection(self.host, self.port, timeout=2)
+        connection.request("POST", "/", body=body,
+                           headers={"Accept": accept, "Content-Type": content_type})
+        response = connection.getresponse()
+        raw_body = response.read()
+        connection.close()
+        return response, raw_body
+
+    def test_initialize_handshake_and_capabilities_are_wire_valid(self):
+        response, body = self.post({
+            "jsonrpc": "2.0",
+            "id": 7,
+            "method": "initialize",
+            "params": {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {"roots": {"listChanged": True}},
+                "clientInfo": {"name": "wire-test", "version": "1.0"},
+            },
+        })
+        message = json.loads(body)
+        self.assertEqual(response.status, 200)
+        self.assertEqual(response.getheader("Content-Type"), "application/json")
+        self.assertEqual(int(response.getheader("Content-Length")), len(body))
+        self.assertEqual(message["jsonrpc"], "2.0")
+        self.assertEqual(message["id"], 7)
+        self.assertEqual(message["result"]["protocolVersion"], "2024-11-05")
+        self.assertEqual(message["result"]["capabilities"], {"tools": {}, "resources": {}})
+        self.assertEqual(set(message["result"]["serverInfo"]), {"name", "version"})
+
+    def test_initialize_rejects_unsupported_version_with_invalid_params(self):
+        response, body = self.post({
+            "jsonrpc": "2.0", "id": "version", "method": "initialize",
+            "params": {
+                "protocolVersion": "2099-01-01", "capabilities": {},
+                "clientInfo": {"name": "wire-test", "version": "1.0"},
+            },
+        })
+        message = json.loads(body)
+        self.assertEqual(response.status, 200)
+        self.assertEqual(message["error"]["code"], -32602)
+        self.assertEqual(message["error"]["data"], {
+            "supported": ["2024-11-05"], "requested": "2099-01-01"})
+
+    def test_json_rpc_error_code_mapping_is_wire_valid(self):
+        cases = [
+            (b"not-json", -32700, 400),
+            ({"jsonrpc": "2.0", "id": 1}, -32600, 400),
+            ({"jsonrpc": "2.0", "id": {"bad": "id"}, "method": "tools/list"}, -32600, 400),
+            ({"jsonrpc": "2.0", "id": 2, "method": "no/such"}, -32601, 200),
+            ({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
+              "params": {"name": "no_such_tool", "arguments": {}}}, -32602, 200),
+        ]
+        for message, expected_code, expected_status in cases:
+            with self.subTest(expected_code=expected_code):
+                response, body = self.post(message)
+                payload = json.loads(body)
+                self.assertEqual(response.status, expected_status)
+                self.assertEqual(payload["jsonrpc"], "2.0")
+                self.assertEqual(payload["error"]["code"], expected_code)
+                if expected_code == -32600:
+                    self.assertIsNone(payload["id"])
+
+    def test_notification_returns_202_without_json_rpc_body(self):
+        response, body = self.post({"jsonrpc": "2.0", "method": "notifications/initialized"})
+        self.assertEqual(response.status, 202)
+        self.assertEqual(body, b"")
+        self.assertEqual(response.getheader("Content-Length"), "0")
+
+    def test_streamable_http_accept_header_is_required(self):
+        response, body = self.post({"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
+                                   accept="application/json")
+        message = json.loads(body)
+        self.assertEqual(response.status, 400)
+        self.assertEqual(message["error"]["code"], -32600)
+
+    def test_json_content_type_must_be_exact_media_type(self):
+        response, body = self.post({"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
+                                   content_type="application/jsonp")
+        message = json.loads(body)
+        self.assertEqual(response.status, 400)
+        self.assertEqual(message["error"]["code"], -32600)
 
 
 if __name__ == "__main__":