shellmcp: run a bash command with timeout and cut output

shellmcp.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""shellmcp: an MCP server with one tool, `run`, that runs a bash command. Stdlib only, stdio.
+
+--mode readonly (default) runs every command in a bubblewrap sandbox: the filesystem is
+read-only except a private /tmp, there is no network, no other process is visible, and the
+--hide paths (always ~/.ssh and ~/.gnupg) look empty. --mode all runs commands with the
+user's full permissions; let the MCP client ask before each call.
+"""
+
+import argparse
+import getpass
+import json
+import os
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+
+ENV = {"PAGER": "cat", "GIT_PAGER": "cat", "NO_COLOR": "1", "TERM": "dumb"}
+
+
+def cut(text, limit):
+    """text, or its head and tail around a note when it is longer than limit."""
+    if len(text) <= limit:
+        return text
+    half = limit // 2
+    return f"{text[:half]}\n[... {len(text) - 2 * half} characters cut ...]\n{text[-half:]}"
+
+
+def run(command, cwd, timeout=60, limit=8000, prefix=()):
+    """The exit code and combined stdout and stderr of `bash -c command`, as one text."""
+    proc = subprocess.Popen([*prefix, "bash", "-c", command], cwd=cwd, stdin=subprocess.DEVNULL,
+                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                            env={**os.environ, **ENV}, start_new_session=True)
+    try:
+        out, _ = proc.communicate(timeout=timeout)
+        head = f"exit code {proc.returncode}"
+    except subprocess.TimeoutExpired:
+        os.killpg(proc.pid, signal.SIGKILL)
+        try:
+            out, _ = proc.communicate(timeout=5)
+        except subprocess.TimeoutExpired:  # a process left the group and holds the pipe
+            out = b""
+        head = f"timed out after {timeout} s, killed"
+    return f"{head}\n{cut(out.decode(errors='replace'), limit)}"

test_shellmcp.py

@@ -0,0 +1,32 @@
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from unittest import mock
+
+import shellmcp
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+
+
+class Run(unittest.TestCase):
+    def test_exit_code_and_combined_output(self):
+        self.assertEqual(shellmcp.run("echo hi; echo oops >&2; exit 3", HERE),
+                         "exit code 3\nhi\noops\n")
+
+    def test_long_output_keeps_head_and_tail(self):
+        out = shellmcp.run("printf 'a%.0s' $(seq 100); printf 'b%.0s' $(seq 100)", HERE, limit=20)
+        self.assertEqual(out, "exit code 0\naaaaaaaaaa\n[... 180 characters cut ...]\nbbbbbbbbbb")
+
+    def test_timeout_kills_the_whole_group(self):
+        start = time.monotonic()
+        out = shellmcp.run("sleep 30 & sleep 30", HERE, timeout=1)
+        self.assertLess(time.monotonic() - start, 5)
+        self.assertEqual(out, "timed out after 1 s, killed\n")
+
+    def test_no_stdin_and_no_pager(self):
+        self.assertEqual(shellmcp.run("cat; echo $PAGER", HERE), "exit code 0\ncat\n")