terminal_controller.py

import asyncio
import os
import subprocess
import platform
import sys
from typing import List, Dict, Optional
from datetime import datetime
from mcp.server.fastmcp import FastMCP

# Initialize MCP server
mcp = FastMCP("terminal-controller", log_level="INFO")

# List to store command history
command_history = []

# Maximum history size
MAX_HISTORY_SIZE = 50

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
        }

@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

@mcp.tool()
async def get_command_history(count: int = 10) -> str:
    """
    Get recent command execution history
    
    Args:
        count: Number of recent commands to return
    
    Returns:
        Formatted command history record
    """
    if not command_history:
        return "No command execution history."
    
    count = min(count, len(command_history))
    recent_commands = command_history[-count:]
    
    output = f"Recent {count} command history:\n\n"
    
    for i, cmd in enumerate(recent_commands):
        status = "✓" if cmd["success"] else "✗"
        output += f"{i+1}. [{status}] {cmd['timestamp']}: {cmd['command']}\n"
    
    return output

@mcp.tool()
async def get_current_directory() -> str:
    """
    Get current working directory
    
    Returns:
        Path of current working directory
    """
    return os.getcwd()

@mcp.tool()
async def change_directory(path: str) -> str:
    """
    Change current working directory
    
    Args:
        path: Directory path to switch to
    
    Returns:
        Operation result information
    """
    try:
        os.chdir(path)
        return f"Switched to directory: {os.getcwd()}"
    except FileNotFoundError:
        return f"Error: Directory '{path}' does not exist"
    except PermissionError:
        return f"Error: No permission to access directory '{path}'"
    except Exception as e:
        return f"Error changing directory: {str(e)}"

@mcp.tool()
async def list_directory(path: Optional[str] = None) -> str:
    """
    List files and subdirectories in the specified directory
    
    Args:
        path: Directory path to list contents, default is current directory
    
    Returns:
        List of directory contents
    """
    if path is None:
        path = os.getcwd()
    
    try:
        items = os.listdir(path)
        
        dirs = []
        files = []
        
        for item in items:
            full_path = os.path.join(path, item)
            if os.path.isdir(full_path):
                dirs.append(f"📁 {item}/")
            else:
                files.append(f"📄 {item}")
        
        # Sort directories and files
        dirs.sort()
        files.sort()
        
        if not dirs and not files:
            return f"Directory '{path}' is empty"
        
        output = f"Contents of directory '{path}':\n\n"
        
        if dirs:
            output += "Directories:\n"
            output += "\n".join(dirs) + "\n\n"
        
        if files:
            output += "Files:\n"
            output += "\n".join(files)
        
        return output
    
    except FileNotFoundError:
        return f"Error: Directory '{path}' does not exist"
    except PermissionError:
        return f"Error: No permission to access directory '{path}'"
    except Exception as e:
        return f"Error listing directory contents: {str(e)}"

@mcp.tool()
async def write_file(path: str, content: str, mode: str = "overwrite") -> str:
    """
    Write content to a file
    
    Args:
        path: Path to the file
        content: Content to write (string or JSON object)
        mode: Write mode ('overwrite' or 'append')
    
    Returns:
        Operation result information
    """
    try:
        # Handle different content types
        if not isinstance(content, str):
            try:
                import json
                # Advanced JSON serialization with better handling of complex objects
                content = json.dumps(content, 
                                    indent=4, 
                                    sort_keys=False, 
                                    ensure_ascii=False, 
                                    default=lambda obj: str(obj) if hasattr(obj, '__dict__') else repr(obj))
            except Exception as e:
                # Try a more aggressive approach if standard serialization fails
                try:
                    # Convert object to dictionary first if it has __dict__
                    if hasattr(content, '__dict__'):
                        import json
                        content = json.dumps(content.__dict__, 
                                           indent=4, 
                                           sort_keys=False, 
                                           ensure_ascii=False)
                    else:
                        # Last resort: convert to string representation
                        content = str(content)
                except Exception as inner_e:
                    return f"Error: Unable to convert complex object to writable string: {str(e)}, then tried alternative method and got: {str(inner_e)}"
        
        # Choose file mode based on the specified writing mode
        file_mode = "w" if mode.lower() == "overwrite" else "a"
        
        # Ensure content ends with a newline if it doesn't already
        if content and not content.endswith('\n'):
            content += '\n'
        
        # Ensure directory exists
        directory = os.path.dirname(os.path.abspath(path))
        if directory and not os.path.exists(directory):
            os.makedirs(directory, exist_ok=True)
            
        with open(path, file_mode, encoding="utf-8") as file:
            file.write(content)
        
        # Verify the write operation was successful
        if os.path.exists(path):
            file_size = os.path.getsize(path)
            return f"Successfully wrote {file_size} bytes to '{path}' in {mode} mode."
        else:
            return f"Write operation completed, but unable to verify file exists at '{path}'."
    
    except FileNotFoundError:
        return f"Error: The directory in path '{path}' does not exist and could not be created."
    except PermissionError:
        return f"Error: No permission to write to file '{path}'."
    except Exception as e:
        return f"Error writing to file: {str(e)}"

