terminal-mcp-server

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { exec } from "child_process";
import * as fs from "fs";
import * as path from "path";
import * as os from "os";

// ─── Config ──────────────────────────────────────────────────────────────────

const MAX_OUTPUT_CHARS = 50_000;
const DEFAULT_TIMEOUT_MS = 30_000;

const DANGEROUS_COMMANDS = [
  /rm\s+-rf\s+\/(?:\s|$)/,
  /mkfs/,
  /:\(\)\{.*\}/,
  /dd\s+if=.*of=\/dev\/(sd|hd)/,
  />\s*\/dev\/(sd|hd)/,
  /shutdown/,
  /reboot/,
  /halt/,
  /init\s+0/,
];

function isSafeCommand(cmd: string): { safe: boolean; reason?: string } {
  if (process.env.ALLOW_DANGEROUS === "1") return { safe: true };
  for (const pattern of DANGEROUS_COMMANDS) {
    if (pattern.test(cmd)) {
      return { safe: false, reason: `Blocked by pattern: ${pattern}` };
    }
  }
  return { safe: true };
}

function truncate(text: string): string {
  if (text.length <= MAX_OUTPUT_CHARS) return text;
  const half = MAX_OUTPUT_CHARS / 2;
  return (
    text.slice(0, half) +
    `\n\n[... ${text.length - MAX_OUTPUT_CHARS} chars truncated ...]\n\n` +
    text.slice(-half)
  );
}

/** Format mtime as "YYYY-MM-DD HH:MM" — precise enough, far shorter than ISO 8601 */
function fmtTime(d: Date): string {
  return d.toISOString().slice(0, 16).replace("T", " ");
}

// ─── Path helpers ─────────────────────────────────────────────────────────────

function resolvePath(userPath: string, cwd: string): string {
  if (userPath.startsWith("~")) {
    return path.resolve(os.homedir(), userPath.slice(userPath.startsWith("~/") ? 2 : 1));
  }
  return path.isAbsolute(userPath) ? userPath : path.resolve(cwd, userPath);
}

// ─── State ────────────────────────────────────────────────────────────────────

function resolveStartingCwd(): string {
  const envCwd = process.env.TERMINAL_CWD;
  if (envCwd) {
    try {
      if (fs.statSync(envCwd).isDirectory()) return path.resolve(envCwd);
      console.error(`[terminal-mcp] TERMINAL_CWD is not a directory: ${envCwd}. Falling back to process.cwd().`);
    } catch {
      console.error(`[terminal-mcp] TERMINAL_CWD does not exist: ${envCwd}. Falling back to process.cwd().`);
    }
  }
  return process.cwd();
}

let currentWorkingDir: string = resolveStartingCwd();

// ─── Helpers ──────────────────────────────────────────────────────────────────

/** Compact MCP text response */
function ok(data: unknown) {
  return { content: [{ type: "text" as const, text: JSON.stringify(data) }] };
}

function err(message: string) {
  return ok({ error: message });
}

// ─── Server ───────────────────────────────────────────────────────────────────

const server = new McpServer({
  name: "terminal-mcp-server",
  version: "1.0.0",
});

// ─── Tools ────────────────────────────────────────────────────────────────────

