diff --git a/stubs/redis/redis/client.pyi b/stubs/redis/redis/client.pyi
index 279db79d0..77291e7f8 100644
--- a/stubs/redis/redis/client.pyi
+++ b/stubs/redis/redis/client.pyi
@@ -5,7 +5,7 @@ from typing import Any, Callable, ClassVar, Generic, Iterable, Iterator, Mapping
 from typing_extensions import Literal
 
 from .commands import CoreCommands, RedisModuleCommands, SentinelCommands
-from .connection import ConnectionPool
+from .connection import ConnectionPool, _ConnectionPoolOptions
 from .lock import Lock
 from .retry import Retry
 
@@ -19,6 +19,11 @@ _VT = TypeVar("_VT")
 _T = TypeVar("_T")
 _ScoreCastFuncReturn = TypeVar("_ScoreCastFuncReturn")
 
+# Keyword arguments that are passed to Redis.parse_response().
+_ParseResponseOptions = Any
+# Keyword arguments that are passed to Redis.execute_command().
+_CommandOptions = _ConnectionPoolOptions | _ParseResponseOptions
+
 SYM_EMPTY: bytes
 EMPTY_RESPONSE: str
 NEVER_DECODE: str
@@ -275,10 +280,9 @@ class Redis(RedisModuleCommands, CoreCommands[_StrType], SentinelCommands, Gener
         thread_local: bool = ...,
     ) -> _LockType: ...
     def pubsub(self, *, shard_hint: Any = ..., ignore_subscribe_messages: bool = ...) -> PubSub: ...
-    def execute_command(self, *args, **options): ...
-    def parse_response(self, connection, command_name, **options): ...
+    def execute_command(self, *args, **options: _CommandOptions): ...
+    def parse_response(self, connection, command_name, **options: _ParseResponseOptions): ...
     def monitor(self) -> Monitor: ...
-    def cluster(self, cluster_arg: str, *args: Any) -> Any: ...
     def __enter__(self) -> Redis[_StrType]: ...
     def __exit__(self, exc_type, exc_value, traceback): ...
     def __del__(self) -> None: ...
diff --git a/stubs/redis/redis/commands/core.pyi b/stubs/redis/redis/commands/core.pyi
index bcd6cbefa..4e6bd0803 100644
--- a/stubs/redis/redis/commands/core.pyi
+++ b/stubs/redis/redis/commands/core.pyi
@@ -4,22 +4,22 @@ from datetime import datetime, timedelta
 from typing import Any, Generic, TypeVar, Union, overload
 from typing_extensions import Literal
 
-from ..client import _Key, _Value
+from ..client import _CommandOptions, _Key, _Value
 
 _ScoreCastFuncReturn = TypeVar("_ScoreCastFuncReturn")
 _StrType = TypeVar("_StrType", bound=Union[str, bytes])
 
 class ACLCommands(Generic[_StrType]):
-    def acl_cat(self, category: str | None = ..., **kwargs) -> list[str]: ...
-    def acl_deluser(self, *username: str, **kwargs) -> int: ...
-    def acl_genpass(self, bits: int | None = ..., **kwargs) -> str: ...
-    def acl_getuser(self, username: str, **kwargs) -> Any | None: ...
-    def acl_help(self, **kwargs): ...
-    def acl_list(self, **kwargs) -> list[str]: ...
-    def acl_log(self, count: int | None = ..., **kwargs): ...
-    def acl_log_reset(self, **kwargs): ...
-    def acl_load(self, **kwargs) -> bool: ...
-    def acl_save(self, **kwargs): ...
+    def acl_cat(self, category: str | None = ..., **kwargs: _CommandOptions) -> list[str]: ...
+    def acl_deluser(self, *username: str, **kwargs: _CommandOptions) -> int: ...
+    def acl_genpass(self, bits: int | None = ..., **kwargs: _CommandOptions) -> str: ...
+    def acl_getuser(self, username: str, **kwargs: _CommandOptions) -> Any | None: ...
+    def acl_help(self, **kwargs: _CommandOptions): ...
+    def acl_list(self, **kwargs: _CommandOptions) -> list[str]: ...
+    def acl_log(self, count: int | None = ..., **kwargs: _CommandOptions): ...
+    def acl_log_reset(self, **kwargs: _CommandOptions): ...
+    def acl_load(self, **kwargs: _CommandOptions) -> bool: ...
+    def acl_save(self, **kwargs: _CommandOptions): ...
     def acl_setuser(
         self,
         username: str,
@@ -33,16 +33,16 @@ class ACLCommands(Generic[_StrType]):
         reset: bool = ...,
         reset_keys: bool = ...,
         reset_passwords: bool = ...,
-        **kwargs,
+        **kwargs: _CommandOptions,
     ) -> bool: ...