@mcp.tool()
async def read_file(path: str, start_row: int = None, end_row: int = None, as_json: bool = False) -> str:
    """
    Read content from a file with optional row selection
    
    Args:
        path: Path to the file
        start_row: Starting row to read from (0-based, optional)
        end_row: Ending row to read to (0-based, inclusive, optional)
        as_json: If True, attempt to parse file content as JSON (optional)
    
    Returns:
        File content or selected lines, optionally parsed as JSON
    """
    try:
        if not os.path.exists(path):
            return f"Error: File '{path}' does not exist."
            
        if not os.path.isfile(path):
            return f"Error: '{path}' is not a file."
        
        # Check file size before reading to prevent memory issues
        file_size = os.path.getsize(path)
        if file_size > 10 * 1024 * 1024:  # 10 MB limit
            return f"Warning: File is very large ({file_size/1024/1024:.2f} MB). Consider using row selection."
            
        with open(path, 'r', encoding='utf-8', errors='replace') as file:
            lines = file.readlines()
            
        # If row selection is specified
        if start_row is not None:
            if start_row < 0:
                return "Error: start_row must be non-negative."
                
            # If only start_row is specified, read just that single row
            if end_row is None:
                if start_row >= len(lines):
                    return f"Error: start_row {start_row} is out of range (file has {len(lines)} lines)."
                content = f"Line {start_row}: {lines[start_row]}"
            else:
                # Both start_row and end_row are specified
                if end_row < start_row:
                    return "Error: end_row must be greater than or equal to start_row."
                    
                if end_row >= len(lines):
                    end_row = len(lines) - 1
                    
                selected_lines = lines[start_row:end_row+1]
                content = ""
                for i, line in enumerate(selected_lines):
                    content += f"Line {start_row + i}: {line}" if not line.endswith('\n') else f"Line {start_row + i}: {line}"
        else:
            # If no row selection, return the entire file
            content = "".join(lines)
        
        # If as_json is True, try to parse the content as JSON
        if as_json:
            try:
                import json
                # If we're showing line numbers, we cannot parse as JSON
                if start_row is not None:
                    return "Error: Cannot parse as JSON when displaying line numbers. Use as_json without row selection."
                
                # Try to parse the content as JSON
                parsed_json = json.loads(content)
                # Return pretty-printed JSON for better readability
                return json.dumps(parsed_json, indent=4, sort_keys=False, ensure_ascii=False)
            except json.JSONDecodeError as e:
                return f"Error: File content is not valid JSON. {str(e)}\n\nRaw content:\n{content}"
        
        return content
            
    except PermissionError:
        return f"Error: No permission to read file '{path}'."
    except Exception as e:
        return f"Error reading file: {str(e)}"

