fix: Accept single memory object in create_long_term_memory tool

LLMs often try to pass a single memory object instead of an array
when calling create_long_term_memory. This change makes the tool
more flexible by accepting both formats:

- Single object: {text: '...', memory_type: '...'}
- Array: [{text: '...', memory_type: '...'}]

When a single object is passed, it's automatically wrapped in an array
before being sent to the API.

Changes:
- Updated function signature to accept dict | list[dict]
- Added automatic wrapping logic for single objects
- Improved description to clarify both formats are accepted
- Added test case for single object handling

This fixes the issue where LLMs would get errors when trying to create
a single memory at a time.

Author: abrookins

agent-memory-client/agent_memory_client/integrations/langchain.py

@@ -158,7 +158,7 @@ def get_memory_tools(
         },
         "create_long_term_memory": {
             "name": "create_long_term_memory",
-            "description": "Create long-term memories directly for immediate storage and retrieval. Use this for important information that should be permanently stored without going through working memory.",
+            "description": "Create long-term memories directly for immediate storage and retrieval. Use this for important information that should be permanently stored without going through working memory. You can pass a single memory object or a list of memory objects. Each memory needs: text (string), memory_type ('episodic' or 'semantic'), and optionally topics (list), entities (list), event_date (ISO string).",
             "func": _create_create_long_term_memory_func(
                 memory_client, namespace, user_id
             ),
@@ -345,8 +345,18 @@ def _create_create_long_term_memory_func(
 ) -> Any:
     """Create create_long_term_memory function."""
 
-    async def create_long_term_memory(memories: list[dict[str, Any]]) -> str:
-        """Create long-term memories directly for immediate storage."""
+    async def create_long_term_memory(
+        memories: list[dict[str, Any]] | dict[str, Any],
+    ) -> str:
+        """Create long-term memories directly for immediate storage.
+
+        Accepts either a single memory object or a list of memory objects.
+        Each memory should have: text, memory_type, and optionally topics, entities, event_date.
+        """
+        # Handle single memory object - wrap it in a list
+        if isinstance(memories, dict):
+            memories = [memories]
+
         result = await client.resolve_function_call(
             function_name="create_long_term_memory",
             function_arguments={"memories": memories},

agent-memory-client/tests/test_langchain_integration.py

@@ -245,6 +245,42 @@ async def test_create_long_term_memory_tool_execution(self):
         assert call_kwargs["function_name"] == "create_long_term_memory"
         assert len(call_kwargs["function_arguments"]["memories"]) == 2
 
+    @pytest.mark.skipif(not _langchain_available(), reason="LangChain not installed")
+    @pytest.mark.asyncio
+    async def test_create_long_term_memory_accepts_single_object(self):
+        """Test that create_long_term_memory accepts a single memory object (not just array)."""
+        from agent_memory_client.integrations.langchain import get_memory_tools
+
+        client = _create_mock_client()
+        client.resolve_function_call = AsyncMock(
+            return_value={
+                "success": True,
+                "formatted_response": "Created 1 long-term memory",
+            }
+        )
+
+        tools = get_memory_tools(
+            memory_client=client,
+            session_id="test_session",
+            user_id="test_user",
+            tools=["create_long_term_memory"],
+        )
+
+        create_tool = tools[0]
+
+        # Execute with single object (what LLMs often do)
+        single_memory = {"text": "User loves pizza", "memory_type": "semantic"}
+        result = await create_tool.ainvoke({"memories": single_memory})
+
+        assert "Created 1 long-term memory" in result
+
+        # Verify it was converted to an array
+        call_args = client.resolve_function_call.call_args
+        memories_arg = call_args.kwargs["function_arguments"]["memories"]
+        assert isinstance(memories_arg, list)
+        assert len(memories_arg) == 1
+        assert memories_arg[0] == single_memory
+
     @pytest.mark.skipif(not _langchain_available(), reason="LangChain not installed")
     def test_tool_has_correct_schema(self):
         """Test that generated tools have correct schemas."""

check_schema.py

@@ -0,0 +1,35 @@
+"""Check the actual schema generated by LangChain for create_long_term_memory."""
+
+import asyncio
+import json
+
+from agent_memory_client import MemoryAPIClient, MemoryClientConfig
+from agent_memory_client.integrations.langchain import get_memory_tools
+
+
+async def main():
+    # Initialize memory client without health check
+    config = MemoryClientConfig(base_url="http://localhost:8000")
+    memory_client = MemoryAPIClient(config)
+
+    # Get memory tools
+    tools = get_memory_tools(
+        memory_client=memory_client,
+        session_id="test_session",
+        user_id="alice",
+        tools=["create_long_term_memory", "add_memory_to_working_memory"],
+    )
+
+    # Check both tools
+    for tool in tools:
+        print("=" * 80)
+        print("Tool Name:", tool.name)
+        print("\nTool Description:")
+        print(tool.description)
+        print("\nTool Args Schema:")
+        print(json.dumps(tool.args_schema.schema(), indent=2))
+        print()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

test_single_memory.py

@@ -0,0 +1,82 @@
+"""Test that create_long_term_memory accepts both single object and array."""
+
+import asyncio
+from unittest.mock import AsyncMock
+
+from agent_memory_client import MemoryAPIClient, MemoryClientConfig
+from agent_memory_client.integrations.langchain import get_memory_tools
+
+
+async def main():
+    # Create mock client
+    config = MemoryClientConfig(base_url="http://localhost:8000")
+    client = MemoryAPIClient(config)
+
+    # Mock the resolve_function_call method
+    client.resolve_function_call = AsyncMock(
+        return_value={
+            "success": True,
+            "formatted_response": "Created 1 memory successfully",
+        }
+    )
+
+    # Get the tool
+    tools = get_memory_tools(
+        memory_client=client,
+        session_id="test",
+        user_id="alice",
+        tools=["create_long_term_memory"],
+    )
+
+    create_tool = tools[0]
+
+    # Test 1: Single memory object (what LLMs often do)
+    print("Test 1: Single memory object")
+    single_memory = {
+        "text": "User loves pizza",
+        "memory_type": "semantic",
+        "topics": ["food", "preferences"],
+    }
+
+    result = await create_tool.func(memories=single_memory)
+    print(f"Result: {result}")
+
+    # Check what was actually called
+    call_args = client.resolve_function_call.call_args
+    print(f"Called with: {call_args}")
+    memories_arg = call_args.kwargs["function_arguments"]["memories"]
+    print(f"Memories argument type: {type(memories_arg)}")
+    print(f"Memories argument value: {memories_arg}")
+    assert isinstance(memories_arg, list), "Should convert single object to list"
+    assert len(memories_arg) == 1, "Should have one memory"
+    print("✅ Single object correctly converted to array\n")
+
+    # Test 2: Array of memories (correct format)
+    print("Test 2: Array of memories")
+    client.resolve_function_call.reset_mock()
+
+    memory_array = [
+        {"text": "User loves pizza", "memory_type": "semantic", "topics": ["food"]},
+        {
+            "text": "User works at TechCorp",
+            "memory_type": "semantic",
+            "topics": ["work"],
+        },
+    ]
+
+    result = await create_tool.func(memories=memory_array)
+    print(f"Result: {result}")
+
+    call_args = client.resolve_function_call.call_args
+    memories_arg = call_args.kwargs["function_arguments"]["memories"]
+    print(f"Memories argument type: {type(memories_arg)}")
+    print(f"Memories argument value: {memories_arg}")
+    assert isinstance(memories_arg, list), "Should keep as list"
+    assert len(memories_arg) == 2, "Should have two memories"
+    print("✅ Array correctly passed through\n")
+
+    print("All tests passed! ✅")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())