-    def acl_users(self, **kwargs) -> list[str]: ...
-    def acl_whoami(self, **kwargs) -> str: ...
+    def acl_users(self, **kwargs: _CommandOptions) -> list[str]: ...
+    def acl_whoami(self, **kwargs: _CommandOptions) -> str: ...
 
 class ManagementCommands:
-    def bgrewriteaof(self, **kwargs): ...
-    def bgsave(self, schedule: bool = ..., **kwargs): ...
+    def bgrewriteaof(self, **kwargs: _CommandOptions): ...
+    def bgsave(self, schedule: bool = ..., **kwargs: _CommandOptions): ...
     def role(self): ...
-    def client_kill(self, address: str, **kwargs) -> bool: ...
+    def client_kill(self, address: str, **kwargs: _CommandOptions) -> bool: ...
     def client_kill_filter(
         self,
         _id: Any | None = ...,
@@ -51,14 +51,16 @@ class ManagementCommands:
         skipme: Any | None = ...,
         laddr: Any | None = ...,
         user: Any | None = ...,
-        **kwargs,
+        **kwargs: _CommandOptions,
     ): ...
-    def client_info(self, **kwargs): ...
-    def client_list(self, _type: str | None = ..., client_id: list[str] = ..., **kwargs) -> list[dict[str, str]]: ...
-    def client_getname(self, **kwargs) -> str | None: ...
-    def client_getredir(self, **kwargs): ...
-    def client_reply(self, reply, **kwargs): ...
-    def client_id(self, **kwargs) -> int: ...
+    def client_info(self, **kwargs: _CommandOptions): ...
+    def client_list(
+        self, _type: str | None = ..., client_id: list[str] = ..., **kwargs: _CommandOptions
+    ) -> list[dict[str, str]]: ...
+    def client_getname(self, **kwargs: _CommandOptions) -> str | None: ...
+    def client_getredir(self, **kwargs: _CommandOptions): ...
+    def client_reply(self, reply, **kwargs: _CommandOptions): ...
+    def client_id(self, **kwargs: _CommandOptions) -> int: ...
     def client_tracking_on(
         self, clientid: Any | None = ..., prefix=..., bcast: bool = ..., optin: bool = ..., optout: bool = ..., noloop: bool = ...
     ): ...
@@ -74,55 +76,64 @@ class ManagementCommands:
         optin: bool = ...,
         optout: bool = ...,
         noloop: bool = ...,
-        **kwargs,
+        **kwargs: _CommandOptions,
     ): ...