@mcp.tool()
async def insert_file_content(path: str, content: str, row: int = None, rows: list = None) -> str:
    """
    Insert content at specific row(s) in a file
    
    Args:
        path: Path to the file
        content: Content to insert (string or JSON object)
        row: Row number to insert at (0-based, optional)
        rows: List of row numbers to insert at (0-based, optional)
    
    Returns:
        Operation result information
    """
    try:
        # Handle different content types
        if not isinstance(content, str):
            try:
                import json
                content = json.dumps(content, indent=4, sort_keys=False, ensure_ascii=False, default=str)
            except Exception as e:
                return f"Error: Unable to convert content to JSON string: {str(e)}"
            
        # Ensure content ends with a newline if it doesn't already
        if content and not content.endswith('\n'):
            content += '\n'
            
        # Create file if it doesn't exist
        directory = os.path.dirname(os.path.abspath(path))
        if not os.path.exists(directory):
            os.makedirs(directory, exist_ok=True)
            
        if not os.path.exists(path):
            with open(path, 'w', encoding='utf-8') as file:
                pass
            
        with open(path, 'r', encoding='utf-8', errors='replace') as file:
            lines = file.readlines()
        
        # Ensure all existing lines end with newlines
        for i in range(len(lines)):
            if lines[i] and not lines[i].endswith('\n'):
                lines[i] += '\n'
        
        # Prepare lines for insertion
        content_lines = content.splitlines(True)  # Keep line endings
        
        # Handle inserting at specific rows
        if rows is not None:
            if not isinstance(rows, list):
                return "Error: 'rows' parameter must be a list of integers."
                
            # Sort rows in descending order to avoid changing indices during insertion
            rows = sorted(rows, reverse=True)
            
            for r in rows:
                if not isinstance(r, int) or r < 0:
                    return "Error: Row numbers must be non-negative integers."
                    
                if r > len(lines):
                    # If row is beyond the file, append necessary empty lines
                    lines.extend(['\n'] * (r - len(lines)))
                    lines.extend(content_lines)
                else:
                    # Insert content at each specified row
                    for line in reversed(content_lines):
                        lines.insert(r, line)
            
            # Write back to the file
            with open(path, 'w', encoding='utf-8') as file:
                file.writelines(lines)
                
            return f"Successfully inserted content at rows {rows} in '{path}'."
            
        # Handle inserting at a single row
        elif row is not None:
            if not isinstance(row, int) or row < 0:
                return "Error: Row number must be a non-negative integer."
                
            if row > len(lines):
                # If row is beyond the file, append necessary empty lines
                lines.extend(['\n'] * (row - len(lines)))
                lines.extend(content_lines)
            else:
                # Insert content at the specified row
                for line in reversed(content_lines):
                    lines.insert(row, line)
            
            # Write back to the file
            with open(path, 'w', encoding='utf-8') as file:
                file.writelines(lines)
                
            return f"Successfully inserted content at row {row} in '{path}'."
        
        # If neither row nor rows specified, append to the end
        else:
            with open(path, 'a', encoding='utf-8') as file:
                file.write(content)
            return f"Successfully appended content to '{path}'."
            
    except PermissionError:
        return f"Error: No permission to modify file '{path}'."
    except Exception as e:
        return f"Error inserting content: {str(e)}"

