diff --git a/redis/commands/core.py b/redis/commands/core.py
index 8bbcda3a69..545bd38864 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -1109,12 +1109,25 @@ def save(self, **kwargs) -> ResponseT:
         """
         return self.execute_command("SAVE", **kwargs)
 
-    def shutdown(self, save: bool = False, nosave: bool = False, **kwargs) -> None:
+    def shutdown(
+        self,
+        save: bool = False,
+        nosave: bool = False,
+        now: bool = False,
+        force: bool = False,
+        abort: bool = False,
+        **kwargs,
+    ) -> None:
         """Shutdown the Redis server.  If Redis has persistence configured,
-        data will be flushed before shutdown.  If the "save" option is set,
-        a data flush will be attempted even if there is no persistence
-        configured.  If the "nosave" option is set, no data flush will be
-        attempted.  The "save" and "nosave" options cannot both be set.
+        data will be flushed before shutdown.
+        It is possible to specify modifiers to alter the behavior of the command:
+        ``save`` will force a DB saving operation even if no save points are configured.
+        ``nosave`` will prevent a DB saving operation even if one or more save points
+        are configured.
+        ``now`` skips waiting for lagging replicas, i.e. it bypasses the first step in
+        the shutdown sequence.
+        ``force`` ignores any errors that would normally prevent the server from exiting
+        ``abort`` cancels an ongoing shutdown and cannot be combined with other flags.
 
         For more information see https://redis.io/commands/shutdown
         """
@@ -1125,6 +1138,12 @@ def shutdown(self, save: bool = False, nosave: bool = False, **kwargs) -> None:
             args.append("SAVE")
         if nosave:
             args.append("NOSAVE")
+        if now:
+            args.append("NOW")
+        if force:
+            args.append("FORCE")
+        if abort:
+            args.append("ABORT")
         try:
             self.execute_command(*args, **kwargs)
         except ConnectionError:
@@ -1235,7 +1254,13 @@ async def memory_help(self, **kwargs) -> None:
         return super().memory_help(**kwargs)
 
     async def shutdown(
-        self, save: bool = False, nosave: bool = False, **kwargs
+        self,
+        save: bool = False,
+        nosave: bool = False,
+        now: bool = False,
+        force: bool = False,
+        abort: bool = False,
+        **kwargs,
     ) -> None:
         """Shutdown the Redis server.  If Redis has persistence configured,
         data will be flushed before shutdown.  If the "save" option is set,
@@ -1252,6 +1277,12 @@ async def shutdown(
             args.append("SAVE")
         if nosave:
             args.append("NOSAVE")
+        if now:
+            args.append("NOW")
+        if force:
+            args.append("FORCE")
+        if abort:
+            args.append("ABORT")
         try:
             await self.execute_command(*args, **kwargs)
         except ConnectionError:
diff --git a/tests/test_commands.py b/tests/test_commands.py
index 59754123ac..92a63c9d9e 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -3,6 +3,7 @@
 import re
 import time
 from string import ascii_letters
+from unittest import mock
 
 import pytest
 
@@ -4633,6 +4634,19 @@ def test_replicaof(self, r):
             assert r.replicaof("NO ONE")
         assert r.replicaof("NO", "ONE")
 
+    def test_shutdown(self, r: redis.Redis):
+        r.execute_command = mock.MagicMock()
+        r.execute_command("SHUTDOWN", "NOSAVE")
+        r.execute_command.assert_called_once_with("SHUTDOWN", "NOSAVE")
+
+    @skip_if_server_version_lt("7.0.0")
+    def test_shutdown_with_params(self, r: redis.Redis):
+        r.execute_command = mock.MagicMock()
+        r.execute_command("SHUTDOWN", "SAVE", "NOW", "FORCE")
+        r.execute_command.assert_called_once_with("SHUTDOWN", "SAVE", "NOW", "FORCE")
+        r.execute_command("SHUTDOWN", "ABORT")
+        r.execute_command.assert_called_with("SHUTDOWN", "ABORT")
+
     @pytest.mark.replica
     @skip_if_server_version_lt("2.8.0")
     @skip_if_redis_enterprise()