Jeremiah Lowin commited on
Commit
0631483
·
unverified ·
2 Parent(s): 2504ec5f07a282

Merge pull request #169 from jlowin/manager-bug

Browse files
docs/servers/fastmcp.mdx CHANGED
@@ -308,18 +308,17 @@ Server behavior, like transport settings (host, port for SSE) and how duplicate
308
 
309
  ```python
310
  from fastmcp import FastMCP
311
- from fastmcp.settings import DuplicateBehavior
312
 
313
  # Configure during initialization
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
@@ -331,5 +330,4 @@ print(mcp.settings.on_duplicate_tools) # Output: DuplicateBehavior.ERROR
331
  - **`on_duplicate_resources`**: How to handle duplicate resource registrations
332
  - **`on_duplicate_prompts`**: How to handle duplicate prompt registrations
333
 
334
- All of these can be configured directly as parameters when creating the `FastMCP` instance.
335
-
 
308
 
309
  ```python
310
  from fastmcp import FastMCP
 
311
 
312
  # Configure during initialization
313
  mcp = FastMCP(
314
  name="ConfiguredServer",
315
  port=8080, # Directly maps to ServerSettings
316
+ on_duplicate_tools="error" # Set duplicate handling
317
  )
318
 
319
  # Settings are accessible via mcp.settings
320
  print(mcp.settings.port) # Output: 8080
321
+ print(mcp.settings.on_duplicate_tools) # Output: "error"
322
  ```
323
 
324
  ### Key Configuration Options
 
330
  - **`on_duplicate_resources`**: How to handle duplicate resource registrations
331
  - **`on_duplicate_prompts`**: How to handle duplicate prompt registrations
332
 
333
+ All of these can be configured directly as parameters when creating the `FastMCP` instance.
 
docs/servers/prompts.mdx CHANGED
@@ -205,25 +205,24 @@ You can configure how the FastMCP server handles attempts to register multiple p
205
 
206
  ```python
207
  from fastmcp import FastMCP
208
- 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()
216
  def greeting(): return "Hello, how can I help you today?"
217
 
218
  # This registration attempt will raise a ValueError because
219
- # "greeting" is already registered and the behavior is ERROR.
220
  # @mcp.prompt()
221
  # def greeting(): return "Hi there! What can I do for you?"
222
  ```
223
 
224
- The `DuplicateBehavior` enum options are:
225
 
226
- - `WARN` (default): Logs a warning, and the new prompt replaces the old one.
227
- - `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
228
- - `REPLACE`: Silently replaces the existing prompt with the new one.
229
- - `IGNORE`: Keeps the original prompt and ignores the new registration attempt.
 
205
 
206
  ```python
207
  from fastmcp import FastMCP
 
208
 
209
  mcp = FastMCP(
210
  name="PromptServer",
211
+ on_duplicate_prompts="error" # Raise an error if a prompt name is duplicated
212
  )
213
 
214
  @mcp.prompt()
215
  def greeting(): return "Hello, how can I help you today?"
216
 
217
  # This registration attempt will raise a ValueError because
218
+ # "greeting" is already registered and the behavior is "error".
219
  # @mcp.prompt()
220
  # def greeting(): return "Hi there! What can I do for you?"
221
  ```
222
 
223
+ The duplicate behavior options are:
224
 
225
+ - `"warn"` (default): Logs a warning, and the new prompt replaces the old one.
226
+ - `"error"`: Raises a `ValueError`, preventing the duplicate registration.
227
+ - `"replace"`: Silently replaces the existing prompt with the new one.
228
+ - `"ignore"`: Keeps the original prompt and ignores the new registration attempt.
docs/servers/resources.mdx CHANGED
@@ -297,25 +297,24 @@ You can configure how the FastMCP server handles attempts to register multiple r
297
 
298
  ```python
299
  from fastmcp import FastMCP
300
- 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")
308
  def get_config_v1(): return {"version": 1}
309
 
310
  # This registration attempt will raise a ValueError because
311
- # "data://config" is already registered and the behavior is ERROR.
312
  # @mcp.resource("data://config")
313
  # def get_config_v2(): return {"version": 2}
314
  ```
315
 
316
- The `DuplicateBehavior` enum options are:
317
 
318
- - `WARN` (default): Logs a warning, and the new resource/template replaces the old one.
319
- - `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
320
- - `REPLACE`: Silently replaces the existing resource/template with the new one.
321
- - `IGNORE`: Keeps the original resource/template and ignores the new registration attempt.
 
297
 
298
  ```python