@mcp.tool()
async def delete_file_content(path: str, row: int = None, rows: list = None, substring: str = None) -> str:
    """
    Delete content at specific row(s) from a file
    
    Args:
        path: Path to the file
        row: Row number to delete (0-based, optional)
        rows: List of row numbers to delete (0-based, optional)
        substring: If provided, only delete this substring within the specified row(s), not the entire row (optional)
    
    Returns:
        Operation result information
    """
    try:
        if not os.path.exists(path):
            return f"Error: File '{path}' does not exist."
            
        if not os.path.isfile(path):
            return f"Error: '{path}' is not a file."
            
        with open(path, 'r', encoding='utf-8', errors='replace') as file:
            lines = file.readlines()
        
        total_lines = len(lines)
        deleted_rows = []
        modified_rows = []
        
        # Handle substring deletion (doesn't delete entire rows)
        if substring is not None:
            # For multiple rows
            if rows is not None:
                if not isinstance(rows, list):
                    return "Error: 'rows' parameter must be a list of integers."
                
                for r in rows:
                    if not isinstance(r, int) or r < 0:
                        return "Error: Row numbers must be non-negative integers."
                        
                    if r < total_lines and substring in lines[r]:
                        original_line = lines[r]
                        lines[r] = lines[r].replace(substring, '')
                        # Ensure line ends with newline if original did
                        if original_line.endswith('\n') and not lines[r].endswith('\n'):
                            lines[r] += '\n'
                        modified_rows.append(r)
            
            # For single row
            elif row is not None:
                if not isinstance(row, int) or row < 0:
                    return "Error: Row number must be a non-negative integer."
                    
                if row >= total_lines:
                    return f"Error: Row {row} is out of range (file has {total_lines} lines)."
                    
                if substring in lines[row]:
                    original_line = lines[row]
                    lines[row] = lines[row].replace(substring, '')
                    # Ensure line ends with newline if original did
                    if original_line.endswith('\n') and not lines[row].endswith('\n'):
                        lines[row] += '\n'
                    modified_rows.append(row)
            
            # For entire file
            else:
                for i in range(len(lines)):
                    if substring in lines[i]:
                        original_line = lines[i]
                        lines[i] = lines[i].replace(substring, '')
                        # Ensure line ends with newline if original did
                        if original_line.endswith('\n') and not lines[i].endswith('\n'):
                            lines[i] += '\n'
                        modified_rows.append(i)
            
            # Write back to the file
            with open(path, 'w', encoding='utf-8') as file:
                file.writelines(lines)
                
            if not modified_rows:
                return f"No occurrences of '{substring}' found in the specified rows."
            return f"Successfully removed '{substring}' from {len(modified_rows)} rows ({modified_rows}) in '{path}'."
        
        # Handle deleting multiple rows
        elif rows is not None:
            if not isinstance(rows, list):
                return "Error: 'rows' parameter must be a list of integers."
                
            # Sort rows in descending order to avoid changing indices during deletion
            rows = sorted(rows, reverse=True)
            
            for r in rows:
                if not isinstance(r, int) or r < 0:
                    return "Error: Row numbers must be non-negative integers."
                    
                if r < total_lines:
                    lines.pop(r)
                    deleted_rows.append(r)
            
            # Write back to the file
            with open(path, 'w', encoding='utf-8') as file:
                file.writelines(lines)
                
            if not deleted_rows:
                return f"No rows were within range to delete (file has {total_lines} lines)."
            return f"Successfully deleted {len(deleted_rows)} rows ({deleted_rows}) from '{path}'."
            
        # Handle deleting a single row
        elif row is not None:
            if not isinstance(row, int) or row < 0:
                return "Error: Row number must be a non-negative integer."
                
            if row >= total_lines:
                return f"Error: Row {row} is out of range (file has {total_lines} lines)."
                
            # Delete the specified row
            lines.pop(row)
            
            # Write back to the file
            with open(path, 'w', encoding='utf-8') as file:
                file.writelines(lines)
                
            return f"Successfully deleted row {row} from '{path}'."
        
        # If neither row nor rows specified, clear the file
        else:
            with open(path, 'w', encoding='utf-8') as file:
                pass
            return f"Successfully cleared all content from '{path}'."
            
    except PermissionError:
        return f"Error: No permission to modify file '{path}'."
    except Exception as e:
        return f"Error deleting content: {str(e)}"

