fix(python-sdks): handle pydantic search results in memory dedup helpers

packages/agent-framework-python/src/supermemory_agent_framework/utils.py

@@ -92,14 +92,15 @@ def deduplicate_memories(
     def extract_memory_text(item: Any) -> Optional[str]:
         if item is None:
             return None
-        if isinstance(item, dict):
-            memory = item.get("memory")
-            if isinstance(memory, str):
-                trimmed = memory.strip()
-                return trimmed if trimmed else None
-            return None
         if isinstance(item, str):
-            trimmed = item.strip()
+            memory: Any = item
+        elif isinstance(item, dict):
+            memory = item.get("memory")
+        else:
+            # SDK search results are pydantic models with a `memory` attribute
+            memory = getattr(item, "memory", None)
+        if isinstance(memory, str):
+            trimmed = memory.strip()
             return trimmed if trimmed else None
         return None
 

packages/agent-framework-python/tests/test_utils.py

@@ -1,5 +1,7 @@
 """Tests for utility functions."""
 
+from types import SimpleNamespace
+
 import pytest
 
 from supermemory_agent_framework.utils import (
@@ -56,6 +58,30 @@ def test_none_items_filtered(self) -> None:
         )
         assert result.static == ["valid"]
 
+    def test_model_items(self) -> None:
+        # SDK search results are pydantic models with a `memory` attribute,
+        # not dicts — they must not be silently dropped (#1266).
+        result = deduplicate_memories(
+            search_results=[SimpleNamespace(memory="User prefers async")],
+        )
+        assert result.search_results == ["User prefers async"]
+
+    def test_model_items_deduplicate_against_static(self) -> None:
+        result = deduplicate_memories(
+            static=[{"memory": "User likes Python"}],
+            search_results=[
+                SimpleNamespace(memory="User likes Python"),
+                SimpleNamespace(memory="User prefers async"),
+            ],
+        )
+        assert result.search_results == ["User prefers async"]
+
+    def test_model_items_without_memory_text_filtered(self) -> None:
+        result = deduplicate_memories(
+            search_results=[SimpleNamespace(memory=None), SimpleNamespace(memory="  ")],
+        )
+        assert result.search_results == []
+
 
 class TestConvertProfileToMarkdown:
     def test_empty_profile(self) -> None:

packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py

@@ -1,7 +1,27 @@
 """Utility functions for Supermemory Cartesia integration."""
 
 from datetime import datetime, timezone
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List
+
+
+def _get_result_field(result: Any, *keys: str) -> Any:
+    """Read a field from a search result that may be a dict or an SDK model.
+
+    The profile endpoint returns search results as plain dicts with camelCase
+    keys, while the typed SDK models expose the same data as snake_case
+    attributes. Accept both shapes so callers don't depend on the SDK version.
+    """
+    if isinstance(result, dict):
+        for key in keys:
+            value = result.get(key)
+            if value is not None:
+                return value
+        return None
+    for key in keys:
+        value = getattr(result, key, None)
+        if value is not None:
+            return value
+    return None
 
 
 def get_last_user_message(messages: List[Dict[str, str]]) -> str | None:
@@ -52,14 +72,15 @@ def format_relative_time(iso_timestamp: str) -> str:
 def deduplicate_memories(
     static: List[str],
     dynamic: List[str],
-    search_results: List[Dict[str, Any]],
-) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]:
+    search_results: List[Any],
+) -> Dict[str, List[Any]]:
     """Deduplicate memories. Priority: static > dynamic > search.
 
     Args:
         static: List of static memory strings.
         dynamic: List of dynamic memory strings.
-        search_results: List of search result dicts with 'memory' and 'updatedAt'.
+        search_results: List of search results with 'memory' and 'updatedAt',
+            either as dicts or as SDK result models.
     """
     seen = set()
 
@@ -71,10 +92,10 @@ def unique_strings(memories: List[str]) -> List[str]:
                 out.append(m)
         return out
 
-    def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    def unique_search(results: List[Any]) -> List[Any]:
         out = []
         for r in results:
-            memory = r.get("memory", "")
+            memory = _get_result_field(r, "memory")
             if memory and memory not in seen:
                 seen.add(memory)
                 out.append(r)
