shellmcp: MCP stdio server and CLI

shellmcp.py

@@ -81,3 +81,73 @@ def describe(mode, cwd, timeout, limit):
     if tools:
         text += " Installed, prefer these: " + "; ".join(tools) + "."
     return text
+
+PROTOCOLS = ("2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05")  # newest first
+HIDDEN = ("~/.ssh", "~/.gnupg")
+
+
+def error(mid, code, message):
+    return {"jsonrpc": "2.0", "id": mid, "error": {"code": code, "message": message}}
+
+
+def handle(message, tool, call):
+    """The JSON-RPC response to one message; None for a notification."""
+    mid, method, params = message.get("id"), message.get("method"), message.get("params") or {}
+    if mid is None:
+        return None
+    if method == "initialize":
+        asked = params.get("protocolVersion")
+        result = {"protocolVersion": asked if asked in PROTOCOLS else PROTOCOLS[0],
+                  "capabilities": {"tools": {}},
+                  "serverInfo": {"name": "shellmcp", "version": "0.1.0"}}
+    elif method == "ping":
+        result = {}
+    elif method == "tools/list":
+        result = {"tools": [tool]}
+    elif method == "tools/call":
+        command = (params.get("arguments") or {}).get("command")
+        if params.get("name") != "run" or not isinstance(command, str):
+            return error(mid, -32602, "call the tool `run` with a string `command`")
+        result = {"content": [{"type": "text", "text": call(command)}], "isError": False}
+    else:
+        return error(mid, -32601, f"unknown method {method}")
+    return {"jsonrpc": "2.0", "id": mid, "result": result}
+
+
+def main(argv=None):
+    p = argparse.ArgumentParser(description="MCP server with one tool, run: a bash command.")
+    p.add_argument("--mode", choices=("readonly", "all"), default="readonly",
+                   help="readonly: bwrap sandbox (default); all: the user's full permissions")
+    p.add_argument("--cwd", default=os.getcwd(), help="directory commands run in")
+    p.add_argument("--timeout", type=int, default=60, help="seconds before a command is killed")
+    p.add_argument("--max-output", type=int, default=8000, help="characters of output kept")
+    p.add_argument("--hide", action="append", default=[], metavar="PATH",
+                   help="readonly: also make PATH look empty (repeatable; ~/.ssh and ~/.gnupg "
+                        "always are)")
+    a = p.parse_args(argv)
+    cwd = os.path.abspath(a.cwd)
+    if a.mode == "readonly" and not shutil.which("bwrap"):
+        sys.exit("shellmcp: --mode readonly needs bwrap (bubblewrap)")
+    hide = [os.path.expanduser(h) for h in (*HIDDEN, *a.hide)]
+    prefix = sandbox(cwd, hide) if a.mode == "readonly" else ()
+    tool = {"name": "run", "description": describe(a.mode, cwd, a.timeout, a.max_output),
+            "inputSchema": {"type": "object", "required": ["command"], "properties": {
+                "command": {"type": "string", "description": "a bash command line"}}}}
+    # ponytail: one request at a time; a long command delays the next call (no cancellation)
+    for line in sys.stdin:
+        if not line.strip():
+            continue
+        try:
+            message = json.loads(line)
+        except ValueError:
+            message = None
+        if not isinstance(message, dict):
+            reply = error(None, -32700, "parse error")
+        else:
+            reply = handle(message, tool, lambda c: run(c, cwd, a.timeout, a.max_output, prefix))
+        if reply is not None:
+            print(json.dumps(reply), flush=True)
+
+
+if __name__ == "__main__":
+    main()

test_shellmcp.py

@@ -55,3 +55,36 @@ class Describe(unittest.TestCase):
                       "fd: find files (instead of find).", text)
         self.assertNotIn("jq", text)
         self.assertIn("full permissions", shellmcp.describe("all", "/srv", 60, 8000))
+
+class Protocol(unittest.TestCase):
+    def talk(self, *messages):
+        lines = "".join(json.dumps(m) + "\n" for m in messages) + "not json\n"
+        done = subprocess.run([sys.executable, os.path.join(HERE, "shellmcp.py"), "--mode", "all",
+                               "--cwd", HERE], input=lines, capture_output=True, text=True,
+                              timeout=10)
+        return [json.loads(line) for line in done.stdout.splitlines()]
+
+    def test_initialize_list_and_call(self):
+        replies = self.talk(
+            {"jsonrpc": "2.0", "id": 1, "method": "initialize",
+             "params": {"protocolVersion": "2025-06-18", "capabilities": {},
+                        "clientInfo": {"name": "test", "version": "0"}}},
+            {"jsonrpc": "2.0", "method": "notifications/initialized"},
+            {"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
+            {"jsonrpc": "2.0", "id": 3, "method": "tools/call",
+             "params": {"name": "run", "arguments": {"command": "echo hi"}}},
+            {"jsonrpc": "2.0", "id": 4, "method": "tools/call",
+             "params": {"name": "nope", "arguments": {}}},
+            {"jsonrpc": "2.0", "id": 5, "method": "resources/list"})
+        self.assertEqual([r.get("id") for r in replies], [1, 2, 3, 4, 5, None])
+        self.assertEqual(replies[0]["result"]["protocolVersion"], "2025-06-18")
+        self.assertEqual(replies[1]["result"]["tools"][0]["name"], "run")
+        self.assertEqual(replies[2]["result"],
+                         {"content": [{"type": "text", "text": "exit code 0\nhi\n"}],
+                          "isError": False})
+        self.assertEqual([r["error"]["code"] for r in replies[3:]], [-32602, -32601, -32700])
+
+    def test_unknown_protocol_gets_the_newest(self):
+        reply = self.talk({"jsonrpc": "2.0", "id": 1, "method": "initialize",
+                           "params": {"protocolVersion": "1999-01-01"}})[0]
+        self.assertEqual(reply["result"]["protocolVersion"], shellmcp.PROTOCOLS[0])