299
  from fastmcp import FastMCP
 
300
 
301
  mcp = FastMCP(
302
  name="ResourceServer",
303
+ on_duplicate_resources="error" # Raise error on duplicates
304
  )
305
 
306
  @mcp.resource("data://config")
307
  def get_config_v1(): return {"version": 1}
308
 
309
  # This registration attempt will raise a ValueError because
310
+ # "data://config" is already registered and the behavior is "error".
311
  # @mcp.resource("data://config")
312
  # def get_config_v2(): return {"version": 2}
313
  ```
314
 
315
+ The duplicate behavior options are:
316
 
317
+ - `"warn"` (default): Logs a warning, and the new resource/template replaces the old one.
318
+ - `"error"`: Raises a `ValueError`, preventing the duplicate registration.
319
+ - `"replace"`: Silently replaces the existing resource/template with the new one.
320
+ - `"ignore"`: Keeps the original resource/template and ignores the new registration attempt.
docs/servers/resources_backup.mdx DELETED
@@ -1,270 +0,0 @@
1
- ---
2
- title: Resources & Templates
3
- sidebarTitle: Resources & Templates
4
- description: Expose data sources and dynamic content generators to your MCP client.
5
- icon: database
6
- ---
7
-
8
- Resources represent data or files that an MCP client can read, and resource templates extend this concept by allowing clients to request dynamically generated resources based on parameters passed in the URI.
9
-
10
- FastMCP simplifies defining both static and dynamic resources, primarily using the `@mcp.resource` decorator.
11
-
12
- ## What Are Resources?
13
-
14
- Resources provide read-only access to data for the LLM or client application. When a client requests a resource URI:
15
-
16
- 1. FastMCP finds the corresponding resource definition.
17
- 2. If it's dynamic (defined by a function), the function is executed.
18
- 3. The content (text, JSON, binary data) is returned to the client.
19
-
20
- This allows LLMs to access files, database content, configuration, or dynamically generated information relevant to the conversation.
21
-
22
- ## Defining Resources with `@mcp.resource`
23
-
24
- The most common way to define a resource is by decorating a Python function. The decorator requires the resource's unique URI.
25
-
26
- ```python
27
- import json
28
- from fastmcp import FastMCP
29
-
30
- mcp = FastMCP(name="DataServer")
31
-
32
- # Basic dynamic resource returning a string
33
- @mcp.resource("resource://greeting")
34
- def get_greeting() -> str:
35
- """Provides a simple greeting message."""
36
- return "Hello from FastMCP Resources!"
37
-
38
- # Resource returning JSON data (dict is auto-serialized)
39
- @mcp.resource("data://config")
40
- def get_config() -> dict:
41
- """Provides application configuration as JSON."""
42
- return {
43
- "theme": "dark",
44
- "version": "1.2.0",
45
- "features": ["tools", "resources"],
46
- }
47
- ```
48
-
49
- **Key Concepts:**
50
-
51
- * **URI:** The first argument to `@resource` is the unique URI (e.g., `"resource://greeting"`) clients use to request this data.
52
- * **Lazy Loading:** The decorated function (`get_greeting`, `get_config`) is only executed when a client specifically requests that resource URI via `resources/read`.
53
- * **Inferred Metadata:** By default:
54
- * Resource Name: Taken from the function name (`get_greeting`).
55
- * Resource Description: Taken from the function's docstring.
56
-
57
- ### Return Value Handling
58
-
59
- FastMCP automatically converts your function's return value into the appropriate MCP resource content:
60
-
61
- - **`str`**: Sent as `TextResourceContents` (with `mime_type="text/plain"` by default).
62
- - **`dict`, `list`, `pydantic.BaseModel`**: Automatically serialized to a JSON string and sent as `TextResourceContents` (with `mime_type="application/json"` by default).
63
- - **`bytes`**: Base64 encoded and sent as `BlobResourceContents`. You should specify an appropriate `mime_type` (e.g., `"image/png"`, `"application/octet-stream"`).
64
- - **`None`**: Results in an empty resource content list being returned.
65
-
66
- ### Resource Metadata
67
-
68
- You can customize the resource's properties using arguments in the decorator:
69
-
70
- ```python
71
- from fastmcp import FastMCP
72
-
73
- mcp = FastMCP(name="DataServer")
74
-
75
- # Example specifying metadata
76
- @mcp.resource(
77
- uri="data://app-status", # Explicit URI (required)
78
- name="ApplicationStatus", # Custom name
79
- description="Provides the current status of the application.", # Custom description
80
- mime_type="application/json", # Explicit MIME type
81
- tags={"monitoring", "status"} # Categorization tags
82
- )
83
- def get_application_status() -> dict:
84
- """Internal function description (ignored if description is provided above)."""
85
- return {"status": "ok", "uptime": 12345, "version": mcp.settings.version} # Example usage
86
- ```
87
-
88
- - **`uri`**: The unique identifier for the resource (required).
89
- - **`name`**: A human-readable name (defaults to function name).
90
- - **`description`**: Explanation of the resource (defaults to docstring).
91
- - **`mime_type`**: Specifies the content type (FastMCP often infers a default like `text/plain` or `application/json`, but explicit is better for non-text types).
92
- - **`tags`**: A set of strings for categorization, potentially used by clients for filtering.
93
-
94
- ### Using Context in Resources
95
-
96
- Like tools, resource functions can request the `Context` object to access MCP session capabilities.
97
-
98
- ```python
99
- from fastmcp import FastMCP, Context
100
- import datetime
101
-
102
- mcp = FastMCP(name="DataServer")
103
-
104
- @mcp.resource("data://server-info", tags={"server", "info"})
105
- async def get_server_info(ctx: Context) -> dict:
106
- """Provides information about the server using context."""
107
- await ctx.info(f"Generating server info resource for request {ctx.request_id}")
108
- # You could potentially read other resources via ctx.read_resource here
109
- return {
110
- "server_name": mcp.name,
111
- "timestamp": datetime.datetime.now(datetime.UTC).isoformat(),
112
- "client_id": ctx.client_id or "N/A",
113
- "log_level": mcp.settings.log_level,
114
- }
115
- ```
116
-
117
- ### Asynchronous Resources
118
-
119
- Use `async def` for resource functions that perform I/O operations (e.g., reading from a database or network) to avoid blocking the server.
120
-
121
- ```python
122
- import aiofiles
123
- from fastmcp import FastMCP
124
-
125
- mcp = FastMCP(name="DataServer")
126
-
127
- @mcp.resource("file:///app/data/important_log.txt", mime_type="text/plain")
128
- async def read_important_log() -> str:
129
- """Reads content from a specific log file asynchronously."""
130
- try:
131
- async with aiofiles.open("/app/data/important_log.txt", mode="r") as f:
132
- content = await f.read()
133
- return content
134
- except FileNotFoundError:
135
- return "Log file not found."
136
- ```
137
-
138
- ## (Alternative) Defining Static Resources
139
-
140
- While `@mcp.resource` is ideal for dynamic content, you can directly register pre-defined resources (like static files or simple text) using `mcp.add_resource()` and concrete `Resource` subclasses.
141
-
142
- ```python
143
- from pathlib import Path
144
- from fastmcp import FastMCP
145
- from fastmcp.resources import FileResource, TextResource, DirectoryResource
146
-
147
- mcp = FastMCP(name="DataServer")
148
-
149
- # 1. Exposing a static file directly
150
- readme_path = Path("./README.md").resolve()
151
- if readme_path.exists():
152
- # Use a file:// URI scheme
153
- readme_resource = FileResource(
154
- uri=f"file://{readme_path.as_posix()}",
155
- path=readme_path, # Path to the actual file
156
- name="README File",
157
- description="The project's README.",
158
- mime_type="text/markdown",
159
- tags={"documentation"}
160
- )
161
- mcp.add_resource(readme_resource)
162
-
163
- # 2. Exposing simple, predefined text
164
- notice_resource = TextResource(
165
- uri="resource://notice",
166
- name="Important Notice",
167
- text="System maintenance scheduled for Sunday.",
168
- tags={"notification"}
169
- )
170
- mcp.add_resource(notice_resource)
171
-
172
- # 3. Exposing a directory listing
173
- data_dir_path = Path("./app_data").resolve()
174
- if data_dir_path.is_dir():
175
- data_listing_resource = DirectoryResource(
176
- uri="resource://data-files",
177
- path=data_dir_path, # Path to the directory
178
- name="Data Directory Listing",
179
- description="Lists files available in the data directory.",
180
- recursive=False # Set to True to list subdirectories
181
- )
182
- mcp.add_resource(data_listing_resource) # Returns JSON list of files
183
- ```
184
-
185
- **Common Resource Classes:**
186
-
187
- - `TextResource`: For simple string content.
188
- - `BinaryResource`: For raw `bytes` content.
189
- - `FileResource`: Reads content from a local file path. Handles text/binary modes and lazy reading.
190
- - `HttpResource`: Fetches content from an HTTP(S) URL (requires `httpx`).
191
- - `DirectoryResource`: Lists files in a local directory (returns JSON).
192
- - (`FunctionResource`: Internal class used by `@mcp.resource`).
193
-
194
- Use these when the content is static or sourced directly from a file/URL, bypassing the need for a dedicated Python function.
195
-
196
- ## Defining Resource Templates
197
-
198
- Resource Templates allow clients to request resources whose content depends on parameters embedded in the URI. Define a template using the **same `@mcp.resource` decorator**, but include `{parameter_name}` placeholders in the URI string and add corresponding arguments to your function signature.
199
-
200
- ```python
201
- from fastmcp import FastMCP
202
-
203
- mcp = FastMCP(name="DataServer")
204
-
205
- # Template URI includes {city} placeholder
206
- @mcp.resource("data://weather/{city}")
207
- # Function accepts 'city' parameter matching the placeholder
208
- def get_weather_for_city(city: str) -> dict:
209
- """Provides weather information for a specific city."""
210
- print(f"Server: Generating weather for city: {city}...")
211
- # In reality, call a weather API using the 'city' parameter
212
- temp = 20 + len(city) % 5 # Dummy logic
213
- condition = "Sunny" if len(city) % 2 == 0 else "Cloudy"
214
- return {"city": city.capitalize(), "temperature": temp, "unit": "celsius", "condition": condition}
215
-
216
- # Template with an integer parameter
217
- @mcp.resource("users://{user_id}/profile")
218
- async def get_user_profile(user_id: int) -> dict:
219
- """Retrieves a user's profile information by ID."""
220
- print(f"Server: Generating profile for user ID: {user_id}...")
221
- # In reality, fetch from database using user_id
222
- # FastMCP uses Pydantic to auto-convert the string URI part to int
223
- if user_id == 1:
224
- return {"id": user_id, "name": "Alice", "email": "alice@example.com", "status": "active"}
225
- elif user_id == 2:
226
- return {"id": user_id, "name": "Bob", "email": "bob@example.com", "status": "inactive"}
227
- else:
228
- # Example of returning an error structure
229
- return {"error": f"User with ID {user_id} not found"}
230
- ```
231
-
232
- **How Templates Work:**
233
-
234
- 1. **Definition:** When FastMCP sees `{...}` placeholders in the `@resource` URI and matching function parameters, it registers a `ResourceTemplate`.
235
- 2. **Discovery:** Clients list templates via `resources/listResourceTemplates`.
236
- 3. **Request & Matching:** A client requests a specific URI, e.g., `data://weather/london`. FastMCP matches this to the `data://weather/{city}` template.
237
- 4. **Parameter Extraction:** It extracts the parameter value: `city="london"`.
238
- 5. **Type Conversion & Function Call:** It converts the extracted string `"london"` to the type hinted in the function (`str` in this case) and calls `get_weather_for_city(city="london")`. For `users://1/profile`, it converts `"1"` to `int` before calling `get_user_profile(user_id=1)`.
239
- 6. **Response:** The function's return value is formatted (e.g., dict to JSON) and sent back as the content of the requested resource URI (`data://weather/london`).
240
-
241
- Templates provide a powerful way to expose parameterized data access points following REST-like principles.
242
-
243
- ## Server Behavior: Handling Duplicate Resources
244
-
245
- You can configure how the FastMCP server handles attempts to register multiple resources or templates with the same URI. Use the `on_duplicate_resources` setting during `FastMCP` initialization.
246
-
247
- ```python
248
- from fastmcp import FastMCP
249
- 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")
257
- def get_config_v1(): return {"version": 1}
258
-
259
- # This registration attempt will raise a ValueError because
260
- # "data://config" is already registered and the behavior is ERROR.
261
- # @mcp.resource("data://config")
262
- # def get_config_v2(): return {"version": 2}
263
- ```
264
-
265
- The `DuplicateBehavior` enum options are:
266
-
267
- - `WARN` (default): Logs a warning, and the new resource/template replaces the old one.
268
- - `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
269
- - `REPLACE`: Silently replaces the existing resource/template with the new one.
270
- - `IGNORE`: Keeps the original resource/template and ignores the new registration attempt.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
docs/servers/tools.mdx CHANGED
@@ -310,26 +310,25 @@ You can control how the FastMCP server behaves if you try to register multiple t
310
 
311
  ```python