-    def client_trackinginfo(self, **kwargs): ...
-    def client_setname(self, name: str, **kwargs) -> bool: ...
-    def client_unblock(self, client_id, error: bool = ..., **kwargs): ...
-    def client_pause(self, timeout, all: bool = ..., **kwargs): ...
-    def client_unpause(self, **kwargs): ...
-    def command(self, **kwargs): ...
-    def command_info(self, **kwargs): ...
-    def command_count(self, **kwargs): ...
-    def config_get(self, pattern: str = ..., **kwargs): ...
-    def config_set(self, name, value, **kwargs): ...
-    def config_resetstat(self, **kwargs): ...
-    def config_rewrite(self, **kwargs): ...
-    def dbsize(self, **kwargs) -> int: ...
-    def debug_object(self, key, **kwargs): ...
-    def debug_segfault(self, **kwargs): ...
-    def echo(self, value: _Value, **kwargs) -> bytes: ...
-    def flushall(self, asynchronous: bool = ..., **kwargs) -> bool: ...
-    def flushdb(self, asynchronous: bool = ..., **kwargs) -> bool: ...
+    def client_trackinginfo(self, **kwargs: _CommandOptions): ...
+    def client_setname(self, name: str, **kwargs: _CommandOptions) -> bool: ...
+    def client_unblock(self, client_id, error: bool = ..., **kwargs: _CommandOptions): ...
+    def client_pause(self, timeout, all: bool = ..., **kwargs: _CommandOptions): ...
+    def client_unpause(self, **kwargs: _CommandOptions): ...
+    def command(self, **kwargs: _CommandOptions): ...
+    def command_info(self, **kwargs: _CommandOptions): ...
+    def command_count(self, **kwargs: _CommandOptions): ...
+    def config_get(self, pattern: str = ..., **kwargs: _CommandOptions): ...
+    def config_set(self, name, value, **kwargs: _CommandOptions): ...
+    def config_resetstat(self, **kwargs: _CommandOptions): ...
+    def config_rewrite(self, **kwargs: _CommandOptions): ...
+    def dbsize(self, **kwargs: _CommandOptions) -> int: ...
+    def debug_object(self, key, **kwargs: _CommandOptions): ...
+    def debug_segfault(self, **kwargs: _CommandOptions): ...
+    def echo(self, value: _Value, **kwargs: _CommandOptions) -> bytes: ...
+    def flushall(self, asynchronous: bool = ..., **kwargs: _CommandOptions) -> bool: ...
+    def flushdb(self, asynchronous: bool = ..., **kwargs: _CommandOptions) -> bool: ...
     def sync(self): ...
     def psync(self, replicationid, offset): ...
-    def swapdb(self, first, second, **kwargs): ...
-    def select(self, index, **kwargs): ...
-    def info(self, section: _Key | None = ..., **kwargs) -> Mapping[str, Any]: ...
-    def lastsave(self, **kwargs): ...
-    def lolwut(self, *version_numbers, **kwargs): ...
+    def swapdb(self, first, second, **kwargs: _CommandOptions): ...
+    def select(self, index, **kwargs: _CommandOptions): ...
+    def info(self, section: _Key | None = ..., **kwargs: _CommandOptions) -> Mapping[str, Any]: ...
+    def lastsave(self, **kwargs: _CommandOptions): ...
+    def lolwut(self, *version_numbers, **kwargs: _CommandOptions): ...
     def reset(self) -> None: ...
     def migrate(
-        self, host, port, keys, destination_db, timeout, copy: bool = ..., replace: bool = ..., auth: Any | None = ..., **kwargs
+        self,
+        host,
+        port,
+        keys,
+        destination_db,
+        timeout,
+        copy: bool = ...,
+        replace: bool = ...,
+        auth: Any | None = ...,
+        **kwargs: _CommandOptions,
     ): ...
