feat: async memory support

Adds async support for tools with tests, async execution in the agent executor, and async operations for memory (with aiosqlite). Improves tool decorator typing, ensures _run backward compatibility, updates docs and docstrings, adds tests, and regenerates lockfiles.

Author: greysonlalonde

lib/crewai/pyproject.toml

@@ -38,6 +38,7 @@ dependencies = [
     "pydantic-settings~=2.10.1",
     "mcp~=1.16.0",
     "uv~=0.9.13",
+    "aiosqlite~=0.21.0",
 ]
 
 [project.urls]

lib/crewai/src/crewai/memory/contextual/contextual_memory.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from typing import TYPE_CHECKING
 
 from crewai.memory import (
@@ -16,6 +17,8 @@
 
 
 class ContextualMemory:
+    """Aggregates and retrieves context from multiple memory sources."""
+
     def __init__(
         self,
         stm: ShortTermMemory,
@@ -46,9 +49,14 @@ def __init__(
             self.exm.task = self.task
 
     def build_context_for_task(self, task: Task, context: str) -> str:
-        """
-        Automatically builds a minimal, highly relevant set of contextual information
-        for a given task.
+        """Build contextual information for a task synchronously.
+
+        Args:
+            task: The task to build context for.
+            context: Additional context string.
+
+        Returns:
+            Formatted context string from all memory sources.
         """
         query = f"{task.description} {context}".strip()
 
@@ -63,6 +71,31 @@ def build_context_for_task(self, task: Task, context: str) -> str:
         ]
         return "\n".join(filter(None, context_parts))
 
+    async def abuild_context_for_task(self, task: Task, context: str) -> str:
+        """Build contextual information for a task asynchronously.
+
+        Args:
+            task: The task to build context for.
+            context: Additional context string.
+
+        Returns:
+            Formatted context string from all memory sources.
+        """
+        query = f"{task.description} {context}".strip()
+
+        if query == "":
+            return ""
+
+        # Fetch all contexts concurrently
+        results = await asyncio.gather(
+            self._afetch_ltm_context(task.description),
+            self._afetch_stm_context(query),
+            self._afetch_entity_context(query),
+            self._afetch_external_context(query),
+        )
+
+        return "\n".join(filter(None, results))
+
     def _fetch_stm_context(self, query: str) -> str:
         """
         Fetches recent relevant insights from STM related to the task's description and expected_output,
@@ -135,3 +168,87 @@ def _fetch_external_context(self, query: str) -> str:
             f"- {result['content']}" for result in external_memories
         )
         return f"External memories:\n{formatted_memories}"
+
+    async def _afetch_stm_context(self, query: str) -> str:
+        """Fetch recent relevant insights from STM asynchronously.
+
+        Args:
+            query: The search query.
+
+        Returns:
+            Formatted insights as bullet points, or empty string if none found.
+        """
+        if self.stm is None:
+            return ""
+
+        stm_results = await self.stm.asearch(query)
+        formatted_results = "\n".join(
+            [f"- {result['content']}" for result in stm_results]
+        )
+        return f"Recent Insights:\n{formatted_results}" if stm_results else ""
+
+    async def _afetch_ltm_context(self, task: str) -> str | None:
+        """Fetch historical data from LTM asynchronously.
+
+        Args:
+            task: The task description to search for.
+
+        Returns:
+            Formatted historical data as bullet points, or None if none found.
+        """
+        if self.ltm is None:
+            return ""
+
+        ltm_results = await self.ltm.asearch(task, latest_n=2)
+        if not ltm_results:
+            return None
+
+        formatted_results = [
+            suggestion
+            for result in ltm_results
+            for suggestion in result["metadata"]["suggestions"]
+        ]
+        formatted_results = list(dict.fromkeys(formatted_results))
+        formatted_results = "\n".join([f"- {result}" for result in formatted_results])  # type: ignore # Incompatible types in assignment (expression has type "str", variable has type "list[str]")
+
+        return f"Historical Data:\n{formatted_results}" if ltm_results else ""
+
+    async def _afetch_entity_context(self, query: str) -> str:
+        """Fetch relevant entity information asynchronously.
+
+        Args:
+            query: The search query.
+
+        Returns:
+            Formatted entity information as bullet points, or empty string if none found.
+        """
+        if self.em is None:
+            return ""
+
+        em_results = await self.em.asearch(query)
+        formatted_results = "\n".join(
+            [f"- {result['content']}" for result in em_results]
+        )
+        return f"Entities:\n{formatted_results}" if em_results else ""
+
+    async def _afetch_external_context(self, query: str) -> str:
+        """Fetch relevant information from External Memory asynchronously.
+
+        Args:
+            query: The search query.
+
+        Returns:
+            Formatted information as bullet points, or empty string if none found.
+        """
+        if self.exm is None:
+            return ""
+
+        external_memories = await self.exm.asearch(query)
+
+        if not external_memories:
+            return ""
+
+        formatted_memories = "\n".join(
+            f"- {result['content']}" for result in external_memories
+        )
+        return f"External memories:\n{formatted_memories}"

lib/crewai/src/crewai/memory/entity/entity_memory.py

@@ -26,7 +26,13 @@ class EntityMemory(Memory):
 
     _memory_provider: str | None = PrivateAttr()
 
-    def __init__(self, crew=None, embedder_config=None, storage=None, path=None):
+    def __init__(
+        self,
+        crew: Any = None,
+        embedder_config: Any = None,
+        storage: Any = None,
+        path: str | None = None,
+    ) -> None:
         memory_provider = None
         if embedder_config and isinstance(embedder_config, dict):
             memory_provider = embedder_config.get("provider")
@@ -43,7 +49,7 @@ def __init__(self, crew=None, embedder_config=None, storage=None, path=None):
                 if embedder_config and isinstance(embedder_config, dict)
                 else None
             )
-            storage = Mem0Storage(type="short_term", crew=crew, config=config)
+            storage = Mem0Storage(type="short_term", crew=crew, config=config)  # type: ignore[no-untyped-call]
         else:
             storage = (
                 storage
@@ -170,7 +176,17 @@ def search(
         query: str,
         limit: int = 5,
         score_threshold: float = 0.6,
-    ):
+    ) -> list[Any]:
+        """Search entity memory for relevant entries.
+
+        Args:
+            query: The search query.
+            limit: Maximum number of results to return.
+            score_threshold: Minimum similarity score for results.
+
+        Returns:
+            List of matching memory entries.
+        """
         crewai_event_bus.emit(
             self,
             event=MemoryQueryStartedEvent(
@@ -217,6 +233,168 @@ def search(
             )
             raise
 
+    async def asave(
+        self,
+        value: EntityMemoryItem | list[EntityMemoryItem],
+        metadata: dict[str, Any] | None = None,
+    ) -> None:
+        """Save entity items asynchronously.
+
+        Args:
+            value: Single EntityMemoryItem or list of EntityMemoryItems to save.
+            metadata: Optional metadata dict (not used, for signature compatibility).
+        """
+        if not value:
+            return
+
+        items = value if isinstance(value, list) else [value]
+        is_batch = len(items) > 1
+
+        metadata = {"entity_count": len(items)} if is_batch else items[0].metadata
+        crewai_event_bus.emit(
+            self,
+            event=MemorySaveStartedEvent(
+                metadata=metadata,
+                source_type="entity_memory",
+                from_agent=self.agent,
+                from_task=self.task,
+            ),
+        )
+
+        start_time = time.time()
+        saved_count = 0
+        errors: list[str | None] = []
+
+        async def save_single_item(item: EntityMemoryItem) -> tuple[bool, str | None]:
+            """Save a single item asynchronously."""
+            try:
+                if self._memory_provider == "mem0":
+                    data = f"""
+                    Remember details about the following entity:
+                    Name: {item.name}
+                    Type: {item.type}
+                    Entity Description: {item.description}
+                    """
+                else:
+                    data = f"{item.name}({item.type}): {item.description}"
+
+                await super(EntityMemory, self).asave(data, item.metadata)
+                return True, None
+            except Exception as e:
+                return False, f"{item.name}: {e!s}"
+
+        try:
+            for item in items:
+                success, error = await save_single_item(item)
+                if success:
+                    saved_count += 1
+                else:
+                    errors.append(error)
+
+            if is_batch:
+                emit_value = f"Saved {saved_count} entities"
+                metadata = {"entity_count": saved_count, "errors": errors}
+            else:
+                emit_value = f"{items[0].name}({items[0].type}): {items[0].description}"
+                metadata = items[0].metadata
+
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveCompletedEvent(
+                    value=emit_value,
+                    metadata=metadata,
+                    save_time_ms=(time.time() - start_time) * 1000,
+                    source_type="entity_memory",
+                    from_agent=self.agent,
+                    from_task=self.task,
+                ),
+            )
+
+            if errors:
+                raise Exception(
+                    f"Partial save: {len(errors)} failed out of {len(items)}"
+                )
+
+        except Exception as e:
+            fail_metadata = (
+                {"entity_count": len(items), "saved": saved_count}
+                if is_batch
+                else items[0].metadata
+            )
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveFailedEvent(
+                    metadata=fail_metadata,
+                    error=str(e),
+                    source_type="entity_memory",
+                    from_agent=self.agent,
+                    from_task=self.task,
+                ),
+            )
+            raise
+
+    async def asearch(
+        self,
+        query: str,
+        limit: int = 5,
+        score_threshold: float = 0.6,
+    ) -> list[Any]:
+        """Search entity memory asynchronously.
+
+        Args:
+            query: The search query.
+            limit: Maximum number of results to return.
+            score_threshold: Minimum similarity score for results.
+
+        Returns:
+            List of matching memory entries.
+        """
+        crewai_event_bus.emit(
+            self,
+            event=MemoryQueryStartedEvent(
+                query=query,
+                limit=limit,
+                score_threshold=score_threshold,
+                source_type="entity_memory",
+                from_agent=self.agent,
+                from_task=self.task,
+            ),
+        )
+
+        start_time = time.time()
+        try:
+            results = await super().asearch(
+                query=query, limit=limit, score_threshold=score_threshold
+            )
+
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryCompletedEvent(
+                    query=query,
+                    results=results,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    query_time_ms=(time.time() - start_time) * 1000,
+                    source_type="entity_memory",
+                    from_agent=self.agent,
+                    from_task=self.task,
+                ),
+            )
+
+            return results
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryFailedEvent(
+                    query=query,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    error=str(e),
+                    source_type="entity_memory",
+                ),
+            )
+            raise
+
     def reset(self) -> None:
         try:
             self.storage.reset()