#!/usr/bin/env python3 """MCP server exposing one sandboxed project directory as a terminal. Speaks MCP over stdio; `tunnel-client run` owns the process and carries the traffic. Nothing here listens on a socket. """ from __future__ import annotations import argparse import asyncio import sys from pathlib import Path from mcp.server import MCPServer from processes import ProcessRegistry, run_bounded from sandbox import MODES, Sandboxed, enforcement_holds, sandbox_command from workspace import resolve_cwd, resolve_readable_root, resolve_workspace DEFAULT_TIMEOUT_SECONDS = 120 MAX_TIMEOUT_SECONDS = 3600 OUTPUT_CHAR_LIMIT = 60_000 def _clip(text: str) -> str: if len(text) <= OUTPUT_CHAR_LIMIT: return text return text[:OUTPUT_CHAR_LIMIT] + f"\n[... {len(text) - OUTPUT_CHAR_LIMIT} more characters]" def build_server( workspace: Path, mode: str, network_access: bool, readable_roots: tuple[str, ...] = (), ) -> tuple[MCPServer, ProcessRegistry]: registry = ProcessRegistry() server = MCPServer( name="gptbridge", version="1.0.0", instructions=( f"A terminal scoped to the project directory {workspace}. Every command runs " f"under the {mode} sandbox: writes outside the workspace are refused, and the " "parent process's environment (credentials, tokens) is not inherited. Give each " "command an explicit cwd relative to the workspace root. Use run_command for " "anything that finishes; use start_process for anything that keeps running, such " "as a dev server, then read_output / send_input / stop_process to drive it." ), ) def wrap(command: str, cwd_arg: str | None) -> tuple[Sandboxed, Path]: cwd = resolve_cwd(workspace, cwd_arg) jail = sandbox_command( command, workspace=workspace, cwd=cwd, mode=mode, network_access=network_access, readable_roots=readable_roots, ) return jail, cwd @server.tool() async def run_command(command: str, cwd: str | None = None, timeout_seconds: int | None = None) -> dict: """Run a shell command to completion in the workspace and return its exit code and output. Args: command: shell command to run. cwd: directory to run in, relative to the workspace root; defaults to the root. timeout_seconds: kill the command after this long (default 120, max 3600). """ timeout = max(1, min(int(timeout_seconds or DEFAULT_TIMEOUT_SECONDS), MAX_TIMEOUT_SECONDS)) jail, run_dir = wrap(command, cwd) exit_code, output, truncated = await asyncio.to_thread( run_bounded, jail.argv, cwd=run_dir, env=jail.env, timeout=timeout ) payload = {"exit_code": exit_code, "cwd": str(run_dir), "output": _clip(output)} if exit_code is None: payload["killed"] = "output limit exceeded" if truncated else f"timed out after {timeout}s" return payload @server.tool() async def start_process(command: str, cwd: str | None = None) -> dict: """Start a long-running command in the background and return its process id. Args: command: shell command to start. cwd: directory to run in, relative to the workspace root; defaults to the root. """ jail, run_dir = wrap(command, cwd) managed = registry.start(jail.argv, command=command, cwd=run_dir, env=jail.env) return {"process_id": managed.process_id, "cwd": str(run_dir), "running": managed.running} @server.tool() async def read_output(process_id: str, since: int = 0) -> dict: """Read what a background process has printed since a byte offset. Args: process_id: id returned by start_process. since: byte offset to read from; pass back next_offset from the previous read. """ managed = registry.get(process_id) output, next_offset, dropped = managed.read(max(0, since)) payload = { "output": _clip(output), "next_offset": next_offset, "running": managed.running, "exit_code": managed.exit_code, } if dropped: payload["dropped_bytes"] = dropped return payload @server.tool() async def send_input(process_id: str, text: str) -> dict: """Write text to a background process's stdin. Args: process_id: id returned by start_process. text: exact text to write; include a trailing newline if the process reads lines. """ await asyncio.to_thread(registry.get(process_id).write_stdin, text) return {"sent": True} @server.tool() async def stop_process(process_id: str) -> dict: """Terminate a background process and everything it spawned. Args: process_id: id returned by start_process. """ return {"status": registry.get(process_id).stop()} @server.tool() async def list_processes() -> dict: """List background processes started in this session and whether each is still running.""" return { "processes": [ { "process_id": p.process_id, "command": p.command, "cwd": p.cwd, "running": p.running, "exit_code": p.exit_code, } for p in registry.all() ] } @server.tool() async def workspace_info() -> dict: """Report the workspace root, the sandbox mode, and whether commands may reach the network.""" return {"workspace": str(workspace), "sandbox_mode": mode, "network_access": network_access} return server, registry def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="gptbridge-mcp", description=__doc__) parser.add_argument("--workspace", required=True, help="the one project directory to expose") parser.add_argument("--sandbox-mode", default="workspace-write", choices=sorted(MODES)) parser.add_argument( "--network", action="store_true", help="let sandboxed commands reach the network (off by default: without it, a command " "that reads a file cannot send it anywhere)", ) parser.add_argument( "--readable-root", action="append", default=[], metavar="PATH", dest="readable_roots", help="make PATH readable inside the sandbox (repeatable); for toolchains outside " "the system directories, such as ~/.nvm or ~/.cargo", ) return parser async def serve(workspace: Path, mode: str, network_access: bool, readable_roots: tuple[str, ...]) -> None: server, registry = build_server(workspace, mode, network_access, readable_roots) try: await server.run_stdio_async() finally: registry.stop_all() def main(argv: list[str]) -> int: args = build_parser().parse_args(argv[1:]) try: workspace = resolve_workspace(args.workspace) readable_roots = tuple(str(resolve_readable_root(root)) for root in args.readable_roots) except ValueError as exc: print(f"gptbridge: {exc}", file=sys.stderr) return 2 enforced, detail = enforcement_holds( workspace, mode=args.sandbox_mode, network_access=args.network, readable_roots=readable_roots ) if not enforced: print(f"gptbridge: refusing to serve — {detail}", file=sys.stderr) return 3 print( f"gptbridge: serving {workspace} (sandbox={args.sandbox_mode}, " f"network={'on' if args.network else 'off'}, {detail})", file=sys.stderr, ) asyncio.run(serve(workspace, args.sandbox_mode, args.network, readable_roots)) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv))