Fix --no-worker mode and add deduplicate parameter

- Fix background task execution in --no-worker mode by using HybridBackgroundTasks directly as type annotation instead of Depends(get_background_tasks)
- Add deduplicate parameter to create_long_term_memory endpoint with default true
- Pass background_tasks parameter through search_long_term_memory and memory_prompt endpoints
- Update MCP server to create HybridBackgroundTasks instance for memory_prompt calls
- Update tests to patch actual functions instead of BackgroundTasks class
- Add test to verify tasks run inline in --no-worker mode
- Bump server version to 0.12.3 and client version to 0.12.7

Author: abrookins

agent-memory-client/agent_memory_client/__init__.py

@@ -5,7 +5,7 @@
 memory management capabilities for AI agents and applications.
 """
 
-__version__ = "0.12.6"
+__version__ = "0.12.7"
 
 from .client import MemoryAPIClient, MemoryClientConfig, create_memory_client
 from .exceptions import (

agent-memory-client/agent_memory_client/client.py

@@ -618,13 +618,16 @@ async def add_memories_to_working_memory(
         return await self.put_working_memory(session_id, working_memory)
 
     async def create_long_term_memory(
-        self, memories: Sequence[ClientMemoryRecord | MemoryRecord]
+        self,
+        memories: Sequence[ClientMemoryRecord | MemoryRecord],
+        deduplicate: bool = True,
     ) -> AckResponse:
         """
         Create long-term memories for later retrieval.
 
         Args:
             memories: List of MemoryRecord objects to store
+            deduplicate: Whether to deduplicate memories before indexing (default: True)
 
         Returns:
             AckResponse indicating success
@@ -668,7 +671,10 @@ async def create_long_term_memory(
                 memory.id = str(ULID())
 
         payload = {
-            "memories": [m.model_dump(exclude_none=True, mode="json") for m in memories]
+            "memories": [
+                m.model_dump(exclude_none=True, mode="json") for m in memories
+            ],
+            "deduplicate": deduplicate,
         }
 
         try:

agent_memory_server/__init__.py

@@ -1,3 +1,3 @@
 """Redis Agent Memory Server - A memory system for conversational AI."""
 
-__version__ = "0.12.2"
+__version__ = "0.12.3"

agent_memory_server/api.py

@@ -9,7 +9,7 @@
 from agent_memory_server import long_term_memory, working_memory
 from agent_memory_server.auth import UserInfo, get_current_user
 from agent_memory_server.config import settings
-from agent_memory_server.dependencies import get_background_tasks
+from agent_memory_server.dependencies import HybridBackgroundTasks
 from agent_memory_server.filters import SessionId, UserId
 from agent_memory_server.llms import get_model_client, get_model_config
 from agent_memory_server.logging import get_logger
@@ -456,9 +456,9 @@ async def get_working_memory(
 async def put_working_memory(
     session_id: str,
     memory: UpdateWorkingMemory,
+    background_tasks: HybridBackgroundTasks,
     model_name: ModelNameLiteral | None = None,
     context_window_max: int | None = None,
-    background_tasks=Depends(get_background_tasks),
     current_user: UserInfo = Depends(get_current_user),
 ):
     """
@@ -593,7 +593,7 @@ async def delete_working_memory(
 @router.post("/v1/long-term-memory/", response_model=AckResponse)
 async def create_long_term_memory(
     payload: CreateMemoryRecordRequest,
-    background_tasks=Depends(get_background_tasks),
+    background_tasks: HybridBackgroundTasks,
     current_user: UserInfo = Depends(get_current_user),
 ):
     """
@@ -609,9 +609,9 @@ async def create_long_term_memory(
     if not settings.long_term_memory:
         raise HTTPException(status_code=400, detail="Long-term memory is disabled")
 
-    # Validate and process memories according to Stage 2 requirements
+    # Validate and process memories
     for memory in payload.memories:
-        # Enforce that id is required on memory sent from clients
+        # Enforce that ID is required on memory sent from clients
         if not memory.id:
             raise HTTPException(
                 status_code=400, detail="id is required for all memory records"
@@ -624,13 +624,15 @@ async def create_long_term_memory(
     background_tasks.add_task(
         long_term_memory.index_long_term_memories,
         memories=payload.memories,
+        deduplicate=payload.deduplicate,
     )
     return AckResponse(status="ok")
 
 
 @router.post("/v1/long-term-memory/search", response_model=MemoryRecordResultsResponse)
 async def search_long_term_memory(
     payload: SearchRequest,
+    background_tasks: HybridBackgroundTasks,
     optimize_query: bool = False,
     current_user: UserInfo = Depends(get_current_user),
 ):
@@ -752,7 +754,6 @@ def _vals(f):
         # Update last_accessed in background with rate limiting
         ids = [m.id for m in ranked if m.id]
         if ids:
-            background_tasks = get_background_tasks()
             background_tasks.add_task(long_term_memory.update_last_accessed, ids)
 
         raw_results.memories = ranked
@@ -853,6 +854,7 @@ async def update_long_term_memory(
 @router.post("/v1/memory/prompt", response_model=MemoryPromptResponse)
 async def memory_prompt(
     params: MemoryPromptRequest,
+    background_tasks: HybridBackgroundTasks,
     optimize_query: bool = False,
     current_user: UserInfo = Depends(get_current_user),
 ) -> MemoryPromptResponse:
@@ -992,6 +994,7 @@ async def memory_prompt(
         logger.debug(f"[memory_prompt] Search payload: {search_payload}")
         long_term_memories = await search_long_term_memory(
             search_payload,
+            background_tasks,
             optimize_query=optimize_query,
         )
 

agent_memory_server/dependencies.py

@@ -61,7 +61,20 @@ def get_background_tasks() -> HybridBackgroundTasks:
     """
     Dependency function that returns a HybridBackgroundTasks instance.
 
-    This is used by API endpoints to inject a consistent background tasks object.
+    NOTE: This function is deprecated. Use HybridBackgroundTasks directly as a type
+    annotation in your endpoint instead of Depends(get_background_tasks).
+
+    Example:
+        # Old way (deprecated):
+        async def endpoint(background_tasks=Depends(get_background_tasks)):
+            ...
+
+        # New way (correct):
+        async def endpoint(background_tasks: HybridBackgroundTasks):
+            ...
+
+    FastAPI will automatically inject the correct instance when you use
+    HybridBackgroundTasks as a type annotation.
     """
     logger.info("Getting background tasks class")
     return HybridBackgroundTasks()

agent_memory_server/long_term_memory.py

@@ -431,7 +431,7 @@ async def compact_long_term_memories(
     session_id: str | None = None,
     llm_client: OpenAIClientWrapper | AnthropicClientWrapper | None = None,
     redis_client: Redis | None = None,
-    vector_distance_threshold: float = 0.12,
+    vector_distance_threshold: float = 0.2,
     compact_hash_duplicates: bool = True,
     compact_semantic_duplicates: bool = True,
     perpetual: Perpetual = Perpetual(
@@ -1173,7 +1173,7 @@ async def deduplicate_by_semantic_search(
     namespace: str | None = None,
     user_id: str | None = None,
     session_id: str | None = None,
-    vector_distance_threshold: float = 0.12,
+    vector_distance_threshold: float = 0.2,
 ) -> tuple[MemoryRecord | None, bool]:
     """
     Check if a memory has semantic duplicates and merge if found.

agent_memory_server/mcp.py

@@ -689,8 +689,14 @@ async def memory_prompt(
     if search_payload is not None:
         _params["long_term_search"] = search_payload
 
+    # Create a background tasks instance for the MCP call
+    from agent_memory_server.dependencies import HybridBackgroundTasks
+
+    background_tasks = HybridBackgroundTasks()
+
     return await core_memory_prompt(
         params=MemoryPromptRequest(query=query, **_params),
+        background_tasks=background_tasks,
         optimize_query=optimize_query,
     )
 

agent_memory_server/models.py

@@ -526,6 +526,10 @@ class CreateMemoryRecordRequest(BaseModel):
     """Payload for creating memory records"""
 
     memories: list[ExtractedMemoryRecord]
+    deduplicate: bool = Field(
+        default=True,
+        description="Whether to deduplicate memories before indexing",
+    )
 
 
 class GetSessionsQuery(BaseModel):

tests/conftest.py

@@ -14,7 +14,7 @@
 
 from agent_memory_server.api import router as memory_router
 from agent_memory_server.config import settings
-from agent_memory_server.dependencies import HybridBackgroundTasks, get_background_tasks
+from agent_memory_server.dependencies import HybridBackgroundTasks
 from agent_memory_server.healthcheck import router as health_router
 from agent_memory_server.llms import OpenAIClientWrapper
 from agent_memory_server.models import (
@@ -407,8 +407,8 @@ def app(use_test_redis_connection):
 
 
 @pytest.fixture()
-def app_with_mock_background_tasks(use_test_redis_connection, mock_background_tasks):
-    """Create a test FastAPI app with routers"""
+def app_with_mock_background_tasks(use_test_redis_connection):
+    """Create a test FastAPI app with routers and mocked background tasks"""
     app = FastAPI()
 
     # Include routers
@@ -423,7 +423,6 @@ async def mock_get_redis_conn(*args, **kwargs):
     from agent_memory_server.utils.redis import get_redis_conn
 
     app.dependency_overrides[get_redis_conn] = mock_get_redis_conn
-    app.dependency_overrides[get_background_tasks] = lambda: mock_background_tasks
 
     return app
 
@@ -447,9 +446,27 @@ async def client(app):
 
 
 @pytest.fixture()
-async def client_with_mock_background_tasks(app_with_mock_background_tasks):
-    async with AsyncClient(
-        transport=ASGITransport(app=app_with_mock_background_tasks),
-        base_url="http://test",
-    ) as client:
-        yield client
+async def client_with_mock_background_tasks(
+    app_with_mock_background_tasks, mock_background_tasks
+):
+    """Client with mocked background tasks - patches the HybridBackgroundTasks class"""
+    # Patch the HybridBackgroundTasks class to return our mock
+    # We need to patch it in multiple places since FastAPI creates instances directly
+    patches = [
+        mock.patch(
+            "agent_memory_server.api.HybridBackgroundTasks",
+            return_value=mock_background_tasks,
+        ),
+        mock.patch(
+            "agent_memory_server.dependencies.HybridBackgroundTasks",
+            return_value=mock_background_tasks,
+        ),
+        mock.patch("fastapi.BackgroundTasks", return_value=mock_background_tasks),
+    ]
+
+    with patches[0], patches[1], patches[2]:
+        async with AsyncClient(
+            transport=ASGITransport(app=app_with_mock_background_tasks),
+            base_url="http://test",
+        ) as client:
+            yield client

tests/test_api.py

@@ -4,9 +4,6 @@
 import pytest
 
 from agent_memory_server.config import Settings
-from agent_memory_server.long_term_memory import (
-    promote_working_memory_to_long_term,
-)
 from agent_memory_server.models import (
     MemoryMessage,
     MemoryRecordResult,
@@ -509,11 +506,8 @@ async def test_working_memory_reconstruction_from_long_term(
 
     @pytest.mark.requires_api_keys
     @pytest.mark.asyncio
-    async def test_put_memory_stores_messages_in_long_term_memory(
-        self, client_with_mock_background_tasks, mock_background_tasks
-    ):
+    async def test_put_memory_stores_messages_in_long_term_memory(self, client):
         """Test the put_memory endpoint"""
-        client = client_with_mock_background_tasks
         payload = {
             "messages": [
                 {"role": "user", "content": "Hello"},
@@ -526,7 +520,13 @@ async def test_put_memory_stores_messages_in_long_term_memory(
         }
         mock_settings = Settings(long_term_memory=True)
 
-        with patch("agent_memory_server.api.settings", mock_settings):
+        # Mock the promote function to verify it's called
+        with (
+            patch("agent_memory_server.api.settings", mock_settings),
+            patch(
+                "agent_memory_server.api.long_term_memory.promote_working_memory_to_long_term"
+            ) as mock_promote,
+        ):
             response = await client.put("/v1/working-memory/test-session", json=payload)
 
         assert response.status_code == 200
@@ -537,22 +537,15 @@ async def test_put_memory_stores_messages_in_long_term_memory(
         assert "context" in data
         assert data["context"] == "Previous context"
 
-        # Check that background tasks were called
-        assert mock_background_tasks.add_task.call_count == 1
-
-        # Check that the last call was for long-term memory promotion
-        assert (
-            mock_background_tasks.add_task.call_args_list[-1][0][0]
-            == promote_working_memory_to_long_term
-        )
+        # Check that the promotion function was called as a background task
+        # In --no-worker mode, it runs inline, so it should have been called
+        assert mock_promote.call_count == 1
+        assert mock_promote.call_args[1]["session_id"] == "test-session"
 
     @pytest.mark.requires_api_keys
     @pytest.mark.asyncio
-    async def test_put_memory_with_structured_memories_triggers_promotion(
-        self, client_with_mock_background_tasks, mock_background_tasks
-    ):
+    async def test_put_memory_with_structured_memories_triggers_promotion(self, client):
         """Test that structured memories trigger background promotion task"""
-        client = client_with_mock_background_tasks
         payload = {
             "messages": [],
             "memories": [
@@ -569,7 +562,13 @@ async def test_put_memory_with_structured_memories_triggers_promotion(
         }
         mock_settings = Settings(long_term_memory=True)
 
-        with patch("agent_memory_server.api.settings", mock_settings):
+        # Mock the promote function to verify it's called
+        with (
+            patch("agent_memory_server.api.settings", mock_settings),
+            patch(
+                "agent_memory_server.api.long_term_memory.promote_working_memory_to_long_term"
+            ) as mock_promote,
+        ):
             response = await client.put("/v1/working-memory/test-session", json=payload)
 
         assert response.status_code == 200
@@ -579,20 +578,11 @@ async def test_put_memory_with_structured_memories_triggers_promotion(
         assert len(data["memories"]) == 1
         assert data["memories"][0]["text"] == "User prefers dark mode"
 
-        # Check that promotion background task was called
-        assert mock_background_tasks.add_task.call_count == 1
-
-        # Check that it was the promotion task, not indexing
-        assert (
-            mock_background_tasks.add_task.call_args_list[0][0][0]
-            == promote_working_memory_to_long_term
-        )
-
-        # Check the arguments passed to the promotion task
-        task_call = mock_background_tasks.add_task.call_args_list[0]
-        task_kwargs = task_call[1]
-        assert task_kwargs["session_id"] == "test-session"
-        assert task_kwargs["namespace"] == "test-namespace"
+        # Check that the promotion function was called as a background task
+        # In --no-worker mode, it runs inline, so it should have been called
+        assert mock_promote.call_count == 1
+        assert mock_promote.call_args[1]["session_id"] == "test-session"
+        assert mock_promote.call_args[1]["namespace"] == "test-namespace"
 
     @pytest.mark.requires_api_keys
     @pytest.mark.asyncio