diff --git a/python/semantic_kernel/connectors/mcp.py b/python/semantic_kernel/connectors/mcp.py index 0dd0935188dc..789461b2c805 100644 --- a/python/semantic_kernel/connectors/mcp.py +++ b/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) diff --git a/python/tests/unit/connectors/mcp/test_mcp.py b/python/tests/unit/connectors/mcp/test_mcp.py index 2e2eee7d8403..b4623489de12 100644 --- a/python/tests/unit/connectors/mcp/test_mcp.py +++ b/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