From 14f659105c20f5694fc796d39c08679d1006aabd Mon Sep 17 00:00:00 2001 From: chris hay Date: Wed, 29 Jul 2026 23:42:00 +0100 Subject: [PATCH 1/2] fix(mcp): first-wins tool routing to stop silent cross-server name shadowing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StreamManager mapped each MCP tool name to whichever server registered it last, with no collision check. A second server advertising a name already provided by an earlier server (malicious, compromised, or just reusing a common name like read_file) silently captured every future unpinned call_tool for that name — no warning, leaking arguments to and returning results from the wrong server. 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; the shadowed tool stays reachable via call_tool(name, server_name=...). Adds get_servers_for_tool() and get_tool_collisions() to inspect collisions. Bumps to 0.26.0. Signed-off-by: chris hay --- CHANGELOG.md | 19 ++++ docs/MCP_INTEGRATION.md | 31 ++++++ pyproject.toml | 2 +- src/chuk_tool_processor/mcp/stream_manager.py | 72 ++++++++++--- tests/mcp/test_stream_manager_shadowing.py | 100 ++++++++++++++++++ uv.lock | 2 +- 6 files changed, 212 insertions(+), 14 deletions(-) create mode 100644 tests/mcp/test_stream_manager_shadowing.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c946c2..0ee6187 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,25 @@ 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. + ## [0.25.0] ### Security diff --git a/docs/MCP_INTEGRATION.md b/docs/MCP_INTEGRATION.md index c7481bd..275e604 100644 --- a/docs/MCP_INTEGRATION.md +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 505637e..12da9a0 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/chuk_tool_processor/mcp/stream_manager.py b/src/chuk_tool_processor/mcp/stream_manager.py index 7481763..1e7f51e 100644 --- a/src/chuk_tool_processor/mcp/stream_manager.py +++ b/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: diff --git a/tests/mcp/test_stream_manager_shadowing.py b/tests/mcp/test_stream_manager_shadowing.py new file mode 100644 index 0000000..00bf621 --- /dev/null +++ b/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 == {} diff --git a/uv.lock b/uv.lock index 23ee08c..3344d3d 100644 --- a/uv.lock +++ b/uv.lock @@ -212,7 +212,7 @@ wheels = [ [[package]] name = "chuk-tool-processor" -version = "0.25.0" +version = "0.26.0" source = { editable = "." } dependencies = [ { name = "chuk-mcp" }, From 596b3d045c71d17f747bce923cb4e162896fb18a Mon Sep 17 00:00:00 2001 From: chris hay Date: Thu, 30 Jul 2026 11:25:24 +0100 Subject: [PATCH 2/2] fix(subprocess): shut process pool down directly to remove flaky shutdown race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit shutdown() offloaded pool.shutdown(wait=False) to a thread-pool executor guarded by a 1s wait_for. On a busy event loop the executor could fail to run before the timeout fired, so the pool was cancelled without ever being shut down — an intermittent 'shutdown called 0 times' test failure and a real pool-leak risk. shutdown(wait=False) is non-blocking, so call it directly, matching the other cleanup paths in this module. Signed-off-by: chris hay --- CHANGELOG.md | 7 ++++ .../strategies/subprocess_strategy.py | 33 +++++++------------ 2 files changed, 19 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ee6187..b7e35fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,13 @@ All notable changes to this project are documented here. This project follows `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 diff --git a/src/chuk_tool_processor/execution/strategies/subprocess_strategy.py b/src/chuk_tool_processor/execution/strategies/subprocess_strategy.py index 26f7d7d..bc91394 100644 --- a/src/chuk_tool_processor/execution/strategies/subprocess_strategy.py +++ b/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: