Add "reset" action to mem0_memory actions

pyproject.toml

@@ -82,6 +82,7 @@ docs = [
 mem0_memory = [
     # Need to be optional as a fix for https://github.com/strands-agents/docs/issues/19
     "mem0ai>=0.1.99,<1.0.0",
+    "langchain-aws>=0.2.23",
     "opensearch-py>=2.8.0,<3.0.0",
 ]
 local_chromium_browser = ["nest-asyncio>=1.5.0,<2.0.0", "playwright>=1.42.0,<2.0.0"]

src/strands_tools/mem0_memory.py

@@ -1,5 +1,5 @@
 """
-Tool for managing memories using Mem0 (store, delete, list, get, and retrieve)
+Tool for managing memories using Mem0 (store, delete, list, get, retrieve, and reset)
 
 This module provides comprehensive memory management capabilities using
 Mem0 as the backend. It handles all aspects of memory management with
@@ -13,6 +13,7 @@
    • list: Retrieve all memories for a user or agent
    • get: Retrieve specific memories by memory ID
    • retrieve: Perform semantic search across all memories
+   • reset: Remove all existing memories and relationships
 
 2. Safety Features:
    • User confirmation for mutative operations
@@ -95,14 +96,16 @@
         "2. Retrieve memories by ID or semantic search (requires user_id or agent_id)\n"
         "3. List all memories for a user/agent (requires user_id or agent_id)\n"
         "4. Delete memories\n"
-        "5. Get memory history\n\n"
+        "5. Get memory history\n"
+        "6. Reset all memories\n\n"
         "Actions:\n"
         "- store: Store new memory (requires user_id or agent_id)\n"
         "- get: Get memory by ID\n"
         "- list: List all memories (requires user_id or agent_id)\n"
         "- retrieve: Semantic search (requires user_id or agent_id)\n"
         "- delete: Delete memory\n"
-        "- history: Get memory history\n\n"
+        "- history: Get memory history\n"
+        "- reset: Delete all memories\n\n"
         "Note: Most operations require either user_id or agent_id to be specified. The tool will automatically "
         "attempt to retrieve relevant memories when user_id or agent_id is available."
     ),
@@ -112,8 +115,8 @@
             "properties": {
                 "action": {
                     "type": "string",
-                    "description": ("Action to perform (store, get, list, retrieve, delete, history)"),
-                    "enum": ["store", "get", "list", "retrieve", "delete", "history"],
+                    "description": ("Action to perform (store, get, list, retrieve, delete, history, reset)"),
+                    "enum": ["store", "get", "list", "retrieve", "delete", "history", "reset"],
                 },
                 "content": {
                     "type": "string",
@@ -129,11 +132,11 @@
                 },
                 "user_id": {
                     "type": "string",
-                    "description": "User ID for the memory operations (required for store, list, retrieve actions)",
+                    "description": "User ID for the memory operations (either user_id or agent_id is required for store, list, retrieve, and reset actions)",
                 },
                 "agent_id": {
                     "type": "string",
-                    "description": "Agent ID for the memory operations (required for store, list, retrieve actions)",
+                    "description": "Agent ID for the memory operations (either user_id or agent_id is required for store, list, retrieve, and reset actions)",
                 },
                 "metadata": {
                     "type": "object",
@@ -415,6 +418,13 @@ def get_memory_history(self, memory_id: str):
         """Get the history of a memory by ID."""
         return self.mem0.history(memory_id)
 
+    def reset_memories(self, user_id: Optional[str] = None, agent_id: Optional[str] = None):
+        """Delete all memories for a user or agent."""
+        if not user_id and not agent_id:
+            raise ValueError("Either user_id or agent_id must be provided")
+
+        return self.mem0.delete_all(user_id=user_id, agent_id=agent_id)
+
 
 def format_get_response(memory: Dict) -> Panel:
     """Format get memory response."""
@@ -672,7 +682,7 @@ def mem0_memory(tool: ToolUse, **kwargs: Any) -> ToolResult:
         action = tool_input["action"]
 
         # For mutative operations, show confirmation dialog unless in BYPASS_TOOL_CONSENT mode
-        mutative_actions = {"store", "delete"}
+        mutative_actions = {"store", "delete", "reset"}
         needs_confirmation = action in mutative_actions and not strands_dev
 
         if needs_confirmation:
@@ -724,6 +734,27 @@ def mem0_memory(tool: ToolUse, **kwargs: Any) -> ToolResult:
                             border_style="red",
                         )
                     )
+            elif action == "reset":
+                # Try to get memory info first for better context
+                try:
+                    client.reset_memories(tool_input.get("user_id"), tool_input.get("agent_id"))
+
+                    console.print(
+                        Panel(
+                            "Deleting memories...",
+                            title="[bold red]⚠️ All Memories to be permanently deleted",
+                            border_style="red",
+                        )
+                    )
+                except Exception:
+                    # Fall back to basic info if we can't get memory details
+                    console.print(
+                        Panel(
+                            "Deleting memories...",
+                            title="[bold red]⚠️ All Memories to be permanently deleted",
+                            border_style="red",
+                        )
+                    )
 
         # Execute the requested action
         if action == "store":
@@ -827,6 +858,16 @@ def mem0_memory(tool: ToolUse, **kwargs: Any) -> ToolResult:
                 content=[ToolResultContent(text=f"Memory {tool_input['memory_id']} deleted successfully")],
             )
 
+        elif action == "reset":
+            client.reset_memories(tool_input.get("user_id"), tool_input.get("agent_id"))
+            panel = Panel("All memories deleted.", title="[bold green]Memories Deleted", border_style="green")
+            console.print(panel)
+            return ToolResult(
+                toolUseId=tool_use_id,
+                status="success",
+                content=[ToolResultContent(text=f"Memories deleted successfully")],
+            )
+
         elif action == "history":
             if not tool_input.get("memory_id"):
                 raise ValueError("memory_id is required for history action")