Shell MCP Server

Apache 2.0
OverviewInspectSchema Related Servers Reviews Score
src
shell_mcp_server
"""
Shell MCP Server
==============

This module implements an MCP server that provides secure shell command execution
within specified working directories using specified shells.

Key Features:
- Safe command execution in specified directories
- Supports multiple shell types (bash, sh, cmd, powershell)
- Command timeout handling
- Cross-platform support
- Robust error handling
"""

import asyncio
import os
import sys
import argparse
from typing import Dict, Any, List
import mcp
import mcp.types as types
from mcp.server import Server, InitializationOptions, NotificationOptions
from .config import Settings


# Parse command line arguments for directories and shells
def parse_args() -> tuple[List[str], Dict[str, str]]:
    """Parse command line arguments for directories and shells."""
    parser = argparse.ArgumentParser(description="Shell MCP Server")
    parser.add_argument('directories', nargs='+', help='Allowed directories for command execution')
    parser.add_argument('--shell', action='append', nargs=2, metavar=('name', 'path'),
                       help='Shell specification in format: name path')
    
    args = parser.parse_args()
    
    # Convert shell arguments to dictionary
    shells = {}
    if args.shell:
        shells = {name: path for name, path in args.shell}
    
    # Default to system shell if none specified
    if not shells:
        if sys.platform == 'win32':
            shells = {'cmd': 'cmd.exe', 'powershell': 'powershell.exe'}
        else:
            shells = {'bash': '/bin/bash', 'sh': '/bin/sh'}
    
    return args.directories, shells


# Initialize server settings and create server instance - skip arg parsing for tests
if 'pytest' not in sys.modules:
    directories, shells = parse_args()
    settings = Settings(directories=directories, shells=shells)
else:
    settings = Settings(directories=['/tmp'], shells={'bash': '/bin/bash'})

server = Server(settings.APP_NAME)


async def run_shell_command(shell: str, command: str, cwd: str) -> Dict[str, Any]:
    """
    Execute a shell command safely and return its output.
    
    Args:
        shell (str): Name of the shell to use
        command (str): The command to execute
        cwd (str): Working directory for command execution
        
    Returns:
        Dict[str, Any]: Command execution results including stdout, stderr, and exit code
    """
    if not settings.is_path_allowed(cwd):
        raise ValueError(f"Directory '{cwd}' is not in the allowed directories list")
    
    if shell not in settings.ALLOWED_SHELLS:
        raise ValueError(f"Shell '{shell}' is not allowed. Available shells: {list(settings.ALLOWED_SHELLS.keys())}")
    
    shell_path = settings.ALLOWED_SHELLS[shell]
    
    try:
        if sys.platform == 'win32':
            shell_cmd = [shell_path, '/c', command] if shell == 'cmd' else [shell_path, '-Command', command]
        else:
            shell_cmd = [shell_path, '-c', command]

        process = await asyncio.create_subprocess_exec(
            *shell_cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=cwd
        )

        try:
            stdout, stderr = await asyncio.wait_for(
                process.communicate(), 
                timeout=settings.COMMAND_TIMEOUT
            )
            
            return {
                "stdout": stdout.decode() if stdout else "",
                "stderr": stderr.decode() if stderr else "",
                "exit_code": process.returncode,
                "command": command,
                "shell": shell,
                "cwd": cwd
            }

        except asyncio.TimeoutError:
            try:
                process.kill()
                await process.wait()
            except ProcessLookupError:
                pass
            raise TimeoutError(f"Command execution timed out after {settings.COMMAND_TIMEOUT} seconds")

    except TimeoutError:
        raise
    except Exception as e:
        return {
            "stdout": "",
            "stderr": str(e),
            "exit_code": -1,
            "command": command,
            "shell": shell,
            "cwd": cwd
        }


@server.list_tools()
async def list_tools() -> List[types.Tool]:
    """List available shell tools."""
    return [
        types.Tool(
            name="execute_command",
            description="Execute a shell command in a specified directory using a specified shell",
            inputSchema={
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "The shell command to execute",
                    },
                    "shell": {
                        "type": "string",
                        "description": f"Shell to use for execution. Available: {list(settings.ALLOWED_SHELLS.keys())}",
                    },
                    "cwd": {
                        "type": "string",
                        "description": "Working directory for command execution",
                    },
                },
                "required": ["command", "shell", "cwd"],
            },
        )
    ]


@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[types.TextContent]:
    """
    Handle tool calls for shell command execution.
    
    Args:
        name (str): The name of the tool to call (must be 'execute_command')
        arguments (Dict[str, Any]): Tool arguments including 'command', 'shell', and 'cwd'
        
    Returns:
        List[types.TextContent]: The command execution results or error message
    """
    if name != "execute_command":
        return [types.TextContent(type="text", text=f"Error: Unknown tool {name}")]

    command = arguments["command"]
    shell = arguments["shell"]
    cwd = arguments["cwd"]

    try:
        result = await run_shell_command(shell, command, cwd)
        return [types.TextContent(type="text", text=str(result))]
    except Exception as e:
        return [types.TextContent(type="text", text=f"Error: {str(e)}")]


async def main():
    """Main entry point for the shell MCP server."""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name=settings.APP_NAME,
                server_version=settings.APP_VERSION,
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )