Python: Skip MCP tools and prompts whose normalized names collide

python/semantic_kernel/connectors/mcp.py

@@ -278,6 +278,7 @@ def __init__(
         self.sampling_auto_approve = sampling_auto_approve
         self._sampling_auto_approved_warning_logged = False
         self._mcp_reserved_attribute_names: set[str] | None = None
+        self._mcp_registered_names: dict[str, tuple[str, str]] = {}
         self._current_task: asyncio.Task | None = None
         self._stop_event: asyncio.Event | None = None
 
@@ -538,6 +539,27 @@ def _has_mcp_function_name_conflict(self, item_type: str, remote_name: str, loca
         )
         return True
 
+    def _is_mcp_local_name_taken(self, item_type: str, remote_name: str, local_name: str) -> bool:
+        """Check whether a normalized name is already bound to a different MCP tool or prompt.
+
+        Normalization is not injective, so distinct remote names can collapse into the same local
+        name. Tools and prompts share one attribute namespace, so registering a collision would
+        silently rebind an already advertised name to a different remote item.
+        """
+        owner = self._mcp_registered_names.get(local_name)
+        if owner is None or owner == (item_type, remote_name):
+            return False
+        owner_type, owner_name = owner
+        logger.warning(
+            "Skipping MCP %s '%s' because normalized name '%s' is already registered by %s '%s'.",
+            item_type,
+            remote_name,
+            local_name,
+            owner_type,
+            owner_name,
+        )
+        return True
+
     async def load_prompts(self):
         """Load prompts from the MCP server."""
         try:
@@ -546,8 +568,11 @@ async def load_prompts(self):
             prompt_list = None
         for prompt in prompt_list.prompts if prompt_list else []:
             local_name = _normalize_mcp_name(prompt.name)
+            if self._is_mcp_local_name_taken("prompt", prompt.name, local_name):
+                continue
             if self._has_mcp_function_name_conflict("prompt", prompt.name, local_name):
                 continue
+            self._mcp_registered_names[local_name] = ("prompt", prompt.name)
             func = kernel_function(name=local_name, description=prompt.description)(
                 partial(self.get_prompt, prompt.name)
             )
@@ -563,8 +588,11 @@ async def load_tools(self):
         # Create methods with the kernel_function decorator for each tool
         for tool in tool_list.tools if tool_list else []:
             local_name = _normalize_mcp_name(tool.name)
+            if self._is_mcp_local_name_taken("tool", tool.name, local_name):
+                continue
             if self._has_mcp_function_name_conflict("tool", tool.name, local_name):
                 continue
+            self._mcp_registered_names[local_name] = ("tool", tool.name)
             func = kernel_function(name=local_name, description=tool.description)(partial(self.call_tool, tool.name))
             func.__kernel_function_parameters__ = _get_parameter_dicts_from_mcp_tool(tool)
             setattr(self, local_name, func)

python/tests/unit/connectors/mcp/test_mcp.py

@@ -429,6 +429,99 @@ async def test_mcp_normalization_function(mock_session, list_tool_calls_with_sla
     assert _normalize_mcp_name("Name-With.Dots_And-Hyphens") == "Name-With.Dots_And-Hyphens"
 
 
+async def test_mcp_tool_name_collision_detected(caplog):
+    """Test that tools with names that normalize to the same identifier are detected and skipped."""
+    plugin = MCPSsePlugin(name="TestMCPPlugin", url="http://localhost:8080/sse")
+    session = AsyncMock(spec=ClientSession)
+    session.list_tools.return_value = ListToolsResult(
+        tools=[
+            Tool(name="read-document", description="first tool", inputSchema={}),
+            Tool(name="read document", description="second tool", inputSchema={}),
+        ]
+    )
+    plugin.session = session
+
+    with caplog.at_level(logging.WARNING, logger="semantic_kernel.connectors.mcp"):
+        await plugin.load_tools()
+
+    # Only the first tool should be registered
+    assert hasattr(plugin, "read-document")
+    func = getattr(plugin, "read-document")
+    assert func.__kernel_function_description__ == "first tool"
+    # Warning should be emitted for the collision
+    assert "read document" in caplog.text
+    assert "already registered" in caplog.text
+
+
+async def test_mcp_prompt_name_collision_detected(caplog):
+    """Test that prompts with names that normalize to the same identifier are detected and skipped."""
+    plugin = MCPSsePlugin(name="TestMCPPlugin", url="http://localhost:8080/sse")
+    session = AsyncMock(spec=ClientSession)
+    session.list_tools.return_value = ListToolsResult(tools=[])
+    session.list_prompts.return_value = types.ListPromptsResult(
+        prompts=[
+            types.Prompt(name="get-summary", description="first prompt", arguments=[]),
+            types.Prompt(name="get summary", description="second prompt", arguments=[]),
+        ]
+    )
+    plugin.session = session
+
+    with caplog.at_level(logging.WARNING, logger="semantic_kernel.connectors.mcp"):
+        await plugin.load_prompts()
+
+    # Only the first prompt should be registered
+    assert hasattr(plugin, "get-summary")
+    func = getattr(plugin, "get-summary")
+    assert func.__kernel_function_description__ == "first prompt"
+    # Warning should be emitted for the collision
+    assert "get summary" in caplog.text
+    assert "already registered" in caplog.text
+
+
+async def test_mcp_tool_name_collision_detected_across_reload(caplog):
+    """Test that a later tool reload cannot overwrite a previously registered normalized name."""
+    plugin = MCPSsePlugin(name="TestMCPPlugin", url="http://localhost:8080/sse")
+    session = AsyncMock(spec=ClientSession)
+    session.list_tools.side_effect = [
+        ListToolsResult(tools=[Tool(name="read-document", description="first tool", inputSchema={})]),
+        ListToolsResult(tools=[Tool(name="read document", description="second tool", inputSchema={})]),
+    ]
+    plugin.session = session
+
+    await plugin.load_tools()
+
+    with caplog.at_level(logging.WARNING, logger="semantic_kernel.connectors.mcp"):
+        await plugin.load_tools()
+
+    func = getattr(plugin, "read-document")
+    assert func.__kernel_function_description__ == "first tool"
+    assert "read document" in caplog.text
+    assert "already registered" in caplog.text
+
+
+async def test_mcp_prompt_does_not_replace_registered_tool_name(caplog):
+    """Test that a prompt does not rebind a normalized name already registered by a tool."""
+    plugin = MCPSsePlugin(name="TestMCPPlugin", url="http://localhost:8080/sse")
+    session = AsyncMock(spec=ClientSession)
+    session.list_tools.return_value = ListToolsResult(
+        tools=[Tool(name="read-document", description="first tool", inputSchema={})]
+    )
+    session.list_prompts.return_value = types.ListPromptsResult(
+        prompts=[types.Prompt(name="read document", description="second item", arguments=[])]
+    )
+    plugin.session = session
+
+    await plugin.load_tools()
+
+    with caplog.at_level(logging.WARNING, logger="semantic_kernel.connectors.mcp"):
+        await plugin.load_prompts()
+
+    func = getattr(plugin, "read-document")
+    assert func.__kernel_function_description__ == "first tool"
+    assert "read document" in caplog.text
+    assert "already registered" in caplog.text
+
+
 async def test_excluded_function_cannot_be_called(kernel: "Kernel"):
     """Test that excluded functions are rejected at call time, not just hidden from listing."""
     from semantic_kernel.connectors.mcp import create_mcp_server_from_kernel