execute_command
Run shell commands in a specified directory using a chosen shell (bash, sh) with the Shell MCP Server, ensuring secure and isolated execution environments.
Instructions
Execute a shell command in a specified directory using a specified shell
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | The shell command to execute | |
| cwd | Yes | Working directory for command execution | |
| shell | Yes | Shell to use for execution. Available: ['bash', 'sh'] |
Implementation Reference
- src/shell_mcp_server/server.py:62-129 (handler)Core handler function that executes the shell command using asyncio subprocess, performs security checks on cwd and shell, handles timeouts, and returns stdout/stderr/exit_code.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 }
- src/shell_mcp_server/server.py:131-157 (registration)Registers the 'execute_command' tool with the MCP server, including its description and input schema.@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"], }, ) ]
- src/shell_mcp_server/server.py:160-184 (handler)MCP server tool call handler that validates the tool name 'execute_command', extracts arguments, calls the run_shell_command function, and returns the result as text content.@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)}")]
- Input schema definition for the 'execute_command' tool, specifying required string parameters for command, shell, and working directory.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"], },