Add initial doc part of keyed mcp server params

This commit fixes a bug with concurrent processess and Chroma where
`table collections already exists` (and similar) were raised.

https://cookbook.chromadb.dev/core/system_constraints/

.gitignore

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

docs/en/mcp/overview.mdx

@@ -172,6 +172,60 @@ def another_agent(self):
     )
 ```
 
+### Using Multiple MCP Servers with CrewBase
+
+You can configure multiple MCP servers and assign different servers and tools to different agents. Use a dictionary to pass multiple named MCP servers.
+
+```python
+@CrewBase
+class CrewWithMultipleMCP:
+  # ... define your agents and tasks config file ...
+
+  # MCP servers keyed by server name
+  mcp_server_params = {
+      "web_tools": {
+        "url": "http://localhost:8000/mcp",
+        "transport": "streamable-http"
+      },
+      "data_tools": {
+        "url":
+        "http://localhost:8001/sse",
+        "transport": "sse"
+      },
+      "local_tools": StdioServerParameters(
+          command="python3",
+          args=["servers/local_server.py"],
+          env={"UV_PYTHON": "3.12", **os.environ},
+      )
+  }
+
+  @agent
+  def web_researcher(self):
+      # Use tools from specific server
+      return Agent(
+          config=self.agents_config["web_researcher"],
+          tools=self.get_mcp_tools(server="web_tools")
+      )
+
+  @agent
+  def data_analyst(self):
+      # Use specific tools from specific server
+      return Agent(
+          config=self.agents_config["data_analyst"],
+          tools=self.get_mcp_tools("analyze_csv", "create_chart", server="data_tools")
+      )
+
+  @agent
+  def multi_tool_agent(self):
+      # Use tools from all servers
+      return Agent(
+          config=self.agents_config["multi_tool_agent"],
+          tools=self.get_mcp_tools()  # No server specified = all tools
+      )
+
+  # ... rest of your crew setup ...
+```
+
 ## Explore MCP Integrations
 
 <CardGroup cols={2}>

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}
-- Supports AI-powered task generation and robust evaluation workflows :contentReference[oaicite:3]{index=3}
+- Captures thoughts, tool calls, responses, errors, and token stats
+- 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")
-        chroma_client = chromadb.PersistentClient(
-            path=base_path,
+        self.app = create_persistent_client(
+            path=os.path.join(db_storage_path(), "knowledge"),
             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(
-                path=base_path,
-                settings=Settings(allow_reset=True),
+            self.app = create_persistent_client(
+                path=base_path, 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/project/crew_base.py

@@ -1,7 +1,7 @@
 import inspect
 import logging
 from pathlib import Path
-from typing import Any, Callable, Dict, TypeVar, cast, List
+from typing import Any, Callable, Dict, TypeVar, cast, List, Union
 from crewai.tools import BaseTool
 
 import yaml
@@ -28,7 +28,8 @@ def CrewBase(cls: T) -> T:
         )
         original_tasks_config_path = getattr(cls, "tasks_config", "config/tasks.yaml")
 
-        mcp_server_params: Any = getattr(cls, "mcp_server_params", None)
+        mcp_server_params: Union[list[str | dict[str, str]], dict[str, str], None] = getattr(cls, "mcp_server_params", None)
+        _mcp_server_adapter: Union[dict[str, Any], Any, None] = None
 
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
@@ -67,36 +68,57 @@ def CrewBase(cls: T) -> T:
                 self._original_functions, "is_kickoff"
             )
 
-            # Add close mcp server method to after kickoff
-            bound_method = self._create_close_mcp_server_method()
-            self._after_kickoff['_close_mcp_server'] = bound_method
+            # Add close mcp servers method to after kickoff
+            bound_method = self._create_close_mcp_servers_method()
+            self._after_kickoff['_close_mcp_servers'] = bound_method
 
-        def _create_close_mcp_server_method(self):
-            def _close_mcp_server(self, instance, outputs):
-                adapter = getattr(self, '_mcp_server_adapter', None)
-                if adapter is not None:
+        def _create_close_mcp_servers_method(self):
+            def _close_mcp_servers(self, instance, outputs):
+                if self._mcp_server_adapter is None:
+                    return outputs
+                for adapter in self._mcp_server_adapter.values():
                     try:
                         adapter.stop()
                     except Exception as e:
                         logging.warning(f"Error stopping MCP server: {e}")
                 return outputs
 
-            _close_mcp_server.is_after_kickoff = True
+            _close_mcp_servers.is_after_kickoff = True
 
             import types
-            return types.MethodType(_close_mcp_server, self)
+            return types.MethodType(_close_mcp_servers, self)
 
-        def get_mcp_tools(self, *tool_names: list[str]) -> List[BaseTool]:
+        def get_mcp_tools(self, *tool_names: list[str], server: str | None = None) -> List[BaseTool]:
             if not self.mcp_server_params:
                 return []
 
             from crewai_tools import MCPServerAdapter
 
-            adapter = getattr(self, '_mcp_server_adapter', None)
-            if not adapter:
-                self._mcp_server_adapter = MCPServerAdapter(self.mcp_server_params)
+            if isinstance(self.mcp_server_params, list):
+                if self._mcp_server_adapter is None:
+                    self._mcp_server_adapter = MCPServerAdapter(self.mcp_server_params)
+                if server is not None and len(self.mcp_server_params) > 1:
+                    logging.warning("Using list of MCP server parameters. To use server parameter, please use a dictionary of MCP server parameters.")
+                # Type assertion: when mcp_server_params is a list, _mcp_server_adapter is a single MCPServerAdapter
+                adapter = cast(Any, self._mcp_server_adapter)
+                return adapter.tools.filter_by_names(tool_names or None)
 
-            return self._mcp_server_adapter.tools.filter_by_names(tool_names or None)
+            # Separated MCP adapters for each server.
+            elif isinstance(self.mcp_server_params, dict):
+                if self._mcp_server_adapter is None:
+                    self._mcp_server_adapter = {}
+                aggregated_tools = []
+                for server_name, params in self.mcp_server_params.items():
+                    if server is not None and server_name != server:
+                        continue
+
+                    adapter = self._mcp_server_adapter.get(server_name, None)
+                    if not adapter:
+                        self._mcp_server_adapter[server_name] = MCPServerAdapter(params)
+                    aggregated_tools.extend(
+                        self._mcp_server_adapter[server_name].tools.filter_by_names(tool_names or None))
+
+            return aggregated_tools
 
 
         def load_configurations(self):

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/project_test.py

@@ -87,7 +87,7 @@ class InternalCrew:
 
 @CrewBase
 class InternalCrewWithMCP(InternalCrew):
-    mcp_server_params = {"host": "localhost", "port": 8000}
+    mcp_server_params = [{"url": "localhost", "port": 8000}]
 
     @agent
     def reporting_analyst(self):
@@ -97,6 +97,19 @@ class InternalCrewWithMCP(InternalCrew):
     def researcher(self):
         return Agent(config=self.agents_config["researcher"], tools=self.get_mcp_tools("simple_tool"))  # type: ignore[index]
 
+@CrewBase
+class InternalCrewWithMultipleMCP(InternalCrew):
+    mcp_server_params = {"mcp1": {"url": "localhost", "port": 8000}, "mcp2": {"url": "localhost", "port": 8001}}
+
+    @agent
+    def reporting_analyst(self):
+        return Agent(config=self.agents_config["reporting_analyst"], tools=self.get_mcp_tools(server="mcp1"))  # type: ignore[index]
+
+    @agent
+    def researcher(self):
+        return Agent(config=self.agents_config["researcher"], tools=self.get_mcp_tools("simple_tool", server="mcp2"))  # type: ignore[index]
+
+
 def test_agent_memoization():
     crew = SimpleCrew()
     first_call_result = crew.simple_agent()
@@ -270,4 +283,21 @@ def test_internal_crew_with_mcp():
         assert crew.reporting_analyst().tools == [simple_tool, another_simple_tool]
         assert crew.researcher().tools == [simple_tool]
 
-    adapter_mock.assert_called_once_with({"host": "localhost", "port": 8000})
+    adapter_mock.assert_called_once_with([{"url": "localhost", "port": 8000}])
+
+
+def test_internal_crew_with_multiple_mcp():
+    from crewai_tools import MCPServerAdapter
+    from crewai_tools.adapters.mcp_adapter import ToolCollection
+    from unittest.mock import call
+
+    mock = Mock(spec=MCPServerAdapter)
+    mock.tools = ToolCollection([simple_tool, another_simple_tool])
+    with patch("crewai_tools.MCPServerAdapter", return_value=mock) as adapter_mock:
+        crew = InternalCrewWithMultipleMCP()
+        assert crew.reporting_analyst().tools == [simple_tool, another_simple_tool]
+        assert crew.researcher().tools == [simple_tool]
+        adapter_mock.assert_has_calls([
+            call({"url": "localhost", "port": 8000}),
+            call({"url": "localhost", "port": 8001})
+        ], any_order=True)

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))