@mcp.tool()
async def update_file_content(path: str, content: str, row: int = None, rows: list = None, substring: str = None) -> str:
    """
    Update content at specific row(s) in a file
    
    Args:
        path: Path to the file
        content: New content to place at the specified row(s)
        row: Row number to update (0-based, optional)
        rows: List of row numbers to update (0-based, optional)
        substring: If provided, only replace this substring within the specified row(s), not the entire row
    
    Returns:
        Operation result information
    """
    try:
        # Handle different content types
        if not isinstance(content, str):
            try:
                import json
                content = json.dumps(content, indent=4, sort_keys=False, ensure_ascii=False, default=str)
            except Exception as e:
                return f"Error: Unable to convert content to JSON string: {str(e)}"
        
        if not os.path.exists(path):
            return f"Error: File '{path}' does not exist."
            
        if not os.path.isfile(path):
            return f"Error: '{path}' is not a file."
            
        with open(path, 'r', encoding='utf-8', errors='replace') as file:
            lines = file.readlines()
        
        total_lines = len(lines)
        updated_rows = []
        
        # Ensure content ends with a newline if replacing a full line and doesn't already have one
        if substring is None and content and not content.endswith('\n'):
            content += '\n'
        
        # Prepare lines for update
        content_lines = content.splitlines(True) if substring is None else [content]
        
        # Handle updating multiple rows
        if rows is not None:
            if not isinstance(rows, list):
                return "Error: 'rows' parameter must be a list of integers."
                
            for r in rows:
                if not isinstance(r, int) or r < 0:
                    return "Error: Row numbers must be non-negative integers."
                    
                if r < total_lines:
                    # If substring is provided, only replace that part
                    if substring is not None:
                        # Only update if substring exists in the line
                        if substring in lines[r]:
                            original_line = lines[r]
                            lines[r] = lines[r].replace(substring, content)
                            # Ensure line ends with newline if original did
                            if original_line.endswith('\n') and not lines[r].endswith('\n'):
                                lines[r] += '\n'
                            updated_rows.append(r)
                    else:
                        # Otherwise, replace the entire line
                        # If we have multiple content lines, use them in sequence
                        if len(content_lines) > 1:
                            content_index = r % len(content_lines)
                            lines[r] = content_lines[content_index]
                        else:
                            # If we have only one content line, use it for all rows
                            lines[r] = content_lines[0] if content_lines else "\n"
                        updated_rows.append(r)
            
            # Write back to the file
            with open(path, 'w', encoding='utf-8') as file:
                file.writelines(lines)
                
            if not updated_rows:
                if substring is not None:
                    return f"No occurrences of substring '{substring}' found in the specified rows (file has {total_lines} lines)."
                else:
                    return f"No rows were within range to update (file has {total_lines} lines)."
            
            if substring is not None:
                return f"Successfully updated substring in {len(updated_rows)} rows ({updated_rows}) in '{path}'."
            else:
                return f"Successfully updated {len(updated_rows)} rows ({updated_rows}) in '{path}'."
            
        # Handle updating a single row
        elif row is not None:
            if not isinstance(row, int) or row < 0:
                return "Error: Row number must be a non-negative integer."
                
            if row >= total_lines:
                return f"Error: Row {row} is out of range (file has {total_lines} lines)."
                
            # If substring is provided, only replace that part
            if substring is not None:
                # Only update if substring exists in the line
                if substring in lines[row]:
                    original_line = lines[row]
                    lines[row] = lines[row].replace(substring, content)
                    # Ensure line ends with newline if original did
                    if original_line.endswith('\n') and not lines[row].endswith('\n'):
                        lines[row] += '\n'
                else:
                    return f"Substring '{substring}' not found in row {row}."
            else:
                # Otherwise, replace the entire line
                lines[row] = content_lines[0] if content_lines else "\n"
            
            # Write back to the file
            with open(path, 'w', encoding='utf-8') as file:
                file.writelines(lines)
                
            if substring is not None:
                return f"Successfully updated substring in row {row} in '{path}'."
            else:
                return f"Successfully updated row {row} in '{path}'."
        
        # If neither row nor rows specified, update the entire file
        else:
            if substring is not None:
                # Replace substring throughout the file
                updated_count = 0
                for i in range(len(lines)):
                    if substring in lines[i]:
                        original_line = lines[i]
                        lines[i] = lines[i].replace(substring, content)
                        # Ensure line ends with newline if original did
                        if original_line.endswith('\n') and not lines[i].endswith('\n'):
                            lines[i] += '\n'
                        updated_count += 1
                
                with open(path, 'w', encoding='utf-8') as file:
                    file.writelines(lines)
                
                if updated_count == 0:
                    return f"Substring '{substring}' not found in any line of '{path}'."
                return f"Successfully updated substring in {updated_count} lines in '{path}'."
            else:
                # Replace entire file content
                with open(path, 'w', encoding='utf-8') as file:
                    file.write(content)
                return f"Successfully updated all content in '{path}'."
            
    except PermissionError:
        return f"Error: No permission to modify file '{path}'."
    except Exception as e:
        return f"Error updating content: {str(e)}"

def main():
    """
    Entry point function that runs the MCP server.
    """
    print("Starting Terminal Controller MCP Server...", file=sys.stderr)
    mcp.run(transport='stdio')

# Make the module callable
def __call__():
    """
    Make the module callable for uvx.
    This function is called when the module is executed directly.
    """
    print("Terminal Controller MCP Server starting via __call__...", file=sys.stderr)
    mcp.run(transport='stdio')

# Add this for compatibility with uvx
sys.modules[__name__].__call__ = __call__

# Run the server when the script is executed directly
if __name__ == "__main__":
    main()