Jeremiah Lowin commited on
Commit
2c1fc32
·
1 Parent(s): 736b52b

Fix bug with duplicate behavior == ignore

Browse files
docs/servers/fastmcp.mdx CHANGED
@@ -314,12 +314,12 @@ from fastmcp.settings import DuplicateBehavior
314
  mcp = FastMCP(
315
  name="ConfiguredServer",
316
  port=8080, # Directly maps to ServerSettings
317
- on_duplicate_tools=DuplicateBehavior.ERROR # Set duplicate handling
318
  )
319
 
320
  # Settings are accessible via mcp.settings
321
  print(mcp.settings.port) # Output: 8080
322
- print(mcp.settings.on_duplicate_tools) # Output: DuplicateBehavior.ERROR
323
  ```
324
 
325
  ### Key Configuration Options
 
314
  mcp = FastMCP(
315
  name="ConfiguredServer",
316
  port=8080, # Directly maps to ServerSettings
317
+ on_duplicate_tools="error" # Set duplicate handling
318
  )
319
 
320
  # Settings are accessible via mcp.settings
321
  print(mcp.settings.port) # Output: 8080
322
+ print(mcp.settings.on_duplicate_tools) # Output: "error"
323
  ```
324
 
325
  ### Key Configuration Options
docs/servers/prompts.mdx CHANGED
@@ -209,7 +209,7 @@ from fastmcp.settings import DuplicateBehavior
209
 
210
  mcp = FastMCP(
211
  name="PromptServer",
212
- on_duplicate_prompts=DuplicateBehavior.ERROR # Raise an error if a prompt name is duplicated
213
  )
214
 
215
  @mcp.prompt()
 
209
 
210
  mcp = FastMCP(
211
  name="PromptServer",
212
+ on_duplicate_prompts="error" # Raise an error if a prompt name is duplicated
213
  )
214
 
215
  @mcp.prompt()
docs/servers/resources.mdx CHANGED
@@ -301,7 +301,7 @@ from fastmcp.settings import DuplicateBehavior
301
 
302
  mcp = FastMCP(
303
  name="ResourceServer",
304
- on_duplicate_resources=DuplicateBehavior.ERROR # Raise error on duplicates
305
  )
306
 
307
  @mcp.resource("data://config")
 
301
 
302
  mcp = FastMCP(
303
  name="ResourceServer",
304
+ on_duplicate_resources="error" # Raise error on duplicates
305
  )
306
 
307
  @mcp.resource("data://config")
docs/servers/resources_backup.mdx CHANGED
@@ -250,7 +250,7 @@ from fastmcp.settings import DuplicateBehavior
250
 
251
  mcp = FastMCP(
252
  name="ResourceServer",
253
- on_duplicate_resources=DuplicateBehavior.ERROR # Raise error on duplicates
254
  )
255
 
256
  @mcp.resource("data://config")
 
250
 
251
  mcp = FastMCP(
252
  name="ResourceServer",
253
+ on_duplicate_resources="error" # Raise error on duplicates
254
  )
255
 
256
  @mcp.resource("data://config")
docs/servers/tools.mdx CHANGED
@@ -315,7 +315,7 @@ from fastmcp.settings import DuplicateBehavior
315
  mcp = FastMCP(
316
  name="StrictServer",
317
  # Configure behavior for duplicate tool names
318
- on_duplicate_tools=DuplicateBehavior.ERROR
319
  )
320
 
321
  @mcp.tool()
 
315
  mcp = FastMCP(
316
  name="StrictServer",
317
  # Configure behavior for duplicate tool names
318
+ on_duplicate_tools="error"
319
  )
320
 
321
  @mcp.tool()
src/fastmcp/prompts/prompt_manager.py CHANGED
@@ -15,8 +15,19 @@ logger = get_logger(__name__)
15
  class PromptManager:
16
  """Manages FastMCP prompts."""
17
 
18
- def __init__(self, duplicate_behavior: DuplicateBehavior = DuplicateBehavior.WARN):
19
  self._prompts: dict[str, Prompt] = {}
 
 
 
 
 
 
 
 
 
 
 
20
  self.duplicate_behavior = duplicate_behavior
21
 
22
  def get_prompt(self, name: str) -> Prompt | None:
@@ -44,17 +55,17 @@ class PromptManager:
44
  # Check for duplicates
45
  existing = self._prompts.get(prompt.name)
46
  if existing:
47
- if self.duplicate_behavior == DuplicateBehavior.WARN:
48
  logger.warning(f"Prompt already exists: {prompt.name}")
49
  self._prompts[prompt.name] = prompt
50
- elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
51
  self._prompts[prompt.name] = prompt
52
- elif self.duplicate_behavior == DuplicateBehavior.ERROR:
53
  raise ValueError(f"Prompt already exists: {prompt.name}")
54
- elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
55
- pass
56
-
57
- self._prompts[prompt.name] = prompt
58
  return prompt
59
 
