fix(python-sdks): migrate agent-framework, cartesia, and pipecat to v4 APIs

Use client.add and search.memories hybrid mode, improve profile memory
deduplication for string/pydantic items, and add dedupe unit tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

packages/agent-framework-python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "supermemory-agent-framework"
-version = "1.0.0"
+version = "1.0.1"
 description = "Memory tools and middleware for Microsoft Agent Framework with supermemory"
 readme = "README.md"
 license = "MIT"

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

@@ -72,19 +72,20 @@ class SupermemoryTools:
         ] = True,
         limit: Annotated[int, "Maximum number of results to return"] = 10,
     ) -> str:
-        """Search (recall) memories/details/information about the user or other facts or entities. Run when explicitly asked or when context about user's past choices would be helpful."""
+        """Search stored memories for facts, preferences, history, and context. Use proactively before answering whenever memory could help — not only when explicitly asked."""
         try:
-            response = await self._client.search.execute(
+            response = await self._client.search.memories(
                 q=information_to_get,
                 container_tags=[self._connection.container_tag],
                 limit=limit,
-                chunk_threshold=0.6,
-                include_full_docs=include_full_docs,
+                threshold=0.6,
+                search_mode="hybrid",
             )
+            results = response.results or []
             result: MemorySearchResult = {
                 "success": True,
-                "results": response.results,
-                "count": len(response.results) if response.results else 0,
+                "results": results,
+                "count": len(results),
             }
             return json.dumps(result, default=str)
         except Exception as error:
@@ -152,9 +153,9 @@ class SupermemoryTools:
             tool(
                 name="search_memories",
                 description=(
-                    "Search (recall) memories/details/information about the user or other "
-                    "facts or entities. Run when explicitly asked or when context about "
-                    "user's past choices would be helpful."
+                    "Search (recall) stored memories for facts, preferences, history, and context "
+                    "about the user or any topic. Use proactively before answering whenever memory "
+                    "could help — do not wait for the user to explicitly ask you to search or recall."
                 ),
             )(self.search_memories),
             tool(

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

@@ -92,14 +92,19 @@ def deduplicate_memories(
     def extract_memory_text(item: Any) -> Optional[str]:
         if item is None:
             return None
+        if isinstance(item, str):
+            trimmed = item.strip()
+            return trimmed if trimmed else 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()
+        # Stainless SDK returns pydantic models (attribute access, snake_case).
+        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

@@ -56,6 +56,20 @@ class TestDeduplicateMemories:
         )
         assert result.static == ["valid"]
 
+    def test_pydantic_like_search_results(self) -> None:
+        """SDK search results are pydantic models, not dicts (#1266)."""
+        from types import SimpleNamespace
+
+        result = deduplicate_memories(
+            static=["User likes Python"],
+            search_results=[
+                SimpleNamespace(memory="User prefers async", updated_at="2026-01-01T00:00:00Z"),
+                SimpleNamespace(memory="User likes Python", updated_at=None),
+            ],
+        )
+        assert result.static == ["User likes Python"]
+        assert result.search_results == ["User prefers async"]
+
 
 class TestConvertProfileToMarkdown:
     def test_empty_profile(self) -> None:

packages/cartesia-sdk-python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "supermemory-cartesia"
-version = "0.1.1"
+version = "0.1.2"
 description = "Supermemory integration for Cartesia Line - memory-enhanced voice agents"
 readme = "README.md"
 license = "MIT"

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

@@ -151,31 +151,35 @@ class SupermemoryCartesiaAgent:
             raise MemoryRetrievalError("Supermemory client not initialized")
 
         try:
-            # Use primary container tag for profile retrieval
-            kwargs: Dict[str, Any] = {"container_tag": self.container_tags[0]}
+            logger.info(f"[Supermemory] Retrieving memories for query: {query[:50]}...")
 
+            # One profile call: static + dynamic, and (when mode/query allow)
+            # search_results via `q` — keeps a single round trip for latency.
+            kwargs: Dict[str, Any] = {"container_tag": self.container_tags[0]}
             if self.config.mode != "profile" and query:
                 kwargs["q"] = query
                 kwargs["threshold"] = self.config.search_threshold
                 kwargs["extra_body"] = {"limit": self.config.search_limit}
 
-            logger.info(f"[Supermemory] Retrieving memories for query: {query[:50]}...")
-
             response = await asyncio.wait_for(
                 self._supermemory_client.profile(**kwargs),
-                timeout=10.0
+                timeout=10.0,
             )
 
             # A user with no stored memories yet gets a null profile back, which
             # is a normal case, not an error. Guard against it so we return an
             # empty profile instead of raising AttributeError on response.profile.
             profile = getattr(response, "profile", None)
-            profile_static = profile.static if profile is not None and profile.static else []
-            profile_dynamic = profile.dynamic if profile is not None and profile.dynamic else []
+            profile_static = (
+                profile.static if profile is not None and profile.static else []
+            )
+            profile_dynamic = (
+                profile.dynamic if profile is not None and profile.dynamic else []
+            )
 
-            search_results = []
+            search_results: List[Any] = []
             if response.search_results and response.search_results.results:
-                search_results = response.search_results.results
+                search_results = list(response.search_results.results)
 
             logger.info(
                 f"[Supermemory] Retrieved memories - static: {len(profile_static)}, "

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

@@ -49,17 +49,37 @@ def format_relative_time(iso_timestamp: str) -> str:
         return ""
 
 
+def _field(item: Any, *names: str, default: Any = None) -> Any:
+    """Read a field from a dict or pydantic/SDK model.
+
+    Accepts camelCase and snake_case names so helpers work with both raw JSON
+    dicts and Stainless-generated response models.
+    """
+    if item is None:
+        return default
+    if isinstance(item, dict):
+        for name in names:
+            if name in item and item[name] is not None:
+                return item[name]
+        return default
+    for name in names:
+        value = getattr(item, name, None)
+        if value is not None:
+            return value
+    return default
+
+
 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, Union[List[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: Search result dicts or pydantic models with a memory field.
     """
     seen = set()
 
@@ -71,10 +91,14 @@ def deduplicate_memories(
                 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", "")
+            # v4 search.memories/hybrid uses `memory` or `chunk`.
+            memory = _field(r, "memory", "chunk", "content", default="")
+            if not isinstance(memory, str):
+                memory = ""
+            memory = memory.strip()
             if memory and memory not in seen:
                 seen.add(memory)
                 out.append(r)
@@ -88,7 +112,7 @@ def deduplicate_memories(
 
 
 def format_memories_to_text(
-    memories: Dict[str, Union[List[str], List[Dict[str, Any]]]],
+    memories: Dict[str, Union[List[str], List[Any]]],
     system_prompt: str = "Based on previous conversations, I recall:\n\n",
     include_static: bool = True,
     include_dynamic: bool = True,
@@ -116,16 +140,17 @@ 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 = _field(item, "memory", "chunk", "content", default="")
+            updated_at = _field(item, "updatedAt", "updated_at", default="")
+            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/test_dedupe_utils.py

@@ -0,0 +1,120 @@
+"""Regression tests for pydantic/dict memory helpers (#1266)."""
+
+from __future__ import annotations
+
+import sys
+import types
+import unittest
+from types import SimpleNamespace
+
+
+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()
+
+from supermemory_cartesia.utils import deduplicate_memories, format_memories_to_text
+
+
+class TestDeduplicateMemories(unittest.TestCase):
+    def test_accepts_dict_search_results(self) -> None:
+        result = deduplicate_memories(
+            static=["User likes Python"],
+            dynamic=[],
+            search_results=[{"memory": "User prefers async", "updatedAt": "2026-01-01T00:00:00Z"}],
+        )
+        self.assertEqual(result["static"], ["User likes Python"])
+        self.assertEqual(len(result["search_results"]), 1)
+
+    def test_accepts_pydantic_like_search_results(self) -> None:
+        # Mirrors supermemory.types.search_memories_response.Result
+        model = SimpleNamespace(
+            id="mem_1",
+            similarity=0.9,
+            memory="User prefers async",
+            updated_at="2026-01-01T00:00:00Z",
+        )
+        result = deduplicate_memories(
+            static=[],
+            dynamic=[],
+            search_results=[model],
+        )
+        self.assertEqual(len(result["search_results"]), 1)
+        self.assertIs(result["search_results"][0], model)
+
+    def test_dedupes_model_against_static_string(self) -> None:
+        model = SimpleNamespace(memory="User likes Python", updated_at=None)
+        result = deduplicate_memories(
+            static=["User likes Python"],
+            dynamic=[],
+            search_results=[model],
+        )
+        self.assertEqual(result["search_results"], [])
+
+
+class TestFormatMemoriesToText(unittest.TestCase):
+    def test_formats_pydantic_like_search_results(self) -> None:
+        text = format_memories_to_text(
+            {
+                "static": [],
+                "dynamic": [],
+                "search_results": [
+                    SimpleNamespace(
+                        memory="User prefers async",
+                        updated_at="2020-01-01T00:00:00Z",
+                    )
+                ],
+            }
+        )
+        self.assertIn("User prefers async", text)
+        self.assertIn("Relevant Memories", text)
+
+    def test_formats_search_execute_content_field(self) -> None:
+        text = format_memories_to_text(
+            {
+                "static": [],
+                "dynamic": [],
+                "search_results": [
+                    SimpleNamespace(
+                        content="User owns a telescope",
+                        updated_at="2020-01-01T00:00:00Z",
+                        memory=None,
+                    )
+                ],
+            }
+        )
+        self.assertIn("User owns a telescope", text)
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -71,6 +71,10 @@ class TestSupermemoryCartesiaNullProfile(unittest.IsolatedAsyncioTestCase):
                 "search_results": [],
             },
         )
+        agent._supermemory_client.profile.assert_awaited_once()
+        kwargs = agent._supermemory_client.profile.await_args.kwargs
+        self.assertEqual(kwargs["container_tag"], "user-123")
+        self.assertEqual(kwargs["q"], "Hello world")
 
 
 if __name__ == "__main__":

packages/pipecat-sdk-python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "supermemory-pipecat"
-version = "0.1.1"
+version = "0.1.2"
 description = "Supermemory integration for Pipecat - memory-enhanced conversational AI pipelines"
 readme = "README.md"
 license = "MIT"

packages/pipecat-sdk-python/src/supermemory_pipecat/service.py

@@ -137,8 +137,9 @@ class SupermemoryPipecatService(FrameProcessor):
             )
 
         try:
+            # One profile call: static + dynamic, and (when mode/query allow)
+            # search_results via `q`. This is the intended profile API shape.
             kwargs: Dict[str, Any] = {"container_tag": self.container_tag}
-
             if self.params.mode != "profile" and query:
                 kwargs["q"] = query
                 kwargs["threshold"] = self.params.search_threshold
@@ -149,9 +150,9 @@ class SupermemoryPipecatService(FrameProcessor):
             profile = getattr(response, "profile", None)
             search_results_response = getattr(response, "search_results", None)
 
-            search_results = []
+            search_results: List[Any] = []
             if search_results_response and search_results_response.results:
-                search_results = search_results_response.results
+                search_results = list(search_results_response.results)
 
             return {
                 "profile": {
@@ -179,7 +180,7 @@ class SupermemoryPipecatService(FrameProcessor):
             if self.session_id:
                 add_params["custom_id"] = self.session_id
 
-            await self._supermemory_client.memories.add(**add_params)
+            await self._supermemory_client.add(**add_params)
 
         except Exception as e:
             logger.error(f"Error storing messages: {e}")

packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py

@@ -49,17 +49,37 @@ def format_relative_time(iso_timestamp: str) -> str:
         return ""
 
 
+def _field(item: Any, *names: str, default: Any = None) -> Any:
+    """Read a field from a dict or pydantic/SDK model.
+
+    Accepts camelCase and snake_case names so helpers work with both raw JSON
+    dicts and Stainless-generated response models.
+    """
+    if item is None:
+        return default
+    if isinstance(item, dict):
+        for name in names:
+            if name in item and item[name] is not None:
+                return item[name]
+        return default
+    for name in names:
+        value = getattr(item, name, None)
+        if value is not None:
+            return value
+    return default
+
+
 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, Union[List[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: Search result dicts or pydantic models with a memory field.
     """
     seen = set()
 
@@ -71,10 +91,14 @@ def deduplicate_memories(
                 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", "")
+            # v4 search.memories/hybrid uses `memory` or `chunk`.
+            memory = _field(r, "memory", "chunk", "content", default="")
+            if not isinstance(memory, str):
+                memory = ""
+            memory = memory.strip()
             if memory and memory not in seen:
                 seen.add(memory)
                 out.append(r)
@@ -88,7 +112,7 @@ def deduplicate_memories(
 
 
 def format_memories_to_text(
-    memories: Dict[str, Union[List[str], List[Dict[str, Any]]]],
+    memories: Dict[str, Union[List[str], List[Any]]],
     system_prompt: str = "Based on previous conversations, I recall:\n\n",
     include_static: bool = True,
     include_dynamic: bool = True,
@@ -116,16 +140,17 @@ 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 = _field(item, "memory", "chunk", "content", default="")
+            updated_at = _field(item, "updatedAt", "updated_at", default="")
+            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/pipecat-sdk-python/tests/test_dedupe_utils.py

@@ -0,0 +1,180 @@
+"""Regression tests for pydantic/dict memory helpers (#1266)."""
+
+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 warning(self, *_args, **_kwargs):
+                return None
+
+            def error(self, *_args, **_kwargs):
+                return None
+
+            def info(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
+
+    if "pipecat" not in sys.modules:
+        pipecat_module = types.ModuleType("pipecat")
+        sys.modules["pipecat"] = pipecat_module
+
+        frames_module = types.ModuleType("pipecat.frames.frames")
+
+        class Frame:
+            pass
+
+        class InputAudioRawFrame:
+            pass
+
+        class LLMContextFrame:
+            pass
+
+        class LLMMessagesFrame:
+            pass
+
+        frames_module.Frame = Frame
+        frames_module.InputAudioRawFrame = InputAudioRawFrame
+        frames_module.LLMContextFrame = LLMContextFrame
+        frames_module.LLMMessagesFrame = LLMMessagesFrame
+
+        llm_context_module = types.ModuleType(
+            "pipecat.processors.aggregators.llm_context"
+        )
+
+        class LLMContext:
+            pass
+
+        llm_context_module.LLMContext = LLMContext
+
+        openai_context_module = types.ModuleType(
+            "pipecat.processors.aggregators.openai_llm_context"
+        )
+
+        class OpenAILLMContextFrame:
+            pass
+
+        openai_context_module.OpenAILLMContextFrame = OpenAILLMContextFrame
+
+        frame_processor_module = types.ModuleType("pipecat.processors.frame_processor")
+
+        class FrameDirection:
+            pass
+
+        class FrameProcessor:
+            def __init__(self, *args, **kwargs):
+                return None
+
+        frame_processor_module.FrameDirection = FrameDirection
+        frame_processor_module.FrameProcessor = FrameProcessor
+
+        sys.modules["pipecat.frames.frames"] = frames_module
+        sys.modules["pipecat.processors.aggregators.llm_context"] = llm_context_module
+        sys.modules[
+            "pipecat.processors.aggregators.openai_llm_context"
+        ] = openai_context_module
+        sys.modules["pipecat.processors.frame_processor"] = frame_processor_module
+
+
+_install_test_stubs()
+
+from supermemory_pipecat.service import SupermemoryPipecatService
+from supermemory_pipecat.utils import deduplicate_memories, format_memories_to_text
+
+
+class TestDeduplicateMemories(unittest.TestCase):
+    def test_accepts_dict_search_results(self) -> None:
+        result = deduplicate_memories(
+            static=["User likes Python"],
+            dynamic=[],
+            search_results=[{"memory": "User prefers async", "updatedAt": "2026-01-01T00:00:00Z"}],
+        )
+        self.assertEqual(result["static"], ["User likes Python"])
+        self.assertEqual(len(result["search_results"]), 1)
+
+    def test_accepts_pydantic_like_search_results(self) -> None:
+        model = SimpleNamespace(
+            id="mem_1",
+            similarity=0.9,
+            memory="User prefers async",
+            updated_at="2026-01-01T00:00:00Z",
+        )
+        result = deduplicate_memories(
+            static=[],
+            dynamic=[],
+            search_results=[model],
+        )
+        self.assertEqual(len(result["search_results"]), 1)
+        self.assertIs(result["search_results"][0], model)
+
+    def test_dedupes_model_against_static_string(self) -> None:
+        model = SimpleNamespace(memory="User likes Python", updated_at=None)
+        result = deduplicate_memories(
+            static=["User likes Python"],
+            dynamic=[],
+            search_results=[model],
+        )
+        self.assertEqual(result["search_results"], [])
+
+
+class TestFormatMemoriesToText(unittest.TestCase):
+    def test_formats_pydantic_like_search_results(self) -> None:
+        text = format_memories_to_text(
+            {
+                "static": [],
+                "dynamic": [],
+                "search_results": [
+                    SimpleNamespace(
+                        memory="User prefers async",
+                        updated_at="2020-01-01T00:00:00Z",
+                    )
+                ],
+            }
+        )
+        self.assertIn("User prefers async", text)
+        self.assertIn("Relevant Memories", text)
+
+
+class TestStoreMessagesUsesClientAdd(unittest.IsolatedAsyncioTestCase):
+    async def test_store_messages_calls_client_add(self) -> None:
+        service = SupermemoryPipecatService(api_key="mock_key", user_id="user-123")
+        service._supermemory_client = SimpleNamespace(add=AsyncMock())
+
+        await service._store_messages(
+            [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "hi"}]
+        )
+
+        service._supermemory_client.add.assert_awaited_once()
+        kwargs = service._supermemory_client.add.await_args.kwargs
+        self.assertIn("hello", kwargs["content"])
+        self.assertEqual(kwargs["container_tags"], ["user-123"])
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -120,4 +120,8 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase):
                 "profile": {"static": [], "dynamic": []},
                 "search_results": [],
             },
-        )
+        )
+        service._supermemory_client.profile.assert_awaited_once()
+        kwargs = service._supermemory_client.profile.await_args.kwargs
+        self.assertEqual(kwargs["container_tag"], "new_user_123")
+        self.assertEqual(kwargs["q"], "Hello world")