fix: handle duplicate tool names across MCP servers by auto-renaming

When multiple MCP servers have tools with the same name, instead of raising
a UserError, the SDK now automatically renames duplicate tools with a server
prefix (e.g., 'server_name__tool_name'). This allows users to use multiple
MCP servers even when they have overlapping tool names.

Changes:
- Modified MCPUtil.get_all_function_tools() to detect duplicates and rename
  them instead of raising an error
- Added MCPUtil._rename_tool() helper to properly copy and rename FunctionTool
  instances while preserving all internal metadata (_tool_origin, _mcp_title, etc.)
- Added warning logs to inform users about the renaming
- Updated existing tests to reflect the new behavior
- Added new tests for duplicate name handling scenarios

Fixes #1167
Fixes #464

Author: houguokun

src/agents/mcp/util.py

@@ -211,8 +211,13 @@ async def get_all_function_tools(
         agent: AgentBase,
         failure_error_function: ToolErrorFunction | None = default_tool_error_function,
     ) -> list[Tool]:
-        """Get all function tools from a list of MCP servers."""
-        tools = []
+        """Get all function tools from a list of MCP servers.
+
+        When multiple MCP servers have tools with the same name, the duplicate tools
+        are automatically renamed with a server prefix to avoid collisions. A warning
+        is logged to inform the user about the renaming.
+        """
+        tools: list[Tool] = []
         tool_names: set[str] = set()
         for server in servers:
             server_tools = await cls.get_function_tools(
@@ -223,16 +228,53 @@ async def get_all_function_tools(
                 failure_error_function=failure_error_function,
             )
             server_tool_names = {tool.name for tool in server_tools}
-            if len(server_tool_names & tool_names) > 0:
-                raise UserError(
-                    f"Duplicate tool names found across MCP servers: "
-                    f"{server_tool_names & tool_names}"
+            duplicates = server_tool_names & tool_names
+            if duplicates:
+                logger.warning(
+                    f"Duplicate tool names found across MCP servers: {duplicates}. "
+                    f"Renaming tools from server '{server.name}' with prefix."
                 )
-            tool_names.update(server_tool_names)
-            tools.extend(server_tools)
+                renamed_tools: list[Tool] = []
+                for tool in server_tools:
+                    if tool.name in duplicates:
+                        original_name = tool.name
+                        new_name = f"{server.name}__{original_name}"
+                        # Ensure the new name doesn't also collide
+                        while new_name in tool_names:
+                            new_name = f"{new_name}_"
+                        logger.warning(
+                            f"Renamed MCP tool '{original_name}' from server "
+                            f"'{server.name}' to '{new_name}'"
+                        )
+                        # Create a renamed copy of the tool
+                        renamed_tool = cls._rename_tool(tool, new_name)
+                        renamed_tools.append(renamed_tool)
+                        tool_names.add(new_name)
+                    else:
+                        renamed_tools.append(tool)
+                        tool_names.add(tool.name)
+                tools.extend(renamed_tools)
+            else:
+                tool_names.update(server_tool_names)
+                tools.extend(server_tools)
 
         return tools
 
+    @staticmethod
+    def _rename_tool(tool: Tool, new_name: str) -> Tool:
+        """Create a copy of a tool with a new name.
+
+        For FunctionTool instances, uses the built-in __copy__ method to ensure all
+        internal fields (including _mcp_title, _tool_origin, etc.) are properly copied.
+        """
+        if isinstance(tool, FunctionTool):
+            renamed = tool.__copy__()
+            renamed.name = new_name
+            return renamed
+        # For other tool types, try to set name directly
+        tool.name = new_name
+        return tool
+
     @classmethod
     async def get_function_tools(
         cls,

tests/mcp/test_mcp_duplicate_tools.py

@@ -0,0 +1,134 @@
+"""Tests for MCP duplicate tool name handling."""
+
+import pytest
+
+from agents import Agent, FunctionTool, RunContextWrapper
+from agents.mcp import MCPServer, MCPUtil
+
+from .helpers import FakeMCPServer
+
+
+@pytest.mark.asyncio
+async def test_get_all_function_tools_with_duplicate_names():
+    """Test that duplicate tool names across MCP servers are automatically renamed."""
+    server1 = FakeMCPServer(server_name="server1")
+    server1.add_tool("search", {})
+    server1.add_tool("fetch", {})
+
+    server2 = FakeMCPServer(server_name="server2")
+    server2.add_tool("search", {})  # duplicate name
+    server2.add_tool("update", {})
+
+    servers: list[MCPServer] = [server1, server2]
+    run_context = RunContextWrapper(context=None)
+    agent = Agent(name="test_agent", instructions="Test agent")
+
+    tools = await MCPUtil.get_all_function_tools(servers, False, run_context, agent)
+
+    # Should have 4 tools total
+    assert len(tools) == 4
+
+    tool_names = [tool.name for tool in tools]
+    # Original names from first server should be preserved
+    assert "search" in tool_names
+    assert "fetch" in tool_names
+    # Duplicate from second server should be renamed
+    assert "server2__search" in tool_names
+    assert "update" in tool_names
+
+
+@pytest.mark.asyncio
+async def test_get_all_function_tools_with_duplicate_names_three_servers():
+    """Test duplicate tool name handling with three servers having the same tool name."""
+    server1 = FakeMCPServer(server_name="server1")
+    server1.add_tool("search", {})
+
+    server2 = FakeMCPServer(server_name="server2")
+    server2.add_tool("search", {})  # duplicate
+
+    server3 = FakeMCPServer(server_name="server3")
+    server3.add_tool("search", {})  # another duplicate
+
+    servers: list[MCPServer] = [server1, server2, server3]
+    run_context = RunContextWrapper(context=None)
+    agent = Agent(name="test_agent", instructions="Test agent")
+
+    tools = await MCPUtil.get_all_function_tools(servers, False, run_context, agent)
+
+    assert len(tools) == 3
+    tool_names = [tool.name for tool in tools]
+    assert "search" in tool_names
+    assert "server2__search" in tool_names
+    assert "server3__search" in tool_names
+
+
+@pytest.mark.asyncio
+async def test_get_all_function_tools_no_duplicates():
+    """Test that non-duplicate tool names are not affected."""
+    server1 = FakeMCPServer(server_name="server1")
+    server1.add_tool("search", {})
+
+    server2 = FakeMCPServer(server_name="server2")
+    server2.add_tool("fetch", {})  # no duplicate
+
+    servers: list[MCPServer] = [server1, server2]
+    run_context = RunContextWrapper(context=None)
+    agent = Agent(name="test_agent", instructions="Test agent")
+
+    tools = await MCPUtil.get_all_function_tools(servers, False, run_context, agent)
+
+    assert len(tools) == 2
+    tool_names = [tool.name for tool in tools]
+    assert "search" in tool_names
+    assert "fetch" in tool_names
+    # Should not have any prefixed names
+    assert "server1__search" not in tool_names
+    assert "server2__fetch" not in tool_names
+
+
+@pytest.mark.asyncio
+async def test_get_all_function_tools_preserves_mcp_origin():
+    """Test that renamed tools preserve their MCP origin metadata."""
+    server1 = FakeMCPServer(server_name="server1")
+    server1.add_tool("search", {})
+
+    server2 = FakeMCPServer(server_name="server2")
+    server2.add_tool("search", {})  # duplicate
+
+    servers: list[MCPServer] = [server1, server2]
+    run_context = RunContextWrapper(context=None)
+    agent = Agent(name="test_agent", instructions="Test agent")
+
+    tools = await MCPUtil.get_all_function_tools(servers, False, run_context, agent)
+
+    # Find the renamed tool
+    renamed_tool = next((t for t in tools if t.name == "server2__search"), None)
+    assert renamed_tool is not None
+    assert isinstance(renamed_tool, FunctionTool)
+    # Check that MCP origin is preserved
+    assert renamed_tool._tool_origin is not None
+    assert renamed_tool._tool_origin.mcp_server_name == "server2"
+
+
+@pytest.mark.asyncio
+async def test_renamed_tool_can_be_invoked():
+    """Test that renamed tools can still be invoked successfully."""
+    server1 = FakeMCPServer(server_name="server1")
+    server1.add_tool("search", {})
+
+    server2 = FakeMCPServer(server_name="server2")
+    server2.add_tool("search", {})  # duplicate
+
+    servers: list[MCPServer] = [server1, server2]
+    run_context = RunContextWrapper(context=None)
+    agent = Agent(name="test_agent", instructions="Test agent")
+
+    tools = await MCPUtil.get_all_function_tools(servers, False, run_context, agent)
+
+    # Find the renamed tool and invoke it
+    renamed_tool = next((t for t in tools if t.name == "server2__search"), None)
+    assert renamed_tool is not None
+    assert isinstance(renamed_tool, FunctionTool)
+
+    # The tool should be invocable
+    assert renamed_tool.on_invoke_tool is not None

tests/mcp/test_runner_calls_mcp.py

@@ -8,7 +8,6 @@
     ModelBehaviorError,
     RunContextWrapper,
     Runner,
-    UserError,
     default_tool_error_function,
 )
 from agents.exceptions import AgentsException
@@ -125,14 +124,14 @@ async def test_runner_works_with_multiple_mcp_servers(streaming: bool):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("streaming", [False, True])
-async def test_runner_errors_when_mcp_tools_clash(streaming: bool):
-    """Test that the runner errors when multiple servers have the same tool name."""
+async def test_runner_renames_mcp_tools_when_names_clash(streaming: bool):
+    """Test that the runner auto-renames tools when multiple servers have same name."""
     server1 = FakeMCPServer()
     server1.add_tool("test_tool_1", {})
     server1.add_tool("test_tool_2", {})
 
     server2 = FakeMCPServer()
-    server2.add_tool("test_tool_2", {})
+    server2.add_tool("test_tool_2", {})  # duplicate name
     server2.add_tool("test_tool_3", {})
 
     model = FakeModel()
@@ -145,19 +144,58 @@ async def test_runner_errors_when_mcp_tools_clash(streaming: bool):
     model.add_multiple_turn_outputs(
         [
             # First turn: a message and tool call
+            # test_tool_3 is unique to server2, so it should work without renaming
             [get_text_message("a_message"), get_function_tool_call("test_tool_3", "")],
             # Second turn: text message
             [get_text_message("done")],
         ]
     )
 
-    with pytest.raises(UserError):
-        if streaming:
-            result = Runner.run_streamed(agent, input="user_message")
-            async for _ in result.stream_events():
-                pass
-        else:
-            await Runner.run(agent, input="user_message")
+    if streaming:
+        result = Runner.run_streamed(agent, input="user_message")
+        async for _ in result.stream_events():
+            pass
+    else:
+        await Runner.run(agent, input="user_message")
+
+    # server2's test_tool_3 should be called successfully (no rename needed)
+    assert server2.tool_calls == ["test_tool_3"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_runner_renamed_mcp_tool_can_be_called(streaming: bool):
+    """Test that renamed MCP tools can still be invoked by the model."""
+    server1 = FakeMCPServer(server_name="server1")
+    server1.add_tool("search", {})
+
+    server2 = FakeMCPServer(server_name="server2")
+    server2.add_tool("search", {})  # duplicate name
+
+    model = FakeModel()
+    agent = Agent(
+        name="test",
+        model=model,
+        mcp_servers=[server1, server2],
+    )
+
+    model.add_multiple_turn_outputs(
+        [
+            # The model should use the renamed tool name
+            [get_text_message("a_message"), get_function_tool_call("server2__search", "")],
+            [get_text_message("done")],
+        ]
+    )
+
+    if streaming:
+        result = Runner.run_streamed(agent, input="user_message")
+        async for _ in result.stream_events():
+            pass
+    else:
+        await Runner.run(agent, input="user_message")
+
+    # The renamed tool from server2 should be called
+    assert server2.tool_calls == ["search"]
 
 
 class Foo(BaseModel):