60
  async def render_prompt(
 
15
  class PromptManager:
16
  """Manages FastMCP prompts."""
17
 
18
+ def __init__(self, duplicate_behavior: DuplicateBehavior | None = None):
19
  self._prompts: dict[str, Prompt] = {}
20
+
21
+ # Default to "warn" if None is provided
22
+ if duplicate_behavior is None:
23
+ duplicate_behavior = "warn"
24
+
25
+ if duplicate_behavior not in DuplicateBehavior.__args__:
26
+ raise ValueError(
27
+ f"Invalid duplicate_behavior: {duplicate_behavior}. "
28
+ f"Must be one of: {', '.join(DuplicateBehavior.__args__)}"
29
+ )
30
+
31
  self.duplicate_behavior = duplicate_behavior
32
 
33
  def get_prompt(self, name: str) -> Prompt | None:
 
55
  # Check for duplicates
56
  existing = self._prompts.get(prompt.name)
57
  if existing:
58
+ if self.duplicate_behavior == "warn":
59
  logger.warning(f"Prompt already exists: {prompt.name}")
60
  self._prompts[prompt.name] = prompt
61
+ elif self.duplicate_behavior == "replace":
62
  self._prompts[prompt.name] = prompt
63
+ elif self.duplicate_behavior == "error":
64
  raise ValueError(f"Prompt already exists: {prompt.name}")
65
+ elif self.duplicate_behavior == "ignore":
66
+ return existing
67
+ else:
68
+ self._prompts[prompt.name] = prompt
69
  return prompt
70
 
71
  async def render_prompt(
src/fastmcp/resources/resource_manager.py CHANGED
@@ -19,9 +19,20 @@ logger = get_logger(__name__)
19
  class ResourceManager:
20
  """Manages FastMCP resources."""
21
 
22
- def __init__(self, duplicate_behavior: DuplicateBehavior = DuplicateBehavior.WARN):
23
  self._resources: dict[str, Resource] = {}
24
  self._templates: dict[str, ResourceTemplate] = {}
 
 
 
 
 
 
 
 
 
 
 
25
  self.duplicate_behavior = duplicate_behavior
26
 
27
  def add_resource_or_template_from_fn(
@@ -114,16 +125,17 @@ class ResourceManager:
114
  )
115
  existing = self._resources.get(str(resource.uri))
116
  if existing:
117
- if self.duplicate_behavior == DuplicateBehavior.WARN:
118
  logger.warning(f"Resource already exists: {resource.uri}")
119
  self._resources[str(resource.uri)] = resource
120
- elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
121
  self._resources[str(resource.uri)] = resource
122
- elif self.duplicate_behavior == DuplicateBehavior.ERROR:
123
  raise ValueError(f"Resource already exists: {resource.uri}")
124
- elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
125
- pass
126
- self._resources[str(resource.uri)] = resource
 
127
  return resource
128
 
129
  def add_template_from_fn(
@@ -167,16 +179,17 @@ class ResourceManager:
167
  )
168
  existing = self._templates.get(str(template.uri_template))
169
  if existing:
170
- if self.duplicate_behavior == DuplicateBehavior.WARN:
171
  logger.warning(f"Resource already exists: {template.uri_template}")
172
  self._templates[str(template.uri_template)] = template
173
- elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
174
  self._templates[str(template.uri_template)] = template
175
- elif self.duplicate_behavior == DuplicateBehavior.ERROR:
176
  raise ValueError(f"Resource already exists: {template.uri_template}")
177
- elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
178
- pass
179
- self._templates[template.uri_template] = template
 
180
  return template
181
 
182
  async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
 
19
  class ResourceManager:
20
  """Manages FastMCP resources."""
21
 
22
+ def __init__(self, duplicate_behavior: DuplicateBehavior | None = None):
23
  self._resources: dict[str, Resource] = {}
24
  self._templates: dict[str, ResourceTemplate] = {}
25
+
26
+ # Default to "warn" if None is provided
27
+ if duplicate_behavior is None:
28
+ duplicate_behavior = "warn"
29
+
30
+ if duplicate_behavior not in DuplicateBehavior.__args__:
31
+ raise ValueError(
32
+ f"Invalid duplicate_behavior: {duplicate_behavior}. "
33
+ f"Must be one of: {', '.join(DuplicateBehavior.__args__)}"
34
+ )
35
+
36
  self.duplicate_behavior = duplicate_behavior
37
 
38
  def add_resource_or_template_from_fn(
 
125
  )
126
  existing = self._resources.get(str(resource.uri))
127
  if existing:
128
+ if self.duplicate_behavior == "warn":
129
  logger.warning(f"Resource already exists: {resource.uri}")
130
  self._resources[str(resource.uri)] = resource
131
+ elif self.duplicate_behavior == "replace":
132
  self._resources[str(resource.uri)] = resource
133
+ elif self.duplicate_behavior == "error":
134
  raise ValueError(f"Resource already exists: {resource.uri}")
135
+ elif self.duplicate_behavior == "ignore":
136
+ return existing
137
+ else:
138
+ self._resources[str(resource.uri)] = resource
139
  return resource
140
 
141
  def add_template_from_fn(
 
179
  )
180
  existing = self._templates.get(str(template.uri_template))
181
  if existing:
182
+ if self.duplicate_behavior == "warn":
183
  logger.warning(f"Resource already exists: {template.uri_template}")
184
  self._templates[str(template.uri_template)] = template
185
+ elif self.duplicate_behavior == "replace":
186
  self._templates[str(template.uri_template)] = template
187
+ elif self.duplicate_behavior == "error":
188
  raise ValueError(f"Resource already exists: {template.uri_template}")
189
+ elif self.duplicate_behavior == "ignore":
190
+ return existing
191
+ else:
192
+ self._templates[template.uri_template] = template
193
  return template
194
 
195
  async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
src/fastmcp/settings.py CHANGED
@@ -1,6 +1,5 @@
1
  from __future__ import annotations as _annotations
2
 
3
- from enum import Enum
4
  from typing import TYPE_CHECKING, Literal
5
 
6
  from pydantic import Field
@@ -11,12 +10,7 @@ if TYPE_CHECKING:
11
 
12
  LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
13
 
14
-
15
- class DuplicateBehavior(Enum):
16
- WARN = "warn"
17
- ERROR = "error"
18
- REPLACE = "replace"
19
- IGNORE = "ignore"
20
 
21
 
22
  class Settings(BaseSettings):
@@ -55,13 +49,13 @@ class ServerSettings(BaseSettings):
55
  debug: bool = False
56
 
57
  # resource settings
58
- on_duplicate_resources: DuplicateBehavior = DuplicateBehavior.WARN
59
 
60
  # tool settings
61
- on_duplicate_tools: DuplicateBehavior = DuplicateBehavior.WARN
62
 
63
  # prompt settings
64
- on_duplicate_prompts: DuplicateBehavior = DuplicateBehavior.WARN
65
 
66
  dependencies: list[str] = Field(
67
  default_factory=list,
 
1
  from __future__ import annotations as _annotations
2
 
 
3
  from typing import TYPE_CHECKING, Literal
4
 
5
  from pydantic import Field
 
10
 
11
  LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
12
 
13
+ DuplicateBehavior = Literal["warn", "error", "replace", "ignore"]
 
 
 
 
 
14
 
15
 
16
  class Settings(BaseSettings):
 
49
  debug: bool = False
50
 
51
  # resource settings
52
+ on_duplicate_resources: DuplicateBehavior = "warn"
53
 
54
  # tool settings
55
+ on_duplicate_tools: DuplicateBehavior = "warn"
56
 
57
  # prompt settings
58
+ on_duplicate_prompts: DuplicateBehavior = "warn"
59
 
60
  dependencies: list[str] = Field(
61
  default_factory=list,
src/fastmcp/tools/tool_manager.py CHANGED
@@ -21,8 +21,19 @@ logger = get_logger(__name__)
21
  class ToolManager:
22
  """Manages FastMCP tools."""
23
 
24
- def __init__(self, duplicate_behavior: DuplicateBehavior = DuplicateBehavior.WARN):
25
  self._tools: dict[str, Tool] = {}
 
 
 
 
 
 
 
 
 
 
 
26
  self.duplicate_behavior = duplicate_behavior
27
 
28
  def get_tool(self, name: str) -> Tool | None:
@@ -57,16 +68,17 @@ class ToolManager:
57
  name = name or tool.name
58
  existing = self._tools.get(name)
59
  if existing:
60
- if self.duplicate_behavior == DuplicateBehavior.WARN:
61
  logger.warning(f"Tool already exists: {name}")
62
  self._tools[name] = tool
63
- elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
64
  self._tools[name] = tool
65
- elif self.duplicate_behavior == DuplicateBehavior.ERROR:
66
  raise ValueError(f"Tool already exists: {name}")
67
- elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
68
- pass
69
- self._tools[name] = tool
 
70
  return tool
71
 
72
  async def call_tool(
 
21
  class ToolManager:
22
  """Manages FastMCP tools."""
23
 
24
+ def __init__(self, duplicate_behavior: DuplicateBehavior | None = None):
25
  self._tools: dict[str, Tool] = {}
26
+
27
+ # Default to "warn" if None is provided
28
+ if duplicate_behavior is None:
29
+ duplicate_behavior = "warn"
30
+
31
+ if duplicate_behavior not in DuplicateBehavior.__args__:
32
+ raise ValueError(
33
+ f"Invalid duplicate_behavior: {duplicate_behavior}. "
34
+ f"Must be one of: {', '.join(DuplicateBehavior.__args__)}"
35
+ )
36
+
37
  self.duplicate_behavior = duplicate_behavior
38
 
39
  def get_tool(self, name: str) -> Tool | None:
 
68
  name = name or tool.name
69
  existing = self._tools.get(name)
70
  if existing:
71
+ if self.duplicate_behavior == "warn":
72
  logger.warning(f"Tool already exists: {name}")
73
  self._tools[name] = tool
74
+ elif self.duplicate_behavior == "replace":
75
  self._tools[name] = tool
76
+ elif self.duplicate_behavior == "error":
77
  raise ValueError(f"Tool already exists: {name}")
78
+ elif self.duplicate_behavior == "ignore":
79
+ return existing
80
+ else:
81
+ self._tools[name] = tool
82
  return tool
83
 
84
  async def call_tool(
tests/prompts/test_prompt_manager.py CHANGED
@@ -4,7 +4,6 @@ from fastmcp.exceptions import PromptError
4
  from fastmcp.prompts import Prompt
5
  from fastmcp.prompts.prompt import PromptArgument, TextContent, UserMessage
6
  from fastmcp.prompts.prompt_manager import PromptManager
7
- from fastmcp.settings import DuplicateBehavior
8
 
9
 
10
  class TestPromptManager:
@@ -26,7 +25,7 @@ class TestPromptManager:
26
  def fn() -> str:
27
  return "Hello, world!"
28
 
29
- manager = PromptManager(duplicate_behavior=DuplicateBehavior.WARN)
30
  prompt = Prompt.from_function(fn)
31
  first = manager.add_prompt(prompt)
32
  second = manager.add_prompt(prompt)
@@ -39,13 +38,87 @@ class TestPromptManager:
39
  def fn() -> str:
40
  return "Hello, world!"
41
 
42
- manager = PromptManager(duplicate_behavior=DuplicateBehavior.IGNORE)
43
  prompt = Prompt.from_function(fn)
44
  first = manager.add_prompt(prompt)
45
  second = manager.add_prompt(prompt)
46
  assert first == second
47
  assert "Prompt already exists" not in caplog.text
48
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
49
  def test_list_prompts(self):
50
  """Test listing all prompts."""
51
 
@@ -114,39 +187,6 @@ class TestPromptManager:
114
  with pytest.raises(ValueError, match="Missing required arguments"):
115
  await manager.render_prompt("fn")
116
 
117
- def test_error_on_duplicate_prompts(self):
118
- """Test error on duplicate prompts."""
119
-
120
- def fn() -> str:
121
- return "Hello, world!"
122
-
123
- manager = PromptManager(duplicate_behavior=DuplicateBehavior.ERROR)
124
- prompt = Prompt.from_function(fn)
125
- manager.add_prompt(prompt)
126
-
127
- with pytest.raises(ValueError, match="Prompt already exists"):
128
- manager.add_prompt(prompt)
129
-
130
- def test_replace_duplicate_prompts(self):
131
- """Test replacing duplicate prompts."""
132
-
133
- def fn1() -> str:
134
- return "Original"
135
-
136
- def fn2() -> str:
137
- return "Replacement"
138
-
139
- manager = PromptManager(duplicate_behavior=DuplicateBehavior.REPLACE)
140
- prompt1 = Prompt.from_function(fn1, name="test_prompt")
141
- prompt2 = Prompt.from_function(fn2, name="test_prompt")
142
-
143
- manager.add_prompt(prompt1)
144
- manager.add_prompt(prompt2)
145
-
146
- # Should have replaced the first prompt with the second
147
- stored_prompt = manager.get_prompt("test_prompt")
148
- assert stored_prompt == prompt2
149
-
150
 
151
  class TestPromptTags:
152
  """Test functionality related to prompt tags."""
 
4
  from fastmcp.prompts import Prompt
5
  from fastmcp.prompts.prompt import PromptArgument, TextContent, UserMessage
6
  from fastmcp.prompts.prompt_manager import PromptManager
 
7
 
8
 
9
  class TestPromptManager:
 
25
  def fn() -> str:
26
  return "Hello, world!"
27
 
28
+ manager = PromptManager(duplicate_behavior="warn")
29
  prompt = Prompt.from_function(fn)
30
  first = manager.add_prompt(prompt)
31
  second = manager.add_prompt(prompt)
 
38
  def fn() -> str:
39
  return "Hello, world!"
40
 
41
+ manager = PromptManager(duplicate_behavior="ignore")
42
  prompt = Prompt.from_function(fn)
43
  first = manager.add_prompt(prompt)
44
  second = manager.add_prompt(prompt)
45
  assert first == second
46
  assert "Prompt already exists" not in caplog.text
47
 
48
+ def test_warn_on_duplicate_prompts(self, caplog):
49
+ """Test warning on duplicate prompts."""
50
+ manager = PromptManager(duplicate_behavior="warn")
51
+
52
+ def test_fn() -> str:
53
+ return "Test prompt"
54
+
55
+ prompt = Prompt.from_function(test_fn, name="test_prompt")
56
+
57
+ manager.add_prompt(prompt)
58
+ manager.add_prompt(prompt)
59
+
60
+ assert "Prompt already exists: test_prompt" in caplog.text
61
+ # Should have the prompt
62
+ assert manager.get_prompt("test_prompt") is not None
63
+
64
+ def test_error_on_duplicate_prompts(self):
65
+ """Test error on duplicate prompts."""
66
+ manager = PromptManager(duplicate_behavior="error")
67
+
68
+ def test_fn() -> str:
69
+ return "Test prompt"
70
+
71
+ prompt = Prompt.from_function(test_fn, name="test_prompt")
72
+
73
+ manager.add_prompt(prompt)
74
+
75
+ with pytest.raises(ValueError, match="Prompt already exists: test_prompt"):
76
+ manager.add_prompt(prompt)
77
+
78
+ def test_replace_duplicate_prompts(self):
79
+ """Test replacing duplicate prompts."""
80
+ manager = PromptManager(duplicate_behavior="replace")
81
+
82
+ def original_fn() -> str:
83
+ return "Original prompt"
84
+
85
+ def replacement_fn() -> str:
86
+ return "Replacement prompt"
87
+
88
+ prompt1 = Prompt.from_function(original_fn, name="test_prompt")
89
+ prompt2 = Prompt.from_function(replacement_fn, name="test_prompt")
90
+
91
+ manager.add_prompt(prompt1)
92
+ manager.add_prompt(prompt2)
93
+
94
+ # Should have replaced with the new prompt
95
+ prompt = manager.get_prompt("test_prompt")
96
+ assert prompt is not None
97
+ assert prompt.fn.__name__ == "replacement_fn"
98
+
99
+ def test_ignore_duplicate_prompts(self):
100
+ """Test ignoring duplicate prompts."""
101
+ manager = PromptManager(duplicate_behavior="ignore")
102
+
103
+ def original_fn() -> str:
104
+ return "Original prompt"
105
+
106
+ def replacement_fn() -> str:
107
+ return "Replacement prompt"
108
+
109
+ prompt1 = Prompt.from_function(original_fn, name="test_prompt")
110
+ prompt2 = Prompt.from_function(replacement_fn, name="test_prompt")
111
+
112
+ manager.add_prompt(prompt1)
113
+ result = manager.add_prompt(prompt2)
114
+
115
+ # Should keep the original
116
+ prompt = manager.get_prompt("test_prompt")
117
+ assert prompt is not None
118
+ assert prompt.fn.__name__ == "original_fn"
119
+ # Result should be the original prompt
120
+ assert result.fn.__name__ == "original_fn"
121
+
122
  def test_list_prompts(self):
123
  """Test listing all prompts."""
124
 
 
187
  with pytest.raises(ValueError, match="Missing required arguments"):
188
  await manager.render_prompt("fn")
189
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
190
 
191
  class TestPromptTags:
192
  """Test functionality related to prompt tags."""
tests/resources/test_resource_manager.py CHANGED
@@ -11,7 +11,6 @@ from fastmcp.resources import (
11
  ResourceManager,
12
  ResourceTemplate,
13
  )
14
- from fastmcp.settings import DuplicateBehavior
15
 
16
 
17
  @pytest.fixture
@@ -61,19 +60,24 @@ class TestResourceManager:
61
 
62
  def test_warn_on_duplicate_resources(self, temp_file: Path, caplog):
63
  """Test warning on duplicate resources."""
64
- manager = ResourceManager(duplicate_behavior=DuplicateBehavior.WARN)
 
65
  resource = FileResource(
66
  uri=FileUrl(f"file://{temp_file}"),
67
- name="test",
68
  path=temp_file,
69
  )
 
70
  manager.add_resource(resource)
71
  manager.add_resource(resource)
 
72
  assert "Resource already exists" in caplog.text
 
 
73
 
74
  def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog):
75
  """Test disabling warning on duplicate resources."""
76
- manager = ResourceManager(duplicate_behavior=DuplicateBehavior.IGNORE)
77
  resource = FileResource(
78
  uri=FileUrl(f"file://{temp_file}"),
79
  name="test",
@@ -85,12 +89,14 @@ class TestResourceManager:
85
 
86
  def test_error_on_duplicate_resources(self, temp_file: Path):
87
  """Test error on duplicate resources."""
88
- manager = ResourceManager(duplicate_behavior=DuplicateBehavior.ERROR)
 
89
  resource = FileResource(
90
  uri=FileUrl(f"file://{temp_file}"),
91
- name="test",
92
  path=temp_file,
93
  )
 
94
  manager.add_resource(resource)
95
 
96
  with pytest.raises(ValueError, match="Resource already exists"):
@@ -98,27 +104,153 @@ class TestResourceManager:
98
 
99
  def test_replace_duplicate_resources(self, temp_file: Path):
100
  """Test replacing duplicate resources."""
101
- manager = ResourceManager(duplicate_behavior=DuplicateBehavior.REPLACE)
102
 
103
  resource1 = FileResource(
104
  uri=FileUrl(f"file://{temp_file}"),
105
- name="test1",
106
  path=temp_file,
107
  )
108
 
109
  resource2 = FileResource(
110
  uri=FileUrl(f"file://{temp_file}"),
111
- name="test2", # Different name
112
  path=temp_file,
113
  )
114
 
115
  manager.add_resource(resource1)
116
  manager.add_resource(resource2)
117
 
118
- # Should have replaced the first resource with the second
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
119
  resources = manager.list_resources()
120
  assert len(resources) == 1
121
- assert resources[0].name == "test2"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
122
 
123
  @pytest.mark.anyio
124
  async def test_get_resource(self, temp_file: Path):
 
11
  ResourceManager,
12
  ResourceTemplate,
13
  )
 
14
 
15
 
16
  @pytest.fixture
 
60
 
61
  def test_warn_on_duplicate_resources(self, temp_file: Path, caplog):
62
  """Test warning on duplicate resources."""
63
+ manager = ResourceManager(duplicate_behavior="warn")
64
+
65
  resource = FileResource(
66
  uri=FileUrl(f"file://{temp_file}"),
67
+ name="test_resource",
68
  path=temp_file,
69
  )
70
+
71
  manager.add_resource(resource)
72
  manager.add_resource(resource)
73
+
74
  assert "Resource already exists" in caplog.text
75
+ # Should have the resource
76
+ assert len(manager.list_resources()) == 1
77
 
78
  def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog):
79
  """Test disabling warning on duplicate resources."""
80
+ manager = ResourceManager(duplicate_behavior="ignore")
81
  resource = FileResource(
82
  uri=FileUrl(f"file://{temp_file}"),
83
  name="test",
 
89
 
90
  def test_error_on_duplicate_resources(self, temp_file: Path):
91
  """Test error on duplicate resources."""
92
+ manager = ResourceManager(duplicate_behavior="error")
93
+
94
  resource = FileResource(
95
  uri=FileUrl(f"file://{temp_file}"),
96
+ name="test_resource",
97
  path=temp_file,
98
  )
99
+
100
  manager.add_resource(resource)
101
 
102
  with pytest.raises(ValueError, match="Resource already exists"):
 
104
 
105
  def test_replace_duplicate_resources(self, temp_file: Path):
106
  """Test replacing duplicate resources."""
107
+ manager = ResourceManager(duplicate_behavior="replace")
108
 
109
  resource1 = FileResource(
110
  uri=FileUrl(f"file://{temp_file}"),
111
+ name="original",
112
  path=temp_file,
113
  )
114
 
115
  resource2 = FileResource(
116
  uri=FileUrl(f"file://{temp_file}"),
117
+ name="replacement",
118
  path=temp_file,
119
  )
120
 
121
  manager.add_resource(resource1)
122
  manager.add_resource(resource2)
123
 
124
+ # Should have replaced with the new resource
125
+ resources = manager.list_resources()
126
+ assert len(resources) == 1
127
+ assert resources[0].name == "replacement"
128
+
129
+ def test_ignore_duplicate_resources(self, temp_file: Path):
130
+ """Test ignoring duplicate resources."""
131
+ manager = ResourceManager(duplicate_behavior="ignore")
132
+
133
+ resource1 = FileResource(
134
+ uri=FileUrl(f"file://{temp_file}"),
135
+ name="original",
136
+ path=temp_file,
137
+ )
138
+
139
+ resource2 = FileResource(
140
+ uri=FileUrl(f"file://{temp_file}"),
141
+ name="replacement",
142
+ path=temp_file,
143
+ )
144
+
145
+ manager.add_resource(resource1)
146
+ result = manager.add_resource(resource2)
147
+
148
+ # Should keep the original
149
  resources = manager.list_resources()
150
  assert len(resources) == 1
151
+ assert resources[0].name == "original"
152
+ # Result should be the original resource
153
+ assert result.name == "original"
154
+
155
+ def test_warn_on_duplicate_templates(self, caplog):
156
+ """Test warning on duplicate templates."""
157
+ manager = ResourceManager(duplicate_behavior="warn")
158
+
159
+ def template_fn(id: str) -> str:
160
+ return f"Template {id}"
161
+
162
+ template = ResourceTemplate.from_function(
163
+ fn=template_fn,
164
+ uri_template="test://{id}",
165
+ name="test_template",
166
+ )
167
+
168
+ manager.add_template(template)
169
+ manager.add_template(template)
170
+
171
+ assert "Resource already exists" in caplog.text
172
+ # Should have the template
173
+ assert len(manager.list_templates()) == 1
174
+
175
+ def test_error_on_duplicate_templates(self):
176
+ """Test error on duplicate templates."""
177
+ manager = ResourceManager(duplicate_behavior="error")
178
+
179
+ def template_fn(id: str) -> str:
180
+ return f"Template {id}"
181
+
182
+ template = ResourceTemplate.from_function(
183
+ fn=template_fn,
184
+ uri_template="test://{id}",
185
+ name="test_template",
186
+ )
187
+
188
+ manager.add_template(template)
189
+
190
+ with pytest.raises(ValueError, match="Resource already exists"):
191
+ manager.add_template(template)
192
+
193
+ def test_replace_duplicate_templates(self):
194
+ """Test replacing duplicate templates."""
195
+ manager = ResourceManager(duplicate_behavior="replace")
196
+
197
+ def original_fn(id: str) -> str:
198
+ return f"Original {id}"
199
+
200
+ def replacement_fn(id: str) -> str:
201
+ return f"Replacement {id}"
202
+
203
+ template1 = ResourceTemplate.from_function(
204
+ fn=original_fn,
205
+ uri_template="test://{id}",
206
+ name="original",
207
+ )
208
+
209
+ template2 = ResourceTemplate.from_function(
210
+ fn=replacement_fn,
211
+ uri_template="test://{id}",
212
+ name="replacement",
213
+ )
214
+
215
+ manager.add_template(template1)
216
+ manager.add_template(template2)
217
+
218
+ # Should have replaced with the new template
219
+ templates = manager.list_templates()
220
+ assert len(templates) == 1
221
+ assert templates[0].name == "replacement"
222
+
223
+ def test_ignore_duplicate_templates(self):
224
+ """Test ignoring duplicate templates."""
225
+ manager = ResourceManager(duplicate_behavior="ignore")
226
+
227
+ def original_fn(id: str) -> str:
228
+ return f"Original {id}"
229
+
230
+ def replacement_fn(id: str) -> str:
231
+ return f"Replacement {id}"
232
+
233
+ template1 = ResourceTemplate.from_function(
234
+ fn=original_fn,
235
+ uri_template="test://{id}",
236
+ name="original",
237
+ )
238
+
239
+ template2 = ResourceTemplate.from_function(
240
+ fn=replacement_fn,
241
+ uri_template="test://{id}",
242
+ name="replacement",
243
+ )
244
+
245
+ manager.add_template(template1)
246
+ result = manager.add_template(template2)
247
+
248
+ # Should keep the original
249
+ templates = manager.list_templates()
250
+ assert len(templates) == 1
251
+ assert templates[0].name == "original"
252
+ # Result should be the original template
253
+ assert result.name == "original"
254
 
255
  @pytest.mark.anyio
256
  async def test_get_resource(self, temp_file: Path):
tests/tools/test_tool_manager.py CHANGED
@@ -5,7 +5,6 @@ import pytest
5
  from pydantic import BaseModel
6
 
7
  from fastmcp.exceptions import ToolError
8
- from fastmcp.settings import DuplicateBehavior
9
  from fastmcp.tools import ToolManager
10
  from fastmcp.tools.tool import Tool
11
 
@@ -88,15 +87,17 @@ class TestAddTools:
88
 
89
  def test_warn_on_duplicate_tools(self, caplog):
90
  """Test warning on duplicate tools."""
 
91
 
92
- def f(x: int) -> int:
93
  return x
94
 
95
- manager = ToolManager(duplicate_behavior=DuplicateBehavior.WARN)
96
- manager.add_tool_from_fn(f)
97
- with caplog.at_level(logging.WARNING):
98
- manager.add_tool_from_fn(f)
99
- assert "Tool already exists: f" in caplog.text
 
100
 
101
  def test_disable_warn_on_duplicate_tools(self, caplog):
102
  """Test disabling warning on duplicate tools."""
@@ -104,7 +105,7 @@ class TestAddTools:
104
  def f(x: int) -> int:
105
  return x
106
 
107
- manager = ToolManager(duplicate_behavior=DuplicateBehavior.IGNORE)
108
  manager.add_tool_from_fn(f)
109
  with caplog.at_level(logging.WARNING):
110
  manager.add_tool_from_fn(f)
@@ -112,18 +113,19 @@ class TestAddTools:
112
 
113
  def test_error_on_duplicate_tools(self):
114
  """Test error on duplicate tools."""
 
115
 
116
- def f(x: int) -> int:
117
  return x
118
 
119
- manager = ToolManager(duplicate_behavior=DuplicateBehavior.ERROR)
120
- manager.add_tool_from_fn(f)
121
 
122
- with pytest.raises(ValueError, match="Tool already exists"):
123
- manager.add_tool_from_fn(f)
124
 
125
  def test_replace_duplicate_tools(self):
126
  """Test replacing duplicate tools."""
 
127
 
128
  def original_fn(x: int) -> int:
129
  return x
@@ -131,20 +133,33 @@ class TestAddTools:
131
  def replacement_fn(x: int) -> int:
132
  return x * 2
133
 
134
- manager = ToolManager(duplicate_behavior=DuplicateBehavior.REPLACE)
135
  manager.add_tool_from_fn(original_fn, name="test_tool")
136
- replacement_tool = manager.add_tool_from_fn(replacement_fn, name="test_tool")
137
 
138
- # Should have replaced the first tool with the second
139
- stored_tool = manager.get_tool("test_tool")
140
- assert stored_tool is not None
141
- assert stored_tool == replacement_tool
142
 
143
- # The name should still be the same
144
- assert stored_tool.name == "test_tool"
 
145
 
146
- # But the function is different
147
- assert stored_tool.fn.__name__ == "replacement_fn"
 
 
 
 
 
 
 
 
 
 
 
 
 
148
 
149
 
150
  class TestToolTags:
@@ -630,7 +645,7 @@ class TestCustomToolNames:
630
  assert target_manager.get_tool("prefix/source_fn") is None
631
 
632
  def test_replace_tool_keeps_original_name(self):
633
- """Test that replacing a tool with DuplicateBehavior.REPLACE keeps the original name."""
634
 
635
  def original_fn(x: int) -> int:
636
  return x
@@ -639,7 +654,7 @@ class TestCustomToolNames:
639
  return x * 2
640
 
641
  # Create a manager with REPLACE behavior
642
- manager = ToolManager(duplicate_behavior=DuplicateBehavior.REPLACE)
643
 
644
  # Add the original tool
645
  original_tool = manager.add_tool_from_fn(original_fn, name="test_tool")
 
5
  from pydantic import BaseModel
6
 
7
  from fastmcp.exceptions import ToolError
 
8
  from fastmcp.tools import ToolManager
9
  from fastmcp.tools.tool import Tool
10
 
 
87
 
88
  def test_warn_on_duplicate_tools(self, caplog):
89
  """Test warning on duplicate tools."""
90
+ manager = ToolManager(duplicate_behavior="warn")
91
 
92
+ def test_fn(x: int) -> int:
93
  return x
94
 
95
+ manager.add_tool_from_fn(test_fn, name="test_tool")
96
+ manager.add_tool_from_fn(test_fn, name="test_tool")
97
+
98
+ assert "Tool already exists: test_tool" in caplog.text
99
+ # Should have the tool
100
+ assert manager.get_tool("test_tool") is not None
101
 
102
  def test_disable_warn_on_duplicate_tools(self, caplog):
103
  """Test disabling warning on duplicate tools."""
 
105
  def f(x: int) -> int:
106
  return x
107
 
108
+ manager = ToolManager(duplicate_behavior="ignore")
109
  manager.add_tool_from_fn(f)
110
  with caplog.at_level(logging.WARNING):
111
  manager.add_tool_from_fn(f)
 
113
 
114
  def test_error_on_duplicate_tools(self):
115
  """Test error on duplicate tools."""
116
+ manager = ToolManager(duplicate_behavior="error")
117
 
118
+ def test_fn(x: int) -> int:
119
  return x
120
 
121
+ manager.add_tool_from_fn(test_fn, name="test_tool")
 
122
 
123
+ with pytest.raises(ValueError, match="Tool already exists: test_tool"):
124
+ manager.add_tool_from_fn(test_fn, name="test_tool")
125
 
126
  def test_replace_duplicate_tools(self):
127
  """Test replacing duplicate tools."""
128
+ manager = ToolManager(duplicate_behavior="replace")
129
 
130
  def original_fn(x: int) -> int:
131
  return x
 
133
  def replacement_fn(x: int) -> int:
134
  return x * 2
135
 
 
136
  manager.add_tool_from_fn(original_fn, name="test_tool")
137
+ manager.add_tool_from_fn(replacement_fn, name="test_tool")
138
 
139
+ # Should have replaced with the new function
140
+ tool = manager.get_tool("test_tool")
141
+ assert tool is not None
142
+ assert tool.fn.__name__ == "replacement_fn"
143
 
144
+ def test_ignore_duplicate_tools(self):
145
+ """Test ignoring duplicate tools."""
146
+ manager = ToolManager(duplicate_behavior="ignore")
147
 
148
+ def original_fn(x: int) -> int:
149
+ return x
150
+
151
+ def replacement_fn(x: int) -> int:
152
+ return x * 2
153
+
154
+ manager.add_tool_from_fn(original_fn, name="test_tool")
155
+ result = manager.add_tool_from_fn(replacement_fn, name="test_tool")
156
+
157
+ # Should keep the original
158
+ tool = manager.get_tool("test_tool")
159
+ assert tool is not None
160
+ assert tool.fn.__name__ == "original_fn"
161
+ # Result should be the original tool
162
+ assert result.fn.__name__ == "original_fn"
163
 
164
 
165
  class TestToolTags:
 
645
  assert target_manager.get_tool("prefix/source_fn") is None
646
 
647
  def test_replace_tool_keeps_original_name(self):
648
+ """Test that replacing a tool with "replace" keeps the original name."""
649
 
650
  def original_fn(x: int) -> int:
651
  return x
 
654
  return x * 2
655
 
656
  # Create a manager with REPLACE behavior
657
+ manager = ToolManager(duplicate_behavior="replace")
658
 
659
  # Add the original tool
660
  original_tool = manager.add_tool_from_fn(original_fn, name="test_tool")