-    def object(self, infotype, key, **kwargs): ...
-    def memory_doctor(self, **kwargs): ...
-    def memory_help(self, **kwargs): ...
-    def memory_stats(self, **kwargs) -> dict[str, Any]: ...
-    def memory_malloc_stats(self, **kwargs): ...
-    def memory_usage(self, key, samples: Any | None = ..., **kwargs): ...
-    def memory_purge(self, **kwargs): ...
-    def ping(self, **kwargs) -> bool: ...
-    def quit(self, **kwargs): ...
-    def replicaof(self, *args, **kwargs): ...
-    def save(self, **kwargs) -> bool: ...
-    def shutdown(self, save: bool = ..., nosave: bool = ..., **kwargs) -> None: ...
-    def slaveof(self, host: Any | None = ..., port: Any | None = ..., **kwargs): ...
-    def slowlog_get(self, num: Any | None = ..., **kwargs): ...
-    def slowlog_len(self, **kwargs): ...
-    def slowlog_reset(self, **kwargs): ...
-    def time(self, **kwargs): ...
-    def wait(self, num_replicas, timeout, **kwargs): ...
+    def object(self, infotype, key, **kwargs: _CommandOptions): ...
+    def memory_doctor(self, **kwargs: _CommandOptions): ...
+    def memory_help(self, **kwargs: _CommandOptions): ...
+    def memory_stats(self, **kwargs: _CommandOptions) -> dict[str, Any]: ...
+    def memory_malloc_stats(self, **kwargs: _CommandOptions): ...
+    def memory_usage(self, key, samples: Any | None = ..., **kwargs: _CommandOptions): ...
+    def memory_purge(self, **kwargs: _CommandOptions): ...
+    def ping(self, **kwargs: _CommandOptions) -> bool: ...
+    def quit(self, **kwargs: _CommandOptions): ...
+    def replicaof(self, *args, **kwargs: _CommandOptions): ...
+    def save(self, **kwargs: _CommandOptions) -> bool: ...
+    def shutdown(self, save: bool = ..., nosave: bool = ..., **kwargs: _CommandOptions) -> None: ...
+    def slaveof(self, host: Any | None = ..., port: Any | None = ..., **kwargs: _CommandOptions): ...
+    def slowlog_get(self, num: Any | None = ..., **kwargs: _CommandOptions): ...
+    def slowlog_len(self, **kwargs: _CommandOptions): ...
+    def slowlog_reset(self, **kwargs: _CommandOptions): ...
+    def time(self, **kwargs: _CommandOptions): ...
+    def wait(self, num_replicas, timeout, **kwargs: _CommandOptions): ...
 
 class BasicKeyCommands(Generic[_StrType]):
     def append(self, key, value): ...
@@ -158,7 +169,7 @@ class BasicKeyCommands(Generic[_StrType]):
     def incr(self, name: _Key, amount: int = ...) -> int: ...
     def incrby(self, name: _Key, amount: int = ...) -> int: ...
     def incrbyfloat(self, name: _Key, amount: float = ...) -> float: ...
-    def keys(self, pattern: _Key = ..., **kwargs) -> list[_StrType]: ...
+    def keys(self, pattern: _Key = ..., **kwargs: _CommandOptions) -> list[_StrType]: ...
     def lmove(
         self, first_list: _Key, second_list: _Key, src: Literal["LEFT", "RIGHT"] = ..., dest: Literal["LEFT", "RIGHT"] = ...
     ) -> _Value: ...
@@ -180,7 +191,7 @@ class BasicKeyCommands(Generic[_StrType]):
     def psetex(self, name, time_ms, value): ...
     def pttl(self, name): ...
     def hrandfield(self, key, count: Any | None = ..., withvalues: bool = ...): ...
-    def randomkey(self, **kwargs): ...
+    def randomkey(self, **kwargs: _CommandOptions): ...
     def rename(self, src, dst): ...
     def renamenx(self, src, dst): ...
     def restore(
@@ -214,7 +225,7 @@ class BasicKeyCommands(Generic[_StrType]):
         idx: bool = ...,
         minmatchlen: Any | None = ...,
         withmatchlen: bool = ...,
-        **kwargs,
+        **kwargs: _CommandOptions,
     ): ...
     def strlen(self, name): ...
     def substr(self, name, start, end: int = ...): ...