@@ -88,7 +109,7 @@ def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
 
 
 def format_memories_to_text(
-    memories: Dict[str, Union[List[str], List[Dict[str, Any]]]],
+    memories: Dict[str, List[Any]],
     system_prompt: str = "Based on previous conversations, I recall:\n\n",
     include_static: bool = True,
     include_dynamic: bool = True,
@@ -116,16 +137,16 @@ def format_memories_to_text(
         sections.append("## Relevant Memories")
         lines = []
         for item in search_results:
-            if isinstance(item, dict):
-                memory = item.get("memory", "")
-                updated_at = item.get("updatedAt", "")
-                time_str = format_relative_time(updated_at) if updated_at else ""
-                if time_str:
-                    lines.append(f"- [{time_str}] {memory}")
-                else:
-                    lines.append(f"- {memory}")
-            else:
+            if isinstance(item, str):
                 lines.append(f"- {item}")
+                continue
+            memory = _get_result_field(item, "memory") or ""
+            updated_at = _get_result_field(item, "updatedAt", "updated_at") or ""
+            time_str = format_relative_time(updated_at) if updated_at else ""
+            if time_str:
+                lines.append(f"- [{time_str}] {memory}")
+            else:
+                lines.append(f"- {memory}")
         sections.append("\n".join(lines))
 
     if not sections:

packages/cartesia-sdk-python/tests/conftest.py

@@ -0,0 +1,46 @@
+"""Shared pytest fixtures: stub out heavy optional dependencies.
+
+The stubs are installed at collection time so test modules can import the
+package without pipecat/cartesia, loguru, or pydantic present.
+"""
+
+from __future__ import annotations
+
+import sys
+import types
+
+
+def _install_test_stubs() -> None:
+    if "loguru" not in sys.modules:
+        loguru_module = types.ModuleType("loguru")
+
+        class _Logger:
+            def info(self, *_args, **_kwargs):
+                return None
+
+            def warning(self, *_args, **_kwargs):
+                return None
+
+            def error(self, *_args, **_kwargs):
+                return None
+
+        loguru_module.logger = _Logger()
+        sys.modules["loguru"] = loguru_module
+
+    if "pydantic" not in sys.modules:
+        pydantic_module = types.ModuleType("pydantic")
+
+        class BaseModel:
+            def __init__(self, **kwargs):
+                for key, value in kwargs.items():
+                    setattr(self, key, value)
+
+        def Field(*, default=None, **_kwargs):
+            return default
+
+        pydantic_module.BaseModel = BaseModel
+        pydantic_module.Field = Field
+        sys.modules["pydantic"] = pydantic_module
+
+
+_install_test_stubs()

packages/cartesia-sdk-python/tests/test_empty_profile.py

@@ -1,44 +1,10 @@
 from __future__ import annotations
 
-import sys
-import types
 import unittest
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
-
-def _install_test_stubs() -> None:
-    if "loguru" not in sys.modules:
-        loguru_module = types.ModuleType("loguru")
-
-        class _Logger:
-            def info(self, *_args, **_kwargs):
-                return None
-
-            def warning(self, *_args, **_kwargs):
-                return None
-
-            def error(self, *_args, **_kwargs):
-                return None
-
-        loguru_module.logger = _Logger()
-        sys.modules["loguru"] = loguru_module
-
-    if "pydantic" not in sys.modules:
-        pydantic_module = types.ModuleType("pydantic")
-
-        class BaseModel:
-            def __init__(self, **kwargs):
-                for key, value in kwargs.items():
-                    setattr(self, key, value)
-
-        def Field(*, default=None, **_kwargs):
-            return default
-
-        pydantic_module.BaseModel = BaseModel
-        pydantic_module.Field = Field
-        sys.modules["pydantic"] = pydantic_module
-
+from .conftest import _install_test_stubs
 
 _install_test_stubs()
 

packages/cartesia-sdk-python/tests/test_utils.py

@@ -0,0 +1,92 @@
+"""Tests for the memory dedup/format helpers.
+
+Search results reach these helpers either as plain camelCase dicts (profile
+endpoint) or as typed SDK result models (snake_case attributes, no dict
+interface). Both shapes must be handled — see issue #1266.
+"""
+
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+
+from .conftest import _install_test_stubs
+
+_install_test_stubs()
+
+from supermemory_cartesia.utils import deduplicate_memories, format_memories_to_text
+
+
+def _model_result(memory, updated_at=None):
+    """Stand-in for an SDK result model: attribute access only, no .get()."""
+    return SimpleNamespace(memory=memory, updated_at=updated_at)
+
+
+class TestDeduplicateMemories(unittest.TestCase):
+    def test_dict_results_with_camel_case_keys(self) -> None:
+        results = [{"memory": "User prefers async", "updatedAt": "2026-01-01T00:00:00Z"}]
+
+        deduplicated = deduplicate_memories(static=[], dynamic=[], search_results=results)
+
+        self.assertEqual(deduplicated["search_results"], results)
+
+    def test_model_results_use_attribute_access(self) -> None:
+        result = _model_result("User prefers async")
+
+        deduplicated = deduplicate_memories(static=[], dynamic=[], search_results=[result])
+
+        self.assertEqual(deduplicated["search_results"], [result])
+
+    def test_model_results_deduplicate_against_profile(self) -> None:
+        deduplicated = deduplicate_memories(
+            static=["User prefers async"],
+            dynamic=[],
+            search_results=[
+                _model_result("User prefers async"),
+                _model_result("User works remotely"),
+            ],
+        )
+
+        self.assertEqual(deduplicated["static"], ["User prefers async"])
+        self.assertEqual(
+            [r.memory for r in deduplicated["search_results"]],
+            ["User works remotely"],
+        )
+
+    def test_results_without_memory_are_dropped(self) -> None:
+        deduplicated = deduplicate_memories(
+            static=[],
+            dynamic=[],
+            search_results=[_model_result(None), {"updatedAt": "2026-01-01T00:00:00Z"}],
+        )
+
+        self.assertEqual(deduplicated["search_results"], [])
+
+
+class TestFormatMemoriesToText(unittest.TestCase):
+    def _format(self, search_results) -> str:
+        return format_memories_to_text(
+            {"static": [], "dynamic": [], "search_results": search_results}
+        )
+
+    def test_dict_results_render_memory_and_relative_time(self) -> None:
+        text = self._format(
+            [{"memory": "User prefers async", "updatedAt": "2020-01-05T00:00:00Z"}]
+        )
+
+        self.assertIn("- [5 Jan, 2020] User prefers async", text)
+
+    def test_model_results_render_memory_and_relative_time(self) -> None:
+        text = self._format([_model_result("User prefers async", "2020-01-05T00:00:00Z")])
+
+        self.assertIn("- [5 Jan, 2020] User prefers async", text)
+
+    def test_model_results_without_timestamp_render_memory_only(self) -> None:
+        text = self._format([_model_result("User prefers async")])
+
+        self.assertIn("- User prefers async", text)
+
+    def test_string_results_render_verbatim(self) -> None:
+        text = self._format(["User prefers async"])
+
+        self.assertIn("- User prefers async", text)