diff --git a/python/packages/redis/agent_framework_redis/_history_provider.py b/python/packages/redis/agent_framework_redis/_history_provider.py
index 07827ac62aa..06f34e7c9f5 100644
--- a/python/packages/redis/agent_framework_redis/_history_provider.py
+++ b/python/packages/redis/agent_framework_redis/_history_provider.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import warnings
 from collections.abc import Awaitable, Sequence
 from inspect import isawaitable
@@ -23,6 +24,8 @@
 
 from ._feature_usage import FeatureIndex
 
+logger = logging.getLogger(__name__)
+
 _T = TypeVar("_T")
 
 
@@ -271,7 +274,18 @@ async def get_messages(
         redis_messages = cast("list[str]", await _redis_result(cast("Any", self._redis_client).lrange(key, 0, -1)))
         messages: list[Message] = []
         for serialized in redis_messages:
-            messages.append(Message.from_dict(self._deserialize_json(serialized)))
+            try:
+                payload = self._deserialize_json(serialized)
+                if not isinstance(payload, dict):
+                    raise ValueError(f"Expected a JSON object, got {type(payload).__name__}")
+                messages.append(Message.from_dict(payload))
+            except (ValueError, TypeError, AttributeError, json.JSONDecodeError) as exc:
+                # Log the exception class, not its message: ``Message.from_dict``
+                # embeds the payload's ``type`` value in mismatch errors, so logging
+                # ``str(exc)`` could copy untrusted history data into application logs.
+                logger.warning(
+                    "Skipping malformed Redis history entry (%s)", type(exc).__name__
+                )
         return messages
 
     async def save_messages(
diff --git a/python/packages/redis/tests/test_providers.py b/python/packages/redis/tests/test_providers.py
index c8aba3a2e55..ba4e5356033 100644
--- a/python/packages/redis/tests/test_providers.py
+++ b/python/packages/redis/tests/test_providers.py
@@ -681,6 +681,50 @@ async def test_empty_returns_empty(self, mock_redis_client: MagicMock):
         messages = await provider.get_messages("s1")
         assert messages == []
 
+    async def test_skips_malformed_entries(self, mock_redis_client: MagicMock):
+        msg = Message(role="user", contents=["Hello"])
+        mock_redis_client.lrange = AsyncMock(
+            return_value=[
+                "",
+                "not-json",
+                "null",  # valid JSON, but deserializes to None
+                "[]",  # valid JSON, but deserializes to a list
+                "{}",  # valid JSON, but missing the required "role" field
+                json.dumps(msg.to_dict()),
+            ]
+        )
+
+        with patch("agent_framework_redis._history_provider.redis.from_url") as mock_from_url:
+            mock_from_url.return_value = mock_redis_client
+            provider = RedisHistoryProvider("mem", redis_url="redis://localhost:6379", application_id="test-app")
+
+        messages = await provider.get_messages("s1")
+        assert len(messages) == 1
+        assert messages[0].text == "Hello"
+
+    async def test_malformed_entries_do_not_leak_payload_in_logs(
+        self, mock_redis_client: MagicMock, caplog: pytest.LogCaptureFixture
+    ):
+        # A payload whose ``type`` field carries a private value must not be echoed
+        # into the skip warning: ``Message.from_dict`` embeds ``type`` in mismatch
+        # errors, and logging ``str(exc)`` would copy it into application logs.
+        sensitive = "secret-internal-token"
+        mock_redis_client.lrange = AsyncMock(
+            return_value=[json.dumps({"type": sensitive, "role": "unknown"})]
+        )
+
+        with patch("agent_framework_redis._history_provider.redis.from_url") as mock_from_url:
+            mock_from_url.return_value = mock_redis_client
+            provider = RedisHistoryProvider("mem", redis_url="redis://localhost:6379", application_id="test-app")
+
+        with caplog.at_level("WARNING", logger="agent_framework_redis._history_provider"):
+            messages = await provider.get_messages("s1")
+
+        assert messages == []
+        warning_text = "\n".join(r.message for r in caplog.records)
+        assert "Skipping malformed Redis history entry" in warning_text
+        assert sensitive not in warning_text
+
     async def test_returns_messages_when_lrange_is_synchronous(self, mock_redis_client: MagicMock):
         """redis-py types several commands as returning a value or an awaitable; handle both."""
         msg = Message(role="user", contents=["Hello"])