@@ -295,10 +306,15 @@ class ListCommands(Generic[_StrType]):
 
 class ScanCommands(Generic[_StrType]):
     def scan(
-        self, cursor: int = ..., match: _Key | None = ..., count: int | None = ..., _type: str | None = ..., **kwargs
+        self,
+        cursor: int = ...,
+        match: _Key | None = ...,
+        count: int | None = ...,
+        _type: str | None = ...,
+        **kwargs: _CommandOptions,
     ) -> tuple[int, list[_StrType]]: ...
     def scan_iter(
-        self, match: _Key | None = ..., count: int | None = ..., _type: str | None = ..., **kwargs
+        self, match: _Key | None = ..., count: int | None = ..., _type: str | None = ..., **kwargs: _CommandOptions
     ) -> Iterator[_StrType]: ...
     def sscan(
         self, name: _Key, cursor: int = ..., match: _Key | None = ..., count: int | None = ...
@@ -588,10 +604,10 @@ class HashCommands(Generic[_StrType]):
     def hstrlen(self, name, key): ...
 
 class PubSubCommands:
-    def publish(self, channel: _Key, message: _Key, **kwargs) -> int: ...
-    def pubsub_channels(self, pattern: _Key = ..., **kwargs) -> list[str]: ...
-    def pubsub_numpat(self, **kwargs) -> int: ...
-    def pubsub_numsub(self, *args: _Key, **kwargs) -> list[tuple[str, int]]: ...
+    def publish(self, channel: _Key, message: _Key, **kwargs: _CommandOptions) -> int: ...
+    def pubsub_channels(self, pattern: _Key = ..., **kwargs: _CommandOptions) -> list[str]: ...
+    def pubsub_numpat(self, **kwargs: _CommandOptions) -> int: ...
+    def pubsub_numsub(self, *args: _Key, **kwargs: _CommandOptions) -> list[tuple[str, int]]: ...
 
 class ScriptCommands(Generic[_StrType]):
     def eval(self, script, numkeys, *keys_and_args): ...
@@ -698,8 +714,9 @@ class BitFieldOperation:
     def execute(self): ...
 
 class ClusterCommands:
-    def readwrite(self, **kwargs) -> bool: ...
-    def readonly(self, **kwargs) -> bool: ...
+    def cluster(self, cluster_arg: str, *args, **kwargs: _CommandOptions): ...
+    def readwrite(self, **kwargs: _CommandOptions) -> bool: ...
+    def readonly(self, **kwargs: _CommandOptions) -> bool: ...
 
 class DataAccessCommands(
     BasicKeyCommands[_StrType],
diff --git a/stubs/redis/redis/connection.pyi b/stubs/redis/redis/connection.pyi
index 3e0b922d3..7a4f79a08 100644
--- a/stubs/redis/redis/connection.pyi
+++ b/stubs/redis/redis/connection.pyi
@@ -9,6 +9,9 @@ SYM_CRLF: Any
 SYM_EMPTY: Any
 SERVER_CLOSED_CONNECTION_ERROR: Any
 
+# Options as passed to Pool.get_connection().
+_ConnectionPoolOptions = Any
+
 class BaseParser:
     EXCEPTION_CLASSES: Any
     def parse_error(self, response): ...
@@ -176,7 +179,7 @@ class ConnectionPool:
     def __init__(self, connection_class=..., max_connections=..., **connection_kwargs) -> None: ...
     pid: Any
     def reset(self): ...
-    def get_connection(self, command_name, *keys, **options): ...
+    def get_connection(self, command_name, *keys, **options: _ConnectionPoolOptions): ...
     def make_connection(self): ...
     def release(self, connection): ...
     def disconnect(self, inuse_connections: bool = ...): ...
@@ -191,7 +194,7 @@ class BlockingConnectionPool(ConnectionPool):
     pool: Any
     def reset(self): ...
     def make_connection(self): ...
-    def get_connection(self, command_name, *keys, **options): ...
+    def get_connection(self, command_name, *keys, **options: _ConnectionPoolOptions): ...
     def release(self, connection): ...
     def disconnect(self): ...