server.registerTool(
  "terminal_execute",
  {
    title: "Execute Shell Command",
    description:
      "Run a shell command in the current working directory. " +
      "The working directory persists across calls — use terminal_cd to change it. " +
      "Dangerous commands (rm -rf /, mkfs, fork bombs, etc.) are blocked unless ALLOW_DANGEROUS=1.",
    inputSchema: z.object({
      command: z.string().min(1).describe("Shell command to execute"),
      timeout_ms: z
        .number().int().min(1000).max(300_000)
        .default(DEFAULT_TIMEOUT_MS)
        .describe("Timeout in ms (default 30000)"),
      env: z.record(z.string()).optional().describe("Extra env vars for this command"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: true,
      idempotentHint: false,
      openWorldHint: false,
    },
  },
  async ({ command, timeout_ms, env }) => {
    const check = isSafeCommand(command);
    if (!check.safe) {
      return ok({ exit_code: 1, stderr: `Command blocked. ${check.reason}` });
    }

    const controller = new AbortController();
    const timer = setTimeout(() => controller.abort(), timeout_ms);

    try {
      const { stdout, stderr } = await new Promise<{ stdout: string; stderr: string }>(
        (resolve, reject) => {
          exec(
            command,
            {
              cwd: currentWorkingDir,
              env: { ...process.env, ...env },
              maxBuffer: 10 * 1024 * 1024,
              signal: controller.signal,
            },
            (e, stdout, stderr) => {
              if (e) reject(Object.assign(e, { stdout, stderr }));
              else resolve({ stdout, stderr });
            }
          );
        }
      );

      clearTimeout(timer);
      // Only include stdout/stderr when non-empty
      const result: Record<string, unknown> = { exit_code: 0 };
      if (stdout.trim()) result.stdout = truncate(stdout);
      if (stderr.trim()) result.stderr = truncate(stderr);
      return ok(result);
    } catch (e: unknown) {
      clearTimeout(timer);

      const isAbort =
        (e as NodeJS.ErrnoException).code === "ABORT_ERR" ||
        (e as Error).name === "AbortError";

      if (isAbort) {
        return ok({ exit_code: 124, timed_out: true, stderr: `Killed after ${timeout_ms}ms` });
      }

      const execErr = e as { stdout?: string; stderr?: string; code?: number };
      const result: Record<string, unknown> = { exit_code: execErr.code ?? 1 };
      if (execErr.stdout?.trim()) result.stdout = truncate(execErr.stdout);
      if (execErr.stderr?.trim()) result.stderr = truncate(execErr.stderr);
      else result.stderr = String(e);
      return ok(result);
    }
  }
);

server.registerTool(
  "terminal_cd",
  {
    title: "Change Directory",
    description: "Change the working directory for subsequent terminal_execute calls. Supports ~, .., relative and absolute paths.",
    inputSchema: z.object({
      path: z.string().min(1).describe("Target directory path"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: false,
    },
  },
  async ({ path: targetPath }) => {
    const resolved = resolvePath(targetPath, currentWorkingDir);
    try {
      const stat = fs.statSync(resolved);
      if (!stat.isDirectory()) throw new Error(`Not a directory: ${resolved}`);
      currentWorkingDir = resolved;
      return ok({ cwd: currentWorkingDir });
    } catch (e) {
      return err(`Failed to change directory: ${(e as Error).message}`);
    }
  }
);

server.registerTool(
  "terminal_pwd",
  {
    title: "Print Working Directory",
    description: "Return the current working directory used by terminal_execute.",
    inputSchema: z.object({}),
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: false,
    },
  },
  async () => ok({ cwd: currentWorkingDir })
);

server.registerTool(
  "terminal_read_file",
  {
    title: "Read File",
    description:
      "Read a file's contents as UTF-8 text, optionally restricted to a line range (1-based, inclusive). " +
      "Defaults to the first 100 lines when no range is given. " +
      "Returns content, total_lines, and size_bytes.",
    inputSchema: z.object({
      path: z.string().min(1).describe("File path (absolute, relative to cwd, or ~)"),
      start_line: z.number().int().min(1).optional().describe("First line to return (default: 1)"),
      end_line: z.number().int().min(1).optional().describe("Last line to return (default: 100 lines from start_line)"),
    }),
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: false,
    },
  },
  async ({ path: filePath, start_line, end_line }) => {
    const resolved = resolvePath(filePath, currentWorkingDir);
    try {
      const stat = fs.statSync(resolved);
      const raw = fs.readFileSync(resolved, "utf8");
      const allLines = raw.split("\n");
      const totalLines = allLines.length;

      const from = (start_line ?? 1) - 1;
      const to = end_line !== undefined ? end_line : from + 100;

      if (from < 0 || from >= totalLines) {
        return err(`start_line ${start_line} out of range (file has ${totalLines} lines)`);
      }
      if (to < from + 1) {
        return err("end_line must be >= start_line");
      }

      const slice = allLines.slice(from, to);
      const result: Record<string, unknown> = {
        content: slice.join("\n"),
        total_lines: totalLines,
        size_bytes: stat.size,
      };
      // Only include range info when a partial read was requested
      if (start_line !== undefined || end_line !== undefined) {
        result.range = `${from + 1}-${from + slice.length}`;
      }
      return ok(result);
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

server.registerTool(
  "terminal_write_file",
  {
    title: "Write File",
    description:
      "Write UTF-8 text to a file, creating it (and any parent directories) if needed. " +
      "Set append=true to append instead of overwrite.",
    inputSchema: z.object({
      path: z.string().min(1).describe("File path to write"),
      content: z.string().describe("Content to write"),
      append: z.boolean().default(false).describe("Append instead of overwrite"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: true,
      idempotentHint: false,
      openWorldHint: false,
    },
  },
  async ({ path: filePath, content, append }) => {
    const resolved = resolvePath(filePath, currentWorkingDir);
    try {
      fs.mkdirSync(path.dirname(resolved), { recursive: true });
      const buf = Buffer.from(content, "utf8");
      fs.writeFileSync(resolved, buf, { flag: append ? "a" : "w" });
      return ok({ bytes_written: buf.length });
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

server.registerTool(
  "terminal_list_dir",
  {
    title: "List Directory",
    description: "List files and directories with name, type, size, and last-modified time. Defaults to the current working directory.",
    inputSchema: z.object({
      path: z.string().optional().describe("Directory path (default: cwd)"),
      show_hidden: z.boolean().default(false).describe("Include dot-files"),
    }),
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: false,
    },
  },
  async ({ path: dirPath, show_hidden }) => {
    const resolved = dirPath ? resolvePath(dirPath, currentWorkingDir) : currentWorkingDir;
    try {
      const names = fs.readdirSync(resolved);
      const entries = names
        .filter((n) => show_hidden || !n.startsWith("."))
        .map((name) => {
          try {
            const stat = fs.lstatSync(path.join(resolved, name));
            const type = stat.isDirectory()
              ? "dir"
              : stat.isSymbolicLink()
              ? "symlink"
              : stat.isFile()
              ? "file"
              : "other";
            return { name, type, size: stat.size, modified: fmtTime(stat.mtime) };
          } catch {
            return { name, type: "other", size: 0, modified: "" };
          }
        });
      return ok({ entries });
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

server.registerTool(
  "terminal_env",
  {
    title: "Get Environment Variables",
    description:
      "Return environment variables for this process. " +
      "Sensitive names (SECRET, PASSWORD, TOKEN, KEY, CREDENTIAL) are masked unless SHOW_SECRETS=1. " +
      "Use filter to narrow results.",
    inputSchema: z.object({
      filter: z.string().optional().describe("Case-insensitive substring filter on variable names"),
    }),
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: false,
    },
  },
  async ({ filter }) => {
    const SENSITIVE = /secret|password|token|key|credential/i;
    const result: Record<string, string> = {};
    for (const [k, v] of Object.entries(process.env)) {
      if (filter && !k.toLowerCase().includes(filter.toLowerCase())) continue;
      result[k] =
        process.env.SHOW_SECRETS === "1"
          ? (v ?? "")
          : SENSITIVE.test(k)
          ? "***"
          : (v ?? "");
    }
    return ok(result);
  }
);

server.registerTool(
  "terminal_edit_file",
  {
    title: "Edit File",
    description:
      "Apply one or more find-and-replace operations to a file in order. " +
      "Each old_str must exist exactly once. " +
      "Use dry_run=true to preview without writing, show_result=true to include the final content in the response.",
    inputSchema: z.object({
      path: z.string().min(1).describe("Path to the file to edit"),
      edits: z
        .array(
          z.object({
            old_str: z.string().describe("Exact text to find (must appear exactly once)"),
            new_str: z.string().describe("Replacement text (empty string to delete)"),
          })
        )
        .min(1)
        .describe("Ordered list of find-and-replace operations"),
      dry_run: z.boolean().default(false).describe("Preview without writing to disk"),
      show_result: z.boolean().default(false).describe("Include full file content in response"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: true,
      idempotentHint: false,
      openWorldHint: false,
    },
  },
  async ({ path: filePath, edits, dry_run, show_result }) => {
    const resolved = resolvePath(filePath, currentWorkingDir);

    let current: string;
    try {
      current = fs.readFileSync(resolved, "utf8");
    } catch {
      return err(`File not found: ${resolved}`);
    }

    let editsApplied = 0;
    for (const { old_str, new_str } of edits) {
      const count = current.split(old_str).length - 1;
      if (count === 0) return err(`old_str not found: ${JSON.stringify(old_str)}`);
      if (count > 1) return err(`old_str matches ${count} times (must be unique): ${JSON.stringify(old_str)}`);
      current = current.replace(old_str, new_str);
      editsApplied++;
    }

    if (!dry_run) {
      fs.mkdirSync(path.dirname(resolved), { recursive: true });
      fs.writeFileSync(resolved, current, "utf8");
    }

    const response: Record<string, unknown> = { edits_applied: editsApplied };
    if (dry_run) response.dry_run = true;
    if (show_result) response.content = truncate(current);
    return ok(response);
  }
);

server.registerTool(
  "terminal_mkdir",
  {
    title: "Create Directory",
    description: "Create a directory including any missing parents. Idempotent — succeeds silently if it already exists.",
    inputSchema: z.object({
      path: z.string().min(1).describe("Directory path to create"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: false,
    },
  },
  async ({ path: dirPath }) => {
    const resolved = resolvePath(dirPath, currentWorkingDir);
    try {
      const alreadyExists = fs.existsSync(resolved);
      if (alreadyExists) {
        if (!fs.statSync(resolved).isDirectory()) {
          return err(`Path exists but is not a directory: ${resolved}`);
        }
        return ok({ created: false });
      }
      fs.mkdirSync(resolved, { recursive: true });
      return ok({ created: true });
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

server.registerTool(
  "terminal_delete",
  {
    title: "Delete File or Directory",
    description:
      "Delete a file or directory. Directories are removed recursively by default. " +
      "Set recursive=false to fail on non-empty directories.",
    inputSchema: z.object({
      path: z.string().min(1).describe("Path to delete"),
      recursive: z.boolean().default(true).describe("Delete directory contents recursively"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: true,
      idempotentHint: false,
      openWorldHint: false,
    },
  },
  async ({ path: targetPath, recursive }) => {
    const resolved = resolvePath(targetPath, currentWorkingDir);
    try {
      if (!fs.existsSync(resolved)) return err(`Not found: ${resolved}`);
      const isDir = fs.statSync(resolved).isDirectory();
      if (isDir) fs.rmSync(resolved, { recursive, force: false });
      else fs.unlinkSync(resolved);
      return ok({ deleted: resolved });
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

server.registerTool(
  "terminal_copy",
  {
    title: "Copy File or Directory",
    description:
      "Copy a file or directory to a new location (recursive for directories). " +
      "If destination is an existing directory, the source is copied inside it.",
    inputSchema: z.object({
      source: z.string().min(1).describe("Source path"),
      destination: z.string().min(1).describe("Destination path or directory"),
      overwrite: z.boolean().default(false).describe("Overwrite if destination already exists"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: false,
      idempotentHint: false,
      openWorldHint: false,
    },
  },
  async ({ source, destination, overwrite }) => {
    const resolvedSrc = resolvePath(source, currentWorkingDir);
    let resolvedDst = resolvePath(destination, currentWorkingDir);

    try {
      if (!fs.existsSync(resolvedSrc)) return err(`Source not found: ${resolvedSrc}`);

      const srcType = fs.statSync(resolvedSrc).isDirectory() ? "directory" : "file";

      if (fs.existsSync(resolvedDst) && fs.statSync(resolvedDst).isDirectory()) {
        resolvedDst = path.join(resolvedDst, path.basename(resolvedSrc));
      }
      if (fs.existsSync(resolvedDst) && !overwrite) {
        return err(`Destination already exists: ${resolvedDst}. Use overwrite=true.`);
      }

      fs.mkdirSync(path.dirname(resolvedDst), { recursive: true });
      fs.cpSync(resolvedSrc, resolvedDst, { recursive: true, force: overwrite });
      return ok({ destination: resolvedDst, type: srcType });
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

server.registerTool(
  "terminal_move",
  {
    title: "Move / Rename File or Directory",
    description:
      "Move or rename a file or directory. " +
      "If destination is an existing directory, the source is moved inside it. " +
      "Falls back to copy+delete for cross-device moves.",
    inputSchema: z.object({
      source: z.string().min(1).describe("Source path"),
      destination: z.string().min(1).describe("Destination path or directory"),
      overwrite: z.boolean().default(false).describe("Overwrite if destination already exists"),
    }),
    annotations: {
      readOnlyHint: false,
      destructiveHint: true,
      idempotentHint: false,
      openWorldHint: false,
    },
  },
  async ({ source, destination, overwrite }) => {
    const resolvedSrc = resolvePath(source, currentWorkingDir);
    let resolvedDst = resolvePath(destination, currentWorkingDir);

    try {
      if (!fs.existsSync(resolvedSrc)) return err(`Source not found: ${resolvedSrc}`);

      const srcType = fs.statSync(resolvedSrc).isDirectory() ? "directory" : "file";

      if (fs.existsSync(resolvedDst) && fs.statSync(resolvedDst).isDirectory()) {
        resolvedDst = path.join(resolvedDst, path.basename(resolvedSrc));
      }
      if (fs.existsSync(resolvedDst) && !overwrite) {
        return err(`Destination already exists: ${resolvedDst}. Use overwrite=true.`);
      }

      fs.mkdirSync(path.dirname(resolvedDst), { recursive: true });

      try {
        fs.renameSync(resolvedSrc, resolvedDst);
      } catch (renameErr) {
        if ((renameErr as NodeJS.ErrnoException).code === "EXDEV") {
          // Cross-device: copy then delete
          fs.cpSync(resolvedSrc, resolvedDst, { recursive: true, force: overwrite });
          fs.rmSync(resolvedSrc, { recursive: true, force: true });
        } else {
          throw renameErr;
        }
      }

      return ok({ destination: resolvedDst, type: srcType });
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

server.registerTool(
  "terminal_search_files",
  {
    title: "Search Files",
    description:
      "Recursively search for files/directories by name substring. " +
      "Applies include_pattern (whitelist) and exclude_pattern (blacklist) to entry names (case-insensitive).",
    inputSchema: z.object({
      path: z.string().optional().describe("Root directory to search (default: cwd)"),
      include_pattern: z.string().optional().describe("Include entries whose name contains this"),
      exclude_pattern: z.string().optional().describe("Exclude entries whose name contains this"),
      type: z.enum(["file", "directory", "any"]).default("any").describe("Filter by entry type"),
      max_depth: z.number().int().min(1).max(50).default(10).describe("Max recursion depth"),
      max_results: z.number().int().min(1).max(1000).default(200).describe("Max results to return"),
      show_hidden: z.boolean().default(false).describe("Include hidden entries (starting with .)"),
    }),
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: false,
    },
  },
  async ({ path: searchPath, include_pattern, exclude_pattern, type, max_depth, max_results, show_hidden }) => {
    const root = searchPath ? resolvePath(searchPath, currentWorkingDir) : currentWorkingDir;

    interface SearchEntry {
      path: string;
      type: "file" | "directory";
      size: number;
      modified: string;
    }

    const results: SearchEntry[] = [];
    let capped = false;

    function walk(dir: string, depth: number): void {
      if (depth > max_depth || capped) return;
      let names: string[];
      try { names = fs.readdirSync(dir); } catch { return; }

      for (const name of names) {
        if (capped) return;
        if (!show_hidden && name.startsWith(".")) continue;

        const full = path.join(dir, name);
        let stat: fs.Stats;
        try { stat = fs.lstatSync(full); } catch { continue; }

        const isDir = stat.isDirectory();
        const entryType: "file" | "directory" = isDir ? "directory" : "file";

        if (type !== "any" && entryType !== type) {
          if (isDir) walk(full, depth + 1);
          continue;
        }

        const nameLower = name.toLowerCase();
        if (include_pattern && !nameLower.includes(include_pattern.toLowerCase())) {
          if (isDir) walk(full, depth + 1);
          continue;
        }
        if (exclude_pattern && nameLower.includes(exclude_pattern.toLowerCase())) {
          if (isDir) walk(full, depth + 1);
          continue;
        }

        results.push({ path: full, type: entryType, size: stat.size, modified: fmtTime(stat.mtime) });

        if (results.length >= max_results) { capped = true; return; }
        if (isDir) walk(full, depth + 1);
      }
    }

    try {
      if (!fs.existsSync(root)) return err(`Search root not found: ${root}`);
      walk(root, 0);
      const response: Record<string, unknown> = { results };
      if (capped) response.capped = true;
      return ok(response);
    } catch (e) {
      return err((e as Error).message);
    }
  }
);

// ─── Transport ────────────────────────────────────────────────────────────────

async function main(): Promise<void> {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("terminal-mcp-server running on stdio");
}

main().catch((e) => {
  console.error("Fatal error:", e);
  process.exit(1);
});