-
Notifications
You must be signed in to change notification settings - Fork 3.6k
Expand file tree
/
Copy pathtest_mcp_duplicate_tools.py
More file actions
154 lines (112 loc) · 5.49 KB
/
test_mcp_duplicate_tools.py
File metadata and controls
154 lines (112 loc) · 5.49 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
"""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_normalizes_server_name_in_renamed_tool():
"""Test renamed tool names use a function-calling-safe server prefix."""
server1 = FakeMCPServer(server_name="Primary Server")
server1.add_tool("search", {})
server2 = FakeMCPServer(server_name="Secondary-Server")
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)
tool_names = [tool.name for tool in tools]
assert "search" in tool_names
assert "secondary_server__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