312
  from fastmcp import FastMCP
313
- from fastmcp.settings import DuplicateBehavior
314
 
315
  mcp = FastMCP(
316
  name="StrictServer",
317
  # Configure behavior for duplicate tool names
318
- on_duplicate_tools=DuplicateBehavior.ERROR
319
  )
320
 
321
  @mcp.tool()
322
  def my_tool(): return "Version 1"
323
 
324
  # This will now raise a ValueError because 'my_tool' already exists
325
- # and on_duplicate_tools is set to ERROR.
326
  # @mcp.tool()
327
  # def my_tool(): return "Version 2"
328
  ```
329
 
330
- The `DuplicateBehavior` enum options are:
331
 
332
- - `WARN` (default): Logs a warning and the new tool replaces the old one.
333
- - `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
334
- - `REPLACE`: Silently replaces the existing tool with the new one.
335
- - `IGNORE`: Keeps the original tool and ignores the new registration attempt.
 
310
 
311
  ```python
312
  from fastmcp import FastMCP
 
313
 
314
  mcp = FastMCP(
315
  name="StrictServer",
316
  # Configure behavior for duplicate tool names
317
+ on_duplicate_tools="error"
318
  )
319
 
320
  @mcp.tool()
321
  def my_tool(): return "Version 1"
322
 
323
  # This will now raise a ValueError because 'my_tool' already exists
324
+ # and on_duplicate_tools is set to "error".
325
  # @mcp.tool()
326
  # def my_tool(): return "Version 2"
327
  ```
