diff --git a/shellmcp.py b/shellmcp.py old mode 100644 new mode 100755 index 98edcbe..ae2e00f --- a/shellmcp.py +++ b/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() diff --git a/test_shellmcp.py b/test_shellmcp.py index d3c5771..17643fd 100644 --- a/test_shellmcp.py +++ b/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])