fix(mcp): first-wins tool routing to stop silent cross-server name shadowing

CHANGELOG.md

@@ -3,6 +3,32 @@
 All notable changes to this project are documented here. This project follows
 [Semantic Versioning](https://semver.org/).
 
+## [0.26.0]
+
+### Security
+
+- **MCP tool-name shadowing fixed.** When multiple MCP servers advertised the
+  same tool name, `StreamManager` used last-writer-wins with no warning, so a
+  later server (malicious, compromised, or merely reusing a common name like
+  `read_file`) silently captured every future unpinned `call_tool` for that name.
+  Registration is now **first-wins**: the first server to advertise a name owns
+  default routing, a colliding later server is ignored for routing and logged
+  with a prominent warning, and the shadowed tool remains reachable only by
+  passing `server_name=` explicitly.
+
+### Added
+
+- `StreamManager.get_servers_for_tool(name)` and
+  `StreamManager.get_tool_collisions()` to inspect which servers advertise a tool
+  name and surface cross-server name collisions.
+
+### Fixed
+
+- `SubprocessStrategy.shutdown()` now shuts the process pool down directly instead
+  of offloading `pool.shutdown(wait=False)` (already non-blocking) to a thread with
+  a 1s timeout. On a busy event loop the executor could miss the timeout and skip
+  the shutdown entirely — an intermittent test failure and a real pool-leak risk.
+
 ## [0.25.0]
 
 ### Security

docs/MCP_INTEGRATION.md

@@ -198,6 +198,37 @@ processor, manager = await setup_mcp_sse(
 
 ---
 
+## Tool Name Collisions
+
+When you connect several MCP servers, two of them can advertise the **same tool
+name** (common short names like `read_file`, `search`, `query` collide easily). A
+bare tool name is not a trust boundary, so `StreamManager` uses a **first-wins**
+policy: the first server to advertise a name owns it for default (unpinned)
+routing, and a later server advertising the same name is **ignored for routing
+and logged with a warning** — it can never silently take over calls that were
+going to an already-connected server.
+
+```python
+sm = StreamManager()
+await sm.initialize(servers=[...])   # trusted-fs, then community-server
+
+# trusted-fs registered read_file first, so it owns the name:
+await sm.call_tool("read_file", {"path": "..."})            # -> trusted-fs
+# reach the shadowed copy deliberately by pinning the server:
+await sm.call_tool("read_file", {"path": "..."}, server_name="community-server")
+```
+
+Inspect collisions so you can namespace or reconfigure:
+
+```python
+sm.get_server_for_tool("read_file")    # "trusted-fs"  (owns default routing)
+sm.get_servers_for_tool("read_file")   # ["trusted-fs", "community-server"]
+sm.get_tool_collisions()               # {"read_file": ["trusted-fs", "community-server"]}
+```
+
+To avoid collisions entirely, give each server a distinct `namespace=` in the
+`setup_mcp_*` wrappers (e.g. `notion.search`, `db.query`) so names never clash.
+
 ## Middleware Stack
 
 The `MiddlewareStack` provides production-grade resilience for MCP tool calls. It wraps MCP connections with configurable retry, circuit breaker, and rate limiting layers.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "chuk-tool-processor"
-version = "0.25.0"
+version = "0.26.0"
 description = "Async-native framework for registering, discovering, and executing tools referenced in LLM responses"
 readme = "README.md"
 requires-python = ">=3.11"

src/chuk_tool_processor/execution/strategies/subprocess_strategy.py

@@ -767,30 +767,21 @@ async def shutdown(self) -> None:
             except Exception:
                 logger.debug("Active operations completed successfully")
 
-        # Handle process pool shutdown with proper null checks
+        # Handle process pool shutdown with proper null checks.
         if self._process_pool is not None:
             logger.debug("Finalizing process pool")
+            # Store reference and clear immediately to prevent race conditions.
+            pool_to_shutdown = self._process_pool
+            self._process_pool = None
+            # shutdown(wait=False) is non-blocking (it doesn't join worker
+            # processes), so call it directly — the same way the other cleanup
+            # paths in this module do. Offloading it to a thread-pool executor
+            # with a timeout previously raced: on a busy loop the executor could
+            # fail to run before the timeout fired, so the pool was never told to
+            # shut down at all.
             try:
-                # Store reference and null check before async operation
-                pool_to_shutdown = self._process_pool
-                self._process_pool = None  # Clear immediately to prevent race conditions
-
-                # Create shutdown task with the stored reference
-                shutdown_task = asyncio.create_task(
-                    asyncio.get_event_loop().run_in_executor(
-                        None, lambda: pool_to_shutdown.shutdown(wait=False) if pool_to_shutdown else None
-                    )
-                )
-
-                try:
-                    await asyncio.wait_for(shutdown_task, timeout=1.0)
-                    logger.debug("Process pool shutdown completed")
-                except TimeoutError:
-                    logger.debug("Process pool shutdown timed out, forcing cleanup")
-                    if not shutdown_task.done():
-                        shutdown_task.cancel()
-                except Exception as e:
-                    logger.debug(f"Process pool shutdown completed with warning: {e}")
+                pool_to_shutdown.shutdown(wait=False)
+                logger.debug("Process pool shutdown completed")
             except Exception as e:
                 logger.debug(f"Process pool finalization completed: {e}")
         else:

src/chuk_tool_processor/mcp/stream_manager.py

@@ -71,7 +71,13 @@ def __init__(
     ) -> None:
         self.transports: dict[str, MCPBaseTransport] = {}
         self.server_info: list[ServerInfo] = []
+        # Default routing map (tool name -> server). First server to advertise a
+        # name owns it; see _register_tools for the first-wins collision policy.
         self.tool_to_server_map: dict[str, str] = {}
+        # Every server that advertised each tool name, in registration order.
+        # Lets callers detect/resolve name collisions instead of silently
+        # inheriting whichever server registered last.
+        self.tool_to_servers: dict[str, list[str]] = {}
         self.server_names: dict[int, str] = {}
         self.all_tools: list[MCPToolDefinition] = []
         self._lock = asyncio.Lock()
@@ -334,9 +340,7 @@ async def initialize(
                     raw_tools = await asyncio.wait_for(transport.get_tools(), timeout=self.timeout_config.operation)
                     tools = [MCPToolDefinition.model_validate(t) for t in raw_tools]
 
-                    for t in tools:
-                        if t.name:
-                            self.tool_to_server_map[t.name] = server_name
+                    self._register_tools(tools, server_name)
                     self.all_tools.extend(tools)
 
                     self.server_info.append(ServerInfo(id=idx, name=server_name, tools=len(tools), status=status))
@@ -413,9 +417,7 @@ async def initialize_with_sse(
                     raw_tools = await asyncio.wait_for(transport.get_tools(), timeout=self.timeout_config.operation)
                     tools = [MCPToolDefinition.model_validate(t) for t in raw_tools]
 
-                    for t in tools:
-                        if t.name:
-                            self.tool_to_server_map[t.name] = name
+                    self._register_tools(tools, name)
                     self.all_tools.extend(tools)
 
                     self.server_info.append(ServerInfo(id=idx, name=name, tools=len(tools), status=status))
@@ -489,9 +491,7 @@ async def initialize_with_stdio(
                     raw_tools = await asyncio.wait_for(transport.get_tools(), timeout=self.timeout_config.operation)
                     tools = [MCPToolDefinition.model_validate(t) for t in raw_tools]
 
-                    for t in tools:
-                        if t.name:
-                            self.tool_to_server_map[t.name] = name
+                    self._register_tools(tools, name)
                     self.all_tools.extend(tools)
 
                     self.server_info.append(ServerInfo(id=idx, name=name, tools=len(tools), status=status))
@@ -575,9 +575,7 @@ async def initialize_with_http_streamable(
                     raw_tools = await asyncio.wait_for(transport.get_tools(), timeout=self.timeout_config.operation)
                     tools = [MCPToolDefinition.model_validate(t) for t in raw_tools]
 
-                    for t in tools:
-                        if t.name:
-                            self.tool_to_server_map[t.name] = name
+                    self._register_tools(tools, name)
                     self.all_tools.extend(tools)
 
                     self.server_info.append(ServerInfo(id=idx, name=name, tools=len(tools), status=status))
@@ -599,9 +597,58 @@ async def initialize_with_http_streamable(
     def get_all_tools(self) -> list[dict[str, Any]]:
         return [t.model_dump() for t in self.all_tools]
 
+    def _register_tools(self, tools: list[MCPToolDefinition], server_name: str) -> None:
+        """Map tool names to their owning server using a first-wins policy.
+
+        The first server to advertise a given tool name owns it for default
+        (unpinned) routing. If a later server advertises the same name, we keep
+        the original owner and log a prominent warning rather than silently
+        rerouting every future call to the newcomer — a bare tool name is not a
+        trust boundary, so a second server (malicious, compromised, or merely
+        reusing a common name like ``read_file``) must not be able to hijack a
+        name an earlier, trusted server already provides. Callers can still reach
+        the shadowed tool deliberately via ``call_tool(name, server_name=...)``.
+        """
+        for t in tools:
+            if not t.name:
+                continue
+            providers = self.tool_to_servers.setdefault(t.name, [])
+            if server_name not in providers:
+                providers.append(server_name)
+            owner = self.tool_to_server_map.get(t.name)
+            if owner is None:
+                self.tool_to_server_map[t.name] = server_name
+            elif owner != server_name:
+                logger.warning(
+                    "MCP tool name collision: '%s' is already provided by server '%s'; "
+                    "ignoring the tool of the same name from server '%s' for default routing. "
+                    "Call it explicitly with server_name='%s' if you meant that server.",
+                    t.name,
+                    owner,
+                    server_name,
+                    server_name,
+                )
+
     def get_server_for_tool(self, tool_name: str) -> str | None:
         return self.tool_to_server_map.get(tool_name)
 
+    def get_servers_for_tool(self, tool_name: str) -> list[str]:
+        """All servers that advertised ``tool_name``, in registration order.
+
+        More than one entry means the name is shadowed; only the first owns
+        default routing (see :meth:`_register_tools`).
+        """
+        return list(self.tool_to_servers.get(tool_name, []))
+
+    def get_tool_collisions(self) -> dict[str, list[str]]:
+        """Tool names advertised by more than one server, mapped to those servers.
+
+        Empty when there are no collisions. The first server in each list is the
+        one that receives unpinned calls; the rest are reachable only by passing
+        ``server_name`` explicitly to :meth:`call_tool`.
+        """
+        return {name: list(servers) for name, servers in self.tool_to_servers.items() if len(servers) > 1}
+
     def get_server_info(self) -> list[dict[str, Any]]:
         return [s.model_dump() for s in self.server_info]
 
@@ -1045,6 +1092,7 @@ def _cleanup_state(self) -> None:
             self.transports.clear()
             self.server_info.clear()
             self.tool_to_server_map.clear()
+            self.tool_to_servers.clear()
             self.all_tools.clear()
             self.server_names.clear()
         except Exception as e:

tests/mcp/test_stream_manager_shadowing.py

@@ -0,0 +1,100 @@
+"""Regression tests for MCP tool-name shadowing across servers.
+
+A bare tool name is not a trust boundary: if two configured MCP servers advertise
+the same tool name, the earlier (already-trusted) server must keep default
+routing, and the collision must be visible — never a silent last-writer-wins
+hijack of every future unpinned call.
+"""
+
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+from chuk_tool_processor.mcp.stream_manager import StreamManager
+from chuk_tool_processor.mcp.transport.models import MCPToolDefinition
+
+
+class _FakeTransport:
+    def __init__(self, label: str) -> None:
+        self.label = label
+
+    async def call_tool(self, name, arguments):
+        return {"content": [{"type": "text", "text": f"from {self.label}"}]}
+
+
+def _tools(*names: str) -> list[MCPToolDefinition]:
+    return [MCPToolDefinition(name=n, description=n) for n in names]
+
+
+def _manager_with(server_label_map: dict[str, str]) -> StreamManager:
+    sm = StreamManager()
+    for server, label in server_label_map.items():
+        sm.transports[server] = _FakeTransport(label)
+    return sm
+
+
+class TestToolShadowing:
+    def test_first_server_owns_default_routing(self):
+        sm = _manager_with({"trusted": "T", "later": "L"})
+        sm._register_tools(_tools("read_file"), "trusted")
+        sm._register_tools(_tools("read_file"), "later")
+        # First-wins: the trusted server keeps the name for unpinned routing.
+        assert sm.get_server_for_tool("read_file") == "trusted"
+        assert sm.get_servers_for_tool("read_file") == ["trusted", "later"]
+        assert sm.get_tool_collisions() == {"read_file": ["trusted", "later"]}
+
+    def test_collision_logs_warning(self, caplog):
+        sm = _manager_with({"trusted": "T", "later": "L"})
+        sm._register_tools(_tools("read_file"), "trusted")
+        with caplog.at_level(logging.WARNING):
+            sm._register_tools(_tools("read_file"), "later")
+        assert any("collision" in r.getMessage() and "read_file" in r.getMessage() for r in caplog.records), (
+            "expected a collision warning"
+        )
+
+    def test_no_warning_without_collision(self, caplog):
+        sm = _manager_with({"a": "A", "b": "B"})
+        with caplog.at_level(logging.WARNING):
+            sm._register_tools(_tools("read_file"), "a")
+            sm._register_tools(_tools("search"), "b")
+        assert sm.get_tool_collisions() == {}
+        assert not [r for r in caplog.records if "collision" in r.getMessage()]
+
+    def test_reregistering_same_server_is_noop(self, caplog):
+        sm = _manager_with({"a": "A"})
+        sm._register_tools(_tools("read_file"), "a")
+        with caplog.at_level(logging.WARNING):
+            sm._register_tools(_tools("read_file"), "a")  # e.g. reconnect
+        assert sm.get_servers_for_tool("read_file") == ["a"]
+        assert not [r for r in caplog.records if "collision" in r.getMessage()]
+
+    def test_tools_without_name_are_skipped(self):
+        sm = _manager_with({"a": "A"})
+        sm._register_tools([MCPToolDefinition(name="", description="x")], "a")
+        assert sm.tool_to_server_map == {}
+        assert sm.tool_to_servers == {}
+
+    @pytest.mark.asyncio
+    async def test_unpinned_call_stays_with_first_server(self):
+        sm = _manager_with({"trusted": "TRUSTED", "later": "UNTRUSTED"})
+        sm._register_tools(_tools("read_file"), "trusted")
+        sm._register_tools(_tools("read_file"), "later")
+        result = await sm.call_tool("read_file", {})
+        assert result["content"][0]["text"] == "from TRUSTED"
+
+    @pytest.mark.asyncio
+    async def test_pinned_call_reaches_shadowed_server(self):
+        sm = _manager_with({"trusted": "TRUSTED", "later": "UNTRUSTED"})
+        sm._register_tools(_tools("read_file"), "trusted")
+        sm._register_tools(_tools("read_file"), "later")
+        result = await sm.call_tool("read_file", {}, server_name="later")
+        assert result["content"][0]["text"] == "from UNTRUSTED"
+
+    def test_reset_clears_collision_map(self):
+        sm = _manager_with({"a": "A"})
+        sm._register_tools(_tools("read_file"), "a")
+        sm._sync_cleanup()
+        assert sm.tool_to_servers == {}
+        assert sm.tool_to_server_map == {}