328
 
329
+ The duplicate behavior options are:
330
 
331
+ - `"warn"` (default): Logs a warning and the new tool replaces the old one.
332
+ - `"error"`: Raises a `ValueError`, preventing the duplicate registration.
333
+ - `"replace"`: Silently replaces the existing tool with the new one.
334
+ - `"ignore"`: Keeps the original tool and ignores the new registration attempt.
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(
@@ -104,26 +115,28 @@ class ResourceManager:
104
  Args:
105
  resource: A Resource instance to add
106
  """
 
107
  logger.debug(
108
  "Adding resource",
109
  extra={
110
- "uri": resource.uri,
111
  "type": type(resource).__name__,
112
  "resource_name": resource.name,
113
  },
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(
@@ -157,26 +170,28 @@ class ResourceManager:
157
  The added template. If a template with the same URI already exists,
158
  returns the existing template.
159
  """
 
160
  logger.debug(
161
  "Adding resource",
162
  extra={
163
- "uri": template.uri_template,
164
  "type": type(template).__name__,
165
  "resource_name": template.name,
166
  },
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(
 
115
  Args:
116
  resource: A Resource instance to add
117
  """
118
+ uri_str = str(resource.uri)
119
  logger.debug(
120
  "Adding resource",
121
  extra={
122
+ "uri": uri_str,
123
  "type": type(resource).__name__,
124
  "resource_name": resource.name,
125
  },
126
  )
127
+ existing = self._resources.get(uri_str)
128
  if existing:
129
+ if self.duplicate_behavior == "warn":
130
+ logger.warning(f"Resource already exists: {uri_str}")
131
+ self._resources[uri_str] = resource
132
+ elif self.duplicate_behavior == "replace":
133
+ self._resources[uri_str] = resource
134
+ elif self.duplicate_behavior == "error":
135
+ raise ValueError(f"Resource already exists: {uri_str}")
136
+ elif self.duplicate_behavior == "ignore":
137
+ return existing
138
+ else:
139
+ self._resources[uri_str] = resource
140
  return resource
141
 
142
  def add_template_from_fn(
 
170
  The added template. If a template with the same URI already exists,
171
  returns the existing template.
172
  """
173
+ uri_template_str = str(template.uri_template)
174
  logger.debug(
175
  "Adding resource",
176
  extra={
177
+ "uri": uri_template_str,
178
  "type": type(template).__name__,
179
  "resource_name": template.name,
180
  },
181
  )
182
+ existing = self._templates.get(uri_template_str)
183
  if existing:
184
+ if self.duplicate_behavior == "warn":
185
+ logger.warning(f"Resource already exists: {uri_template_str}")
186
+ self._templates[uri_template_str] = template
187
+ elif self.duplicate_behavior == "replace":
188
+ self._templates[uri_template_str] = template
189
+ elif self.duplicate_behavior == "error":
190
+ raise ValueError(f"Resource already exists: {uri_template_str}")
191
+ elif self.duplicate_behavior == "ignore":
192
+ return existing
193
+ else:
194
+ self._templates[uri_template_str] = template
195
  return template
196
 
197
  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")