Terminal Controller for MCP

Instructions

Implementation Reference

• terminal_controller.py:102-143 (handler)

  The primary handler for the 'execute_command' MCP tool. Decorated with @mcp.tool() for automatic registration. Performs basic security checks on dangerous commands, delegates execution to the run_command helper, and formats the success/error output for the user.
  @mcp.tool() async def execute_command(command: str, timeout: int = 30) -> str: """ Execute terminal command and return results Args: command: Command line command to execute timeout: Command timeout in seconds, default is 30 seconds Returns: Output of the command execution """ # Check for dangerous commands (can add more security checks) dangerous_commands = ["rm -rf /", "mkfs"] if any(dc in command.lower() for dc in dangerous_commands): return "For security reasons, this command is not allowed." result = await run_command(command, timeout) if result["success"]: output = f"Command executed successfully (duration: {result['duration']})\n\n" if result["stdout"]: output += f"Output:\n{result['stdout']}\n" else: output += "Command had no output.\n" if result["stderr"]: output += f"\nWarnings/Info:\n{result['stderr']}" return output else: output = f"Command execution failed (duration: {result['duration']})\n" if result["stdout"]: output += f"\nOutput:\n{result['stdout']}\n" if result["stderr"]: output += f"\nError:\n{result['stderr']}" output += f"\nReturn code: {result['return_code']}" return output
• terminal_controller.py:19-100 (helper)

  Core helper function that performs the actual command execution using asyncio.create_subprocess_shell, handles platform differences (Windows vs Unix), timeout with process.kill(), captures stdout/stderr/returncode, computes duration, maintains command history, and returns structured result dictionary.
  async def run_command(cmd: str, timeout: int = 30) -> Dict: """ Execute command and return results Args: cmd: Command to execute timeout: Command timeout in seconds Returns: Dictionary containing command execution results """ start_time = datetime.now() try: # Create command appropriate for current OS if platform.system() == "Windows": process = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, shell=True ) else: process = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, shell=True, executable="/bin/bash" ) try: stdout, stderr = await asyncio.wait_for(process.communicate(), timeout) stdout = stdout.decode('utf-8', errors='replace') stderr = stderr.decode('utf-8', errors='replace') return_code = process.returncode except asyncio.TimeoutError: try: process.kill() except: pass return { "success": False, "stdout": "", "stderr": f"Command timed out after {timeout} seconds", "return_code": -1, "duration": str(datetime.now() - start_time), "command": cmd } duration = datetime.now() - start_time result = { "success": return_code == 0, "stdout": stdout, "stderr": stderr, "return_code": return_code, "duration": str(duration), "command": cmd } # Add to history command_history.append({ "timestamp": datetime.now().isoformat(), "command": cmd, "success": return_code == 0 }) # If history is too long, remove oldest record if len(command_history) > MAX_HISTORY_SIZE: command_history.pop(0) return result except Exception as e: return { "success": False, "stdout": "", "stderr": f"Error executing command: {str(e)}", "return_code": -1, "duration": str(datetime.now() - start_time), "command": cmd }