Merge branch 'main' into tm-multiple-mcp-servers-crew-base

.gitignore

@@ -26,4 +26,5 @@ test_flow.html
 crewairules.mdc
 plan.md
 conceptual_plan.md
-build_image
+build_image
+chromadb-*.lock

docs/en/observability/neatlogs.mdx

@@ -10,8 +10,6 @@ Neatlogs helps you **see what your agent did**, **why**, and **share it**.
 
 It captures every step: thoughts, tool calls, responses, evaluations. No raw logs. Just clear, structured traces. Great for debugging and collaboration.
 
----
-
 ## Why use Neatlogs?
 
 CrewAI agents use multiple tools and reasoning steps. When something goes wrong, you need context — not just errors.
@@ -37,8 +35,6 @@ The best UX to view a CrewAI trace. Post comments anywhere you want. Use AI to d
 ![Ai Chat Bot With A Trace](/images/neatlogs-4.png)
 ![Comments Drawer](/images/neatlogs-5.png)
 
----
-
 ## Core Features
 
 - **Trace Viewer**: Track thoughts, tools, and decisions in sequence
@@ -49,8 +45,6 @@ The best UX to view a CrewAI trace. Post comments anywhere you want. Use AI to d
 - **Ask the Trace (AI)**: Chat with your trace using Neatlogs AI bot
 - **Public Sharing**: Publish trace links to your community
 
----
-
 ## Quick Setup with CrewAI
 
 <Steps>
@@ -61,7 +55,7 @@ The best UX to view a CrewAI trace. Post comments anywhere you want. Use AI to d
     ```bash
     pip install neatlogs
     ```
-    (Latest version 0.8.0, Python 3.8+; MIT license) :contentReference[oaicite:1]{index=1}
+    (Latest version 0.8.0, Python 3.8+; MIT license)
   </Step>
   <Step title="Initialize Neatlogs">
     Before starting Crew agents, add:
@@ -76,18 +70,18 @@ The best UX to view a CrewAI trace. Post comments anywhere you want. Use AI to d
   </Step>
 </Steps>
 
----
+
 
 ## Under the Hood
 
 According to GitHub, Neatlogs:
 
-- Captures thoughts, tool calls, responses, errors, and token stats :contentReference[oaicite:2]{index=2}
+- Captures thoughts, tool calls, responses, errors, and token stats
-- Supports AI-powered task generation and robust evaluation workflows :contentReference[oaicite:3]{index=3}
+- Supports AI-powered task generation and robust evaluation workflows
 
 All with just two lines of code.
 
----
+
 
 ## Watch It Work
 
@@ -113,7 +107,7 @@ All with just two lines of code.
   allowFullScreen
 ></iframe>
 
----
+
 
 ## Links & Support
 
