diff --git a/redisvl/index/index.py b/redisvl/index/index.py
index 25b907f8..2187c759 100644
--- a/redisvl/index/index.py
+++ b/redisvl/index/index.py
@@ -497,6 +497,20 @@ def clear(self) -> int:
 
         return total_records_deleted
 
+    def drop_keys(self, keys: Union[str, List[str]]) -> int:
+        """Remove a specific entry or entries from the index by it's key ID.
+
+        Args:
+            keys (Union[str, List[str]]): The document ID or IDs to remove from the index.
+
+        Returns:
+            int: Count of records deleted from Redis.
+        """
+        if isinstance(keys, List):
+            return self._redis_client.delete(*keys)  # type: ignore
+        else:
+            return self._redis_client.delete(keys)  # type: ignore
+
     def load(
         self,
         data: Iterable[Any],
@@ -935,6 +949,20 @@ async def clear(self) -> int:
 
         return total_records_deleted
 
+    async def drop_keys(self, keys: Union[str, List[str]]) -> int:
+        """Remove a specific entry or entries from the index by it's key ID.
+
+        Args:
+            keys (Union[str, List[str]]): The document ID or IDs to remove from the index.
+
+        Returns:
+            int: Count of records deleted from Redis.
+        """
+        if isinstance(keys, List):
+            return await self._redis_client.delete(*keys)  # type: ignore
+        else:
+            return await self._redis_client.delete(keys)  # type: ignore
+
     async def load(
         self,
         data: Iterable[Any],
diff --git a/tests/integration/test_async_search_index.py b/tests/integration/test_async_search_index.py
index 41e5a03a..5e795391 100644
--- a/tests/integration/test_async_search_index.py
+++ b/tests/integration/test_async_search_index.py
@@ -184,6 +184,39 @@ async def test_search_index_clear(async_client, async_index):
     assert await async_index.exists()
 
 
+@pytest.mark.asyncio
+async def test_search_index_drop_key(async_client, async_index):
+    async_index.set_client(async_client)
+    await async_index.create(overwrite=True, drop=True)
+    data = [{"id": "1", "test": "foo"}, {"id": "2", "test": "bar"}]
+    keys = await async_index.load(data, id_field="id")
+
+    dropped = await async_index.drop_keys(keys[0])
+    assert dropped == 1
+    assert not await async_index.fetch(keys[0])
+    assert await async_index.fetch(keys[1]) is not None
+
+
+@pytest.mark.asyncio
+async def test_search_index_drop_keys(async_client, async_index):
+    async_index.set_client(async_client)
+    await async_index.create(overwrite=True, drop=True)
+    data = [
+        {"id": "1", "test": "foo"},
+        {"id": "2", "test": "bar"},
+        {"id": "3", "test": "baz"},
+    ]
+    keys = await async_index.load(data, id_field="id")
+
+    dropped = await async_index.drop_keys(keys[0:2])
+    assert dropped == 2
+    assert not await async_index.fetch(keys[0])
+    assert not await async_index.fetch(keys[1])
+    assert await async_index.fetch(keys[2]) is not None
+
+    assert await async_index.exists()
+
+
 @pytest.mark.asyncio
 async def test_search_index_load_and_fetch(async_client, async_index):
     async_index.set_client(async_client)
diff --git a/tests/integration/test_search_index.py b/tests/integration/test_search_index.py
index 1c215dc9..574cec91 100644
--- a/tests/integration/test_search_index.py
+++ b/tests/integration/test_search_index.py
@@ -170,6 +170,39 @@ def test_search_index_clear(client, index):
     assert index.exists()
 
 
+def test_search_index_drop_key(client, index):
+    index.set_client(client)
+    index.create(overwrite=True, drop=True)
+    data = [{"id": "1", "test": "foo"}, {"id": "2", "test": "bar"}]
+    keys = index.load(data, id_field="id")
+
+    # test passing a single string key removes only that key
+    dropped = index.drop_keys(keys[0])
+    assert dropped == 1
+    assert not index.fetch(keys[0])
+    assert index.fetch(keys[1]) is not None  # still have all other entries
+
+
+def test_search_index_drop_keys(client, index):
+    index.set_client(client)
+    index.create(overwrite=True, drop=True)
+    data = [
+        {"id": "1", "test": "foo"},
+        {"id": "2", "test": "bar"},
+        {"id": "3", "test": "baz"},
+    ]
+    keys = index.load(data, id_field="id")
+
+    # test passing a list of keys selectively removes only those keys
+    dropped = index.drop_keys(keys[0:2])
+    assert dropped == 2
+    assert not index.fetch(keys[0])
+    assert not index.fetch(keys[1])
+    assert index.fetch(keys[2]) is not None
+
+    assert index.exists()
+
+
 def test_search_index_load_and_fetch(client, index):
     index.set_client(client)
     index.create(overwrite=True, drop=True)