@@ -121,9 +115,9 @@ All with just two lines of code.
 - 🔐 [Dashboard & API Key](https://app.neatlogs.com/)
 - 🐦 [Follow on Twitter](https://twitter.com/neatlogs)
 - 📧 Contact: hello@neatlogs.com
-- 🛠 [GitHub SDK](https://github.com/NeatLogs/neatlogs) :contentReference[oaicite:4]{index=4}
+- 🛠 [GitHub SDK](https://github.com/NeatLogs/neatlogs)
+
 
----
 
 ## TL;DR
 

pyproject.toml

@@ -39,6 +39,7 @@ dependencies = [
     "tomli>=2.0.2",
     "blinker>=1.9.0",
     "json5>=0.10.0",
+    "portalocker==2.7.0",
 ]
 
 [project.urls]

src/crewai/knowledge/storage/knowledge_storage.py

@@ -18,6 +18,7 @@ from crewai.utilities.chromadb import sanitize_collection_name
 from crewai.utilities.constants import KNOWLEDGE_DIRECTORY
 from crewai.utilities.logger import Logger
 from crewai.utilities.paths import db_storage_path
+from crewai.utilities.chromadb import create_persistent_client
 
 
 @contextlib.contextmanager
@@ -84,14 +85,11 @@ class KnowledgeStorage(BaseKnowledgeStorage):
                 raise Exception("Collection not initialized")
 
     def initialize_knowledge_storage(self):
-        base_path = os.path.join(db_storage_path(), "knowledge")
+        self.app = create_persistent_client(
-        chroma_client = chromadb.PersistentClient(
+            path=os.path.join(db_storage_path(), "knowledge"),
-            path=base_path,
             settings=Settings(allow_reset=True),
         )
 
-        self.app = chroma_client
-
         try:
             collection_name = (
                 f"knowledge_{self.collection_name}"
@@ -111,9 +109,8 @@ class KnowledgeStorage(BaseKnowledgeStorage):
     def reset(self):
         base_path = os.path.join(db_storage_path(), KNOWLEDGE_DIRECTORY)
         if not self.app:
-            self.app = chromadb.PersistentClient(
+            self.app = create_persistent_client(
-                path=base_path,
+                path=base_path, settings=Settings(allow_reset=True)
-                settings=Settings(allow_reset=True),
             )
 
         self.app.reset()

src/crewai/memory/storage/rag_storage.py

@@ -4,12 +4,12 @@ import logging
 import os
 import shutil
 import uuid
+
 from typing import Any, Dict, List, Optional
-
 from chromadb.api import ClientAPI
-
 from crewai.memory.storage.base_rag_storage import BaseRAGStorage
 from crewai.utilities import EmbeddingConfigurator
+from crewai.utilities.chromadb import create_persistent_client
 from crewai.utilities.constants import MAX_FILE_NAME_LENGTH
 from crewai.utilities.paths import db_storage_path
 
@@ -60,17 +60,15 @@ class RAGStorage(BaseRAGStorage):
         self.embedder_config = configurator.configure_embedder(self.embedder_config)
 
     def _initialize_app(self):
-        import chromadb
         from chromadb.config import Settings
 
         self._set_embedder_config()
-        chroma_client = chromadb.PersistentClient(
+
+        self.app = create_persistent_client(
             path=self.path if self.path else self.storage_file_name,
             settings=Settings(allow_reset=self.allow_reset),
         )
 
-        self.app = chroma_client
-
         self.collection = self.app.get_or_create_collection(
             name=self.type, embedding_function=self.embedder_config
         )

src/crewai/utilities/chromadb.py

@@ -1,6 +1,10 @@
 import re
+import portalocker
+from chromadb import PersistentClient
+from hashlib import md5
 from typing import Optional
 
+
 MIN_COLLECTION_LENGTH = 3
 MAX_COLLECTION_LENGTH = 63
 DEFAULT_COLLECTION = "default_collection"
@@ -60,3 +64,16 @@ def sanitize_collection_name(name: Optional[str], max_collection_length: int = M
             sanitized = sanitized[:-1] + "z"
 
     return sanitized
+
+
+def create_persistent_client(path: str, **kwargs):
+    """
+    Creates a persistent client for ChromaDB with a lock file to prevent
+    concurrent creations. Works for both multi-threads and multi-processes
+    environments.
+    """
+    lockfile = f"chromadb-{md5(path.encode(), usedforsecurity=False).hexdigest()}.lock"
+    with portalocker.Lock(lockfile):
+        client = PersistentClient(path=path, **kwargs)
+
+    return client

tests/utilities/test_chromadb_utils.py

@@ -1,16 +1,27 @@
+import multiprocessing
+import tempfile
 import unittest
-from typing import Any, Dict, List, Union
 
-import pytest
+from chromadb.config import Settings
+from unittest.mock import patch, MagicMock
 
 from crewai.utilities.chromadb import (
     MAX_COLLECTION_LENGTH,
     MIN_COLLECTION_LENGTH,
     is_ipv4_pattern,
     sanitize_collection_name,
+    create_persistent_client,
 )
 
 
+def persistent_client_worker(path, queue):
+    try:
+        create_persistent_client(path=path)
+        queue.put(None)
+    except Exception as e:
+        queue.put(e)
+
+
 class TestChromadbUtils(unittest.TestCase):
     def test_sanitize_collection_name_long_name(self):
         """Test sanitizing a very long collection name."""
@@ -79,3 +90,34 @@ class TestChromadbUtils(unittest.TestCase):
             self.assertLessEqual(len(sanitized), MAX_COLLECTION_LENGTH)
             self.assertTrue(sanitized[0].isalnum())
             self.assertTrue(sanitized[-1].isalnum())
+
+    def test_create_persistent_client_passes_args(self):
+        with patch(
+            "crewai.utilities.chromadb.PersistentClient"
+        ) as mock_persistent_client, tempfile.TemporaryDirectory() as tmpdir:
+            mock_instance = MagicMock()
+            mock_persistent_client.return_value = mock_instance
+
+            settings = Settings(allow_reset=True)
+            client = create_persistent_client(path=tmpdir, settings=settings)
+
+            mock_persistent_client.assert_called_once_with(
+                path=tmpdir, settings=settings
+            )
+            self.assertIs(client, mock_instance)
+
+    def test_create_persistent_client_process_safe(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            queue = multiprocessing.Queue()
+            processes = [
+                multiprocessing.Process(
+                    target=persistent_client_worker, args=(tmpdir, queue)
+                )
+                for _ in range(5)
+            ]
+
+            [p.start() for p in processes]
+            [p.join() for p in processes]
+
+            errors = [queue.get(timeout=5) for _ in processes]
+            self.assertTrue(all(err is None for err in errors))