mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Merge pull request #169 from jlowin/manager-bug
Fix bug with duplicate behavior == ignore
This commit is contained in:
Jeremiah Lowin
GitHub
commit
e7145a1398
12 changed files with 365 additions and 421 deletions
|
|
@ -308,18 +308,17 @@ Server behavior, like transport settings (host, port for SSE) and how duplicate
|
|||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
|
||||
# Configure during initialization
|
||||
mcp = FastMCP(
|
||||
name="ConfiguredServer",
|
||||
port=8080, # Directly maps to ServerSettings
|
||||
on_duplicate_tools=DuplicateBehavior.ERROR # Set duplicate handling
|
||||
on_duplicate_tools="error" # Set duplicate handling
|
||||
)
|
||||
|
||||
# Settings are accessible via mcp.settings
|
||||
print(mcp.settings.port) # Output: 8080
|
||||
print(mcp.settings.on_duplicate_tools) # Output: DuplicateBehavior.ERROR
|
||||
print(mcp.settings.on_duplicate_tools) # Output: "error"
|
||||
```
|
||||
|
||||
### Key Configuration Options
|
||||
|
|
@ -331,5 +330,4 @@ print(mcp.settings.on_duplicate_tools) # Output: DuplicateBehavior.ERROR
|
|||
- **`on_duplicate_resources`**: How to handle duplicate resource registrations
|
||||
- **`on_duplicate_prompts`**: How to handle duplicate prompt registrations
|
||||
|
||||
All of these can be configured directly as parameters when creating the `FastMCP` instance.
|
||||
|
||||
All of these can be configured directly as parameters when creating the `FastMCP` instance.
|
||||
|
|
@ -205,25 +205,24 @@ You can configure how the FastMCP server handles attempts to register multiple p
|
|||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
|
||||
mcp = FastMCP(
|
||||
name="PromptServer",
|
||||
on_duplicate_prompts=DuplicateBehavior.ERROR # Raise an error if a prompt name is duplicated
|
||||
on_duplicate_prompts="error" # Raise an error if a prompt name is duplicated
|
||||
)
|
||||
|
||||
@mcp.prompt()
|
||||
def greeting(): return "Hello, how can I help you today?"
|
||||
|
||||
# This registration attempt will raise a ValueError because
|
||||
# "greeting" is already registered and the behavior is ERROR.
|
||||
# "greeting" is already registered and the behavior is "error".
|
||||
# @mcp.prompt()
|
||||
# def greeting(): return "Hi there! What can I do for you?"
|
||||
```
|
||||
|
||||
The `DuplicateBehavior` enum options are:
|
||||
The duplicate behavior options are:
|
||||
|
||||
- `WARN` (default): Logs a warning, and the new prompt replaces the old one.
|
||||
- `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
|
||||
- `REPLACE`: Silently replaces the existing prompt with the new one.
|
||||
- `IGNORE`: Keeps the original prompt and ignores the new registration attempt.
|
||||
- `"warn"` (default): Logs a warning, and the new prompt replaces the old one.
|
||||
- `"error"`: Raises a `ValueError`, preventing the duplicate registration.
|
||||
- `"replace"`: Silently replaces the existing prompt with the new one.
|
||||
- `"ignore"`: Keeps the original prompt and ignores the new registration attempt.
|
||||
|
|
@ -297,25 +297,24 @@ You can configure how the FastMCP server handles attempts to register multiple r
|
|||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
|
||||
mcp = FastMCP(
|
||||
name="ResourceServer",
|
||||
on_duplicate_resources=DuplicateBehavior.ERROR # Raise error on duplicates
|
||||
on_duplicate_resources="error" # Raise error on duplicates
|
||||
)
|
||||
|
||||
@mcp.resource("data://config")
|
||||
def get_config_v1(): return {"version": 1}
|
||||
|
||||
# This registration attempt will raise a ValueError because
|
||||
# "data://config" is already registered and the behavior is ERROR.
|
||||
# "data://config" is already registered and the behavior is "error".
|
||||
# @mcp.resource("data://config")
|
||||
# def get_config_v2(): return {"version": 2}
|
||||
```
|
||||
|
||||
The `DuplicateBehavior` enum options are:
|
||||
The duplicate behavior options are:
|
||||
|
||||
- `WARN` (default): Logs a warning, and the new resource/template replaces the old one.
|
||||
- `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
|
||||
- `REPLACE`: Silently replaces the existing resource/template with the new one.
|
||||
- `IGNORE`: Keeps the original resource/template and ignores the new registration attempt.
|
||||
- `"warn"` (default): Logs a warning, and the new resource/template replaces the old one.
|
||||
- `"error"`: Raises a `ValueError`, preventing the duplicate registration.
|
||||
- `"replace"`: Silently replaces the existing resource/template with the new one.
|
||||
- `"ignore"`: Keeps the original resource/template and ignores the new registration attempt.
|
||||
|
|
@ -1,270 +0,0 @@
|
|||
---
|
||||
title: Resources & Templates
|
||||
sidebarTitle: Resources & Templates
|
||||
description: Expose data sources and dynamic content generators to your MCP client.
|
||||
icon: database
|
||||
---
|
||||
|
||||
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.
|
||||
|
||||
FastMCP simplifies defining both static and dynamic resources, primarily using the `@mcp.resource` decorator.
|
||||
|
||||
## What Are Resources?
|
||||
|
||||
Resources provide read-only access to data for the LLM or client application. When a client requests a resource URI:
|
||||
|
||||
1. FastMCP finds the corresponding resource definition.
|
||||
2. If it's dynamic (defined by a function), the function is executed.
|
||||
3. The content (text, JSON, binary data) is returned to the client.
|
||||
|
||||
This allows LLMs to access files, database content, configuration, or dynamically generated information relevant to the conversation.
|
||||
|
||||
## Defining Resources with `@mcp.resource`
|
||||
|
||||
The most common way to define a resource is by decorating a Python function. The decorator requires the resource's unique URI.
|
||||
|
||||
```python
|
||||
import json
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP(name="DataServer")
|
||||
|
||||
# Basic dynamic resource returning a string
|
||||
@mcp.resource("resource://greeting")
|
||||
def get_greeting() -> str:
|
||||
"""Provides a simple greeting message."""
|
||||
return "Hello from FastMCP Resources!"
|
||||
|
||||
# Resource returning JSON data (dict is auto-serialized)
|
||||
@mcp.resource("data://config")
|
||||
def get_config() -> dict:
|
||||
"""Provides application configuration as JSON."""
|
||||
return {
|
||||
"theme": "dark",
|
||||
"version": "1.2.0",
|
||||
"features": ["tools", "resources"],
|
||||
}
|
||||
```
|
||||
|
||||
**Key Concepts:**
|
||||
|
||||
* **URI:** The first argument to `@resource` is the unique URI (e.g., `"resource://greeting"`) clients use to request this data.
|
||||
* **Lazy Loading:** The decorated function (`get_greeting`, `get_config`) is only executed when a client specifically requests that resource URI via `resources/read`.
|
||||
* **Inferred Metadata:** By default:
|
||||
* Resource Name: Taken from the function name (`get_greeting`).
|
||||
* Resource Description: Taken from the function's docstring.
|
||||
|
||||
### Return Value Handling
|
||||
|
||||
FastMCP automatically converts your function's return value into the appropriate MCP resource content:
|
||||
|
||||
- **`str`**: Sent as `TextResourceContents` (with `mime_type="text/plain"` by default).
|
||||
- **`dict`, `list`, `pydantic.BaseModel`**: Automatically serialized to a JSON string and sent as `TextResourceContents` (with `mime_type="application/json"` by default).
|
||||
- **`bytes`**: Base64 encoded and sent as `BlobResourceContents`. You should specify an appropriate `mime_type` (e.g., `"image/png"`, `"application/octet-stream"`).
|
||||
- **`None`**: Results in an empty resource content list being returned.
|
||||
|
||||
### Resource Metadata
|
||||
|
||||
You can customize the resource's properties using arguments in the decorator:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP(name="DataServer")
|
||||
|
||||
# Example specifying metadata
|
||||
@mcp.resource(
|
||||
uri="data://app-status", # Explicit URI (required)
|
||||
name="ApplicationStatus", # Custom name
|
||||
description="Provides the current status of the application.", # Custom description
|
||||
mime_type="application/json", # Explicit MIME type
|
||||
tags={"monitoring", "status"} # Categorization tags
|
||||
)
|
||||
def get_application_status() -> dict:
|
||||
"""Internal function description (ignored if description is provided above)."""
|
||||
return {"status": "ok", "uptime": 12345, "version": mcp.settings.version} # Example usage
|
||||
```
|
||||
|
||||
- **`uri`**: The unique identifier for the resource (required).
|
||||
- **`name`**: A human-readable name (defaults to function name).
|
||||
- **`description`**: Explanation of the resource (defaults to docstring).
|
||||
- **`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).
|
||||
- **`tags`**: A set of strings for categorization, potentially used by clients for filtering.
|
||||
|
||||
### Using Context in Resources
|
||||
|
||||
Like tools, resource functions can request the `Context` object to access MCP session capabilities.
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP, Context
|
||||
import datetime
|
||||
|
||||
mcp = FastMCP(name="DataServer")
|
||||
|
||||
@mcp.resource("data://server-info", tags={"server", "info"})
|
||||
async def get_server_info(ctx: Context) -> dict:
|
||||
"""Provides information about the server using context."""
|
||||
await ctx.info(f"Generating server info resource for request {ctx.request_id}")
|
||||
# You could potentially read other resources via ctx.read_resource here
|
||||
return {
|
||||
"server_name": mcp.name,
|
||||
"timestamp": datetime.datetime.now(datetime.UTC).isoformat(),
|
||||
"client_id": ctx.client_id or "N/A",
|
||||
"log_level": mcp.settings.log_level,
|
||||
}
|
||||
```
|
||||
|
||||
### Asynchronous Resources
|
||||
|
||||
Use `async def` for resource functions that perform I/O operations (e.g., reading from a database or network) to avoid blocking the server.
|
||||
|
||||
```python
|
||||
import aiofiles
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP(name="DataServer")
|
||||
|
||||
@mcp.resource("file:///app/data/important_log.txt", mime_type="text/plain")
|
||||
async def read_important_log() -> str:
|
||||
"""Reads content from a specific log file asynchronously."""
|
||||
try:
|
||||
async with aiofiles.open("/app/data/important_log.txt", mode="r") as f:
|
||||
content = await f.read()
|
||||
return content
|
||||
except FileNotFoundError:
|
||||
return "Log file not found."
|
||||
```
|
||||
|
||||
## (Alternative) Defining Static Resources
|
||||
|
||||
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.
|
||||
|
||||
```python
|
||||
from pathlib import Path
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.resources import FileResource, TextResource, DirectoryResource
|
||||
|
||||
mcp = FastMCP(name="DataServer")
|
||||
|
||||
# 1. Exposing a static file directly
|
||||
readme_path = Path("./README.md").resolve()
|
||||
if readme_path.exists():
|
||||
# Use a file:// URI scheme
|
||||
readme_resource = FileResource(
|
||||
uri=f"file://{readme_path.as_posix()}",
|
||||
path=readme_path, # Path to the actual file
|
||||
name="README File",
|
||||
description="The project's README.",
|
||||
mime_type="text/markdown",
|
||||
tags={"documentation"}
|
||||
)
|
||||
mcp.add_resource(readme_resource)
|
||||
|
||||
# 2. Exposing simple, predefined text
|
||||
notice_resource = TextResource(
|
||||
uri="resource://notice",
|
||||
name="Important Notice",
|
||||
text="System maintenance scheduled for Sunday.",
|
||||
tags={"notification"}
|
||||
)
|
||||
mcp.add_resource(notice_resource)
|
||||
|
||||
# 3. Exposing a directory listing
|
||||
data_dir_path = Path("./app_data").resolve()
|
||||
if data_dir_path.is_dir():
|
||||
data_listing_resource = DirectoryResource(
|
||||
uri="resource://data-files",
|
||||
path=data_dir_path, # Path to the directory
|
||||
name="Data Directory Listing",
|
||||
description="Lists files available in the data directory.",
|
||||
recursive=False # Set to True to list subdirectories
|
||||
)
|
||||
mcp.add_resource(data_listing_resource) # Returns JSON list of files
|
||||
```
|
||||
|
||||
**Common Resource Classes:**
|
||||
|
||||
- `TextResource`: For simple string content.
|
||||
- `BinaryResource`: For raw `bytes` content.
|
||||
- `FileResource`: Reads content from a local file path. Handles text/binary modes and lazy reading.
|
||||
- `HttpResource`: Fetches content from an HTTP(S) URL (requires `httpx`).
|
||||
- `DirectoryResource`: Lists files in a local directory (returns JSON).
|
||||
- (`FunctionResource`: Internal class used by `@mcp.resource`).
|
||||
|
||||
Use these when the content is static or sourced directly from a file/URL, bypassing the need for a dedicated Python function.
|
||||
|
||||
## Defining Resource Templates
|
||||
|
||||
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.
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP(name="DataServer")
|
||||
|
||||
# Template URI includes {city} placeholder
|
||||
@mcp.resource("data://weather/{city}")
|
||||
# Function accepts 'city' parameter matching the placeholder
|
||||
def get_weather_for_city(city: str) -> dict:
|
||||
"""Provides weather information for a specific city."""
|
||||
print(f"Server: Generating weather for city: {city}...")
|
||||
# In reality, call a weather API using the 'city' parameter
|
||||
temp = 20 + len(city) % 5 # Dummy logic
|
||||
condition = "Sunny" if len(city) % 2 == 0 else "Cloudy"
|
||||
return {"city": city.capitalize(), "temperature": temp, "unit": "celsius", "condition": condition}
|
||||
|
||||
# Template with an integer parameter
|
||||
@mcp.resource("users://{user_id}/profile")
|
||||
async def get_user_profile(user_id: int) -> dict:
|
||||
"""Retrieves a user's profile information by ID."""
|
||||
print(f"Server: Generating profile for user ID: {user_id}...")
|
||||
# In reality, fetch from database using user_id
|
||||
# FastMCP uses Pydantic to auto-convert the string URI part to int
|
||||
if user_id == 1:
|
||||
return {"id": user_id, "name": "Alice", "email": "alice@example.com", "status": "active"}
|
||||
elif user_id == 2:
|
||||
return {"id": user_id, "name": "Bob", "email": "bob@example.com", "status": "inactive"}
|
||||
else:
|
||||
# Example of returning an error structure
|
||||
return {"error": f"User with ID {user_id} not found"}
|
||||
```
|
||||
|
||||
**How Templates Work:**
|
||||
|
||||
1. **Definition:** When FastMCP sees `{...}` placeholders in the `@resource` URI and matching function parameters, it registers a `ResourceTemplate`.
|
||||
2. **Discovery:** Clients list templates via `resources/listResourceTemplates`.
|
||||
3. **Request & Matching:** A client requests a specific URI, e.g., `data://weather/london`. FastMCP matches this to the `data://weather/{city}` template.
|
||||
4. **Parameter Extraction:** It extracts the parameter value: `city="london"`.
|
||||
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)`.
|
||||
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`).
|
||||
|
||||
Templates provide a powerful way to expose parameterized data access points following REST-like principles.
|
||||
|
||||
## Server Behavior: Handling Duplicate Resources
|
||||
|
||||
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.
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
|
||||
mcp = FastMCP(
|
||||
name="ResourceServer",
|
||||
on_duplicate_resources=DuplicateBehavior.ERROR # Raise error on duplicates
|
||||
)
|
||||
|
||||
@mcp.resource("data://config")
|
||||
def get_config_v1(): return {"version": 1}
|
||||
|
||||
# This registration attempt will raise a ValueError because
|
||||
# "data://config" is already registered and the behavior is ERROR.
|
||||
# @mcp.resource("data://config")
|
||||
# def get_config_v2(): return {"version": 2}
|
||||
```
|
||||
|
||||
The `DuplicateBehavior` enum options are:
|
||||
|
||||
- `WARN` (default): Logs a warning, and the new resource/template replaces the old one.
|
||||
- `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
|
||||
- `REPLACE`: Silently replaces the existing resource/template with the new one.
|
||||
- `IGNORE`: Keeps the original resource/template and ignores the new registration attempt.
|
||||
|
|
@ -310,26 +310,25 @@ You can control how the FastMCP server behaves if you try to register multiple t
|
|||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
|
||||
mcp = FastMCP(
|
||||
name="StrictServer",
|
||||
# Configure behavior for duplicate tool names
|
||||
on_duplicate_tools=DuplicateBehavior.ERROR
|
||||
on_duplicate_tools="error"
|
||||
)
|
||||
|
||||
@mcp.tool()
|
||||
def my_tool(): return "Version 1"
|
||||
|
||||
# This will now raise a ValueError because 'my_tool' already exists
|
||||
# and on_duplicate_tools is set to ERROR.
|
||||
# and on_duplicate_tools is set to "error".
|
||||
# @mcp.tool()
|
||||
# def my_tool(): return "Version 2"
|
||||
```
|
||||
|
||||
The `DuplicateBehavior` enum options are:
|
||||
The duplicate behavior options are:
|
||||
|
||||
- `WARN` (default): Logs a warning and the new tool replaces the old one.
|
||||
- `ERROR`: Raises a `ValueError`, preventing the duplicate registration.
|
||||
- `REPLACE`: Silently replaces the existing tool with the new one.
|
||||
- `IGNORE`: Keeps the original tool and ignores the new registration attempt.
|
||||
- `"warn"` (default): Logs a warning and the new tool replaces the old one.
|
||||
- `"error"`: Raises a `ValueError`, preventing the duplicate registration.
|
||||
- `"replace"`: Silently replaces the existing tool with the new one.
|
||||
- `"ignore"`: Keeps the original tool and ignores the new registration attempt.
|
||||
|
|
@ -15,8 +15,19 @@ logger = get_logger(__name__)
|
|||
class PromptManager:
|
||||
"""Manages FastMCP prompts."""
|
||||
|
||||
def __init__(self, duplicate_behavior: DuplicateBehavior = DuplicateBehavior.WARN):
|
||||
def __init__(self, duplicate_behavior: DuplicateBehavior | None = None):
|
||||
self._prompts: dict[str, Prompt] = {}
|
||||
|
||||
# Default to "warn" if None is provided
|
||||
if duplicate_behavior is None:
|
||||
duplicate_behavior = "warn"
|
||||
|
||||
if duplicate_behavior not in DuplicateBehavior.__args__:
|
||||
raise ValueError(
|
||||
f"Invalid duplicate_behavior: {duplicate_behavior}. "
|
||||
f"Must be one of: {', '.join(DuplicateBehavior.__args__)}"
|
||||
)
|
||||
|
||||
self.duplicate_behavior = duplicate_behavior
|
||||
|
||||
def get_prompt(self, name: str) -> Prompt | None:
|
||||
|
|
@ -44,17 +55,17 @@ class PromptManager:
|
|||
# Check for duplicates
|
||||
existing = self._prompts.get(prompt.name)
|
||||
if existing:
|
||||
if self.duplicate_behavior == DuplicateBehavior.WARN:
|
||||
if self.duplicate_behavior == "warn":
|
||||
logger.warning(f"Prompt already exists: {prompt.name}")
|
||||
self._prompts[prompt.name] = prompt
|
||||
elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
|
||||
elif self.duplicate_behavior == "replace":
|
||||
self._prompts[prompt.name] = prompt
|
||||
elif self.duplicate_behavior == DuplicateBehavior.ERROR:
|
||||
elif self.duplicate_behavior == "error":
|
||||
raise ValueError(f"Prompt already exists: {prompt.name}")
|
||||
elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
|
||||
pass
|
||||
|
||||
self._prompts[prompt.name] = prompt
|
||||
elif self.duplicate_behavior == "ignore":
|
||||
return existing
|
||||
else:
|
||||
self._prompts[prompt.name] = prompt
|
||||
return prompt
|
||||
|
||||
async def render_prompt(
|
||||
|
|
|
|||
|
|
@ -19,9 +19,20 @@ logger = get_logger(__name__)
|
|||
class ResourceManager:
|
||||
"""Manages FastMCP resources."""
|
||||
|
||||
def __init__(self, duplicate_behavior: DuplicateBehavior = DuplicateBehavior.WARN):
|
||||
def __init__(self, duplicate_behavior: DuplicateBehavior | None = None):
|
||||
self._resources: dict[str, Resource] = {}
|
||||
self._templates: dict[str, ResourceTemplate] = {}
|
||||
|
||||
# Default to "warn" if None is provided
|
||||
if duplicate_behavior is None:
|
||||
duplicate_behavior = "warn"
|
||||
|
||||
if duplicate_behavior not in DuplicateBehavior.__args__:
|
||||
raise ValueError(
|
||||
f"Invalid duplicate_behavior: {duplicate_behavior}. "
|
||||
f"Must be one of: {', '.join(DuplicateBehavior.__args__)}"
|
||||
)
|
||||
|
||||
self.duplicate_behavior = duplicate_behavior
|
||||
|
||||
def add_resource_or_template_from_fn(
|
||||
|
|
@ -104,26 +115,28 @@ class ResourceManager:
|
|||
Args:
|
||||
resource: A Resource instance to add
|
||||
"""
|
||||
uri_str = str(resource.uri)
|
||||
logger.debug(
|
||||
"Adding resource",
|
||||
extra={
|
||||
"uri": resource.uri,
|
||||
"uri": uri_str,
|
||||
"type": type(resource).__name__,
|
||||
"resource_name": resource.name,
|
||||
},
|
||||
)
|
||||
existing = self._resources.get(str(resource.uri))
|
||||
existing = self._resources.get(uri_str)
|
||||
if existing:
|
||||
if self.duplicate_behavior == DuplicateBehavior.WARN:
|
||||
logger.warning(f"Resource already exists: {resource.uri}")
|
||||
self._resources[str(resource.uri)] = resource
|
||||
elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
|
||||
self._resources[str(resource.uri)] = resource
|
||||
elif self.duplicate_behavior == DuplicateBehavior.ERROR:
|
||||
raise ValueError(f"Resource already exists: {resource.uri}")
|
||||
elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
|
||||
pass
|
||||
self._resources[str(resource.uri)] = resource
|
||||
if self.duplicate_behavior == "warn":
|
||||
logger.warning(f"Resource already exists: {uri_str}")
|
||||
self._resources[uri_str] = resource
|
||||
elif self.duplicate_behavior == "replace":
|
||||
self._resources[uri_str] = resource
|
||||
elif self.duplicate_behavior == "error":
|
||||
raise ValueError(f"Resource already exists: {uri_str}")
|
||||
elif self.duplicate_behavior == "ignore":
|
||||
return existing
|
||||
else:
|
||||
self._resources[uri_str] = resource
|
||||
return resource
|
||||
|
||||
def add_template_from_fn(
|
||||
|
|
@ -157,26 +170,28 @@ class ResourceManager:
|
|||
The added template. If a template with the same URI already exists,
|
||||
returns the existing template.
|
||||
"""
|
||||
uri_template_str = str(template.uri_template)
|
||||
logger.debug(
|
||||
"Adding resource",
|
||||
extra={
|
||||
"uri": template.uri_template,
|
||||
"uri": uri_template_str,
|
||||
"type": type(template).__name__,
|
||||
"resource_name": template.name,
|
||||
},
|
||||
)
|
||||
existing = self._templates.get(str(template.uri_template))
|
||||
existing = self._templates.get(uri_template_str)
|
||||
if existing:
|
||||
if self.duplicate_behavior == DuplicateBehavior.WARN:
|
||||
logger.warning(f"Resource already exists: {template.uri_template}")
|
||||
self._templates[str(template.uri_template)] = template
|
||||
elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
|
||||
self._templates[str(template.uri_template)] = template
|
||||
elif self.duplicate_behavior == DuplicateBehavior.ERROR:
|
||||
raise ValueError(f"Resource already exists: {template.uri_template}")
|
||||
elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
|
||||
pass
|
||||
self._templates[template.uri_template] = template
|
||||
if self.duplicate_behavior == "warn":
|
||||
logger.warning(f"Resource already exists: {uri_template_str}")
|
||||
self._templates[uri_template_str] = template
|
||||
elif self.duplicate_behavior == "replace":
|
||||
self._templates[uri_template_str] = template
|
||||
elif self.duplicate_behavior == "error":
|
||||
raise ValueError(f"Resource already exists: {uri_template_str}")
|
||||
elif self.duplicate_behavior == "ignore":
|
||||
return existing
|
||||
else:
|
||||
self._templates[uri_template_str] = template
|
||||
return template
|
||||
|
||||
async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
from __future__ import annotations as _annotations
|
||||
|
||||
from enum import Enum
|
||||
from typing import TYPE_CHECKING, Literal
|
||||
|
||||
from pydantic import Field
|
||||
|
|
@ -11,12 +10,7 @@ if TYPE_CHECKING:
|
|||
|
||||
LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
||||
|
||||
|
||||
class DuplicateBehavior(Enum):
|
||||
WARN = "warn"
|
||||
ERROR = "error"
|
||||
REPLACE = "replace"
|
||||
IGNORE = "ignore"
|
||||
DuplicateBehavior = Literal["warn", "error", "replace", "ignore"]
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
|
|
@ -55,13 +49,13 @@ class ServerSettings(BaseSettings):
|
|||
debug: bool = False
|
||||
|
||||
# resource settings
|
||||
on_duplicate_resources: DuplicateBehavior = DuplicateBehavior.WARN
|
||||
on_duplicate_resources: DuplicateBehavior = "warn"
|
||||
|
||||
# tool settings
|
||||
on_duplicate_tools: DuplicateBehavior = DuplicateBehavior.WARN
|
||||
on_duplicate_tools: DuplicateBehavior = "warn"
|
||||
|
||||
# prompt settings
|
||||
on_duplicate_prompts: DuplicateBehavior = DuplicateBehavior.WARN
|
||||
on_duplicate_prompts: DuplicateBehavior = "warn"
|
||||
|
||||
dependencies: list[str] = Field(
|
||||
default_factory=list,
|
||||
|
|
|
|||
|
|
@ -21,8 +21,19 @@ logger = get_logger(__name__)
|
|||
class ToolManager:
|
||||
"""Manages FastMCP tools."""
|
||||
|
||||
def __init__(self, duplicate_behavior: DuplicateBehavior = DuplicateBehavior.WARN):
|
||||
def __init__(self, duplicate_behavior: DuplicateBehavior | None = None):
|
||||
self._tools: dict[str, Tool] = {}
|
||||
|
||||
# Default to "warn" if None is provided
|
||||
if duplicate_behavior is None:
|
||||
duplicate_behavior = "warn"
|
||||
|
||||
if duplicate_behavior not in DuplicateBehavior.__args__:
|
||||
raise ValueError(
|
||||
f"Invalid duplicate_behavior: {duplicate_behavior}. "
|
||||
f"Must be one of: {', '.join(DuplicateBehavior.__args__)}"
|
||||
)
|
||||
|
||||
self.duplicate_behavior = duplicate_behavior
|
||||
|
||||
def get_tool(self, name: str) -> Tool | None:
|
||||
|
|
@ -57,16 +68,17 @@ class ToolManager:
|
|||
name = name or tool.name
|
||||
existing = self._tools.get(name)
|
||||
if existing:
|
||||
if self.duplicate_behavior == DuplicateBehavior.WARN:
|
||||
if self.duplicate_behavior == "warn":
|
||||
logger.warning(f"Tool already exists: {name}")
|
||||
self._tools[name] = tool
|
||||
elif self.duplicate_behavior == DuplicateBehavior.REPLACE:
|
||||
elif self.duplicate_behavior == "replace":
|
||||
self._tools[name] = tool
|
||||
elif self.duplicate_behavior == DuplicateBehavior.ERROR:
|
||||
elif self.duplicate_behavior == "error":
|
||||
raise ValueError(f"Tool already exists: {name}")
|
||||
elif self.duplicate_behavior == DuplicateBehavior.IGNORE:
|
||||
pass
|
||||
self._tools[name] = tool
|
||||
elif self.duplicate_behavior == "ignore":
|
||||
return existing
|
||||
else:
|
||||
self._tools[name] = tool
|
||||
return tool
|
||||
|
||||
async def call_tool(
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ from fastmcp.exceptions import PromptError
|
|||
from fastmcp.prompts import Prompt
|
||||
from fastmcp.prompts.prompt import PromptArgument, TextContent, UserMessage
|
||||
from fastmcp.prompts.prompt_manager import PromptManager
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
|
||||
|
||||
class TestPromptManager:
|
||||
|
|
@ -26,7 +25,7 @@ class TestPromptManager:
|
|||
def fn() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
manager = PromptManager(duplicate_behavior=DuplicateBehavior.WARN)
|
||||
manager = PromptManager(duplicate_behavior="warn")
|
||||
prompt = Prompt.from_function(fn)
|
||||
first = manager.add_prompt(prompt)
|
||||
second = manager.add_prompt(prompt)
|
||||
|
|
@ -39,13 +38,87 @@ class TestPromptManager:
|
|||
def fn() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
manager = PromptManager(duplicate_behavior=DuplicateBehavior.IGNORE)
|
||||
manager = PromptManager(duplicate_behavior="ignore")
|
||||
prompt = Prompt.from_function(fn)
|
||||
first = manager.add_prompt(prompt)
|
||||
second = manager.add_prompt(prompt)
|
||||
assert first == second
|
||||
assert "Prompt already exists" not in caplog.text
|
||||
|
||||
def test_warn_on_duplicate_prompts(self, caplog):
|
||||
"""Test warning on duplicate prompts."""
|
||||
manager = PromptManager(duplicate_behavior="warn")
|
||||
|
||||
def test_fn() -> str:
|
||||
return "Test prompt"
|
||||
|
||||
prompt = Prompt.from_function(test_fn, name="test_prompt")
|
||||
|
||||
manager.add_prompt(prompt)
|
||||
manager.add_prompt(prompt)
|
||||
|
||||
assert "Prompt already exists: test_prompt" in caplog.text
|
||||
# Should have the prompt
|
||||
assert manager.get_prompt("test_prompt") is not None
|
||||
|
||||
def test_error_on_duplicate_prompts(self):
|
||||
"""Test error on duplicate prompts."""
|
||||
manager = PromptManager(duplicate_behavior="error")
|
||||
|
||||
def test_fn() -> str:
|
||||
return "Test prompt"
|
||||
|
||||
prompt = Prompt.from_function(test_fn, name="test_prompt")
|
||||
|
||||
manager.add_prompt(prompt)
|
||||
|
||||
with pytest.raises(ValueError, match="Prompt already exists: test_prompt"):
|
||||
manager.add_prompt(prompt)
|
||||
|
||||
def test_replace_duplicate_prompts(self):
|
||||
"""Test replacing duplicate prompts."""
|
||||
manager = PromptManager(duplicate_behavior="replace")
|
||||
|
||||
def original_fn() -> str:
|
||||
return "Original prompt"
|
||||
|
||||
def replacement_fn() -> str:
|
||||
return "Replacement prompt"
|
||||
|
||||
prompt1 = Prompt.from_function(original_fn, name="test_prompt")
|
||||
prompt2 = Prompt.from_function(replacement_fn, name="test_prompt")
|
||||
|
||||
manager.add_prompt(prompt1)
|
||||
manager.add_prompt(prompt2)
|
||||
|
||||
# Should have replaced with the new prompt
|
||||
prompt = manager.get_prompt("test_prompt")
|
||||
assert prompt is not None
|
||||
assert prompt.fn.__name__ == "replacement_fn"
|
||||
|
||||
def test_ignore_duplicate_prompts(self):
|
||||
"""Test ignoring duplicate prompts."""
|
||||
manager = PromptManager(duplicate_behavior="ignore")
|
||||
|
||||
def original_fn() -> str:
|
||||
return "Original prompt"
|
||||
|
||||
def replacement_fn() -> str:
|
||||
return "Replacement prompt"
|
||||
|
||||
prompt1 = Prompt.from_function(original_fn, name="test_prompt")
|
||||
prompt2 = Prompt.from_function(replacement_fn, name="test_prompt")
|
||||
|
||||
manager.add_prompt(prompt1)
|
||||
result = manager.add_prompt(prompt2)
|
||||
|
||||
# Should keep the original
|
||||
prompt = manager.get_prompt("test_prompt")
|
||||
assert prompt is not None
|
||||
assert prompt.fn.__name__ == "original_fn"
|
||||
# Result should be the original prompt
|
||||
assert result.fn.__name__ == "original_fn"
|
||||
|
||||
def test_list_prompts(self):
|
||||
"""Test listing all prompts."""
|
||||
|
||||
|
|
@ -114,39 +187,6 @@ class TestPromptManager:
|
|||
with pytest.raises(ValueError, match="Missing required arguments"):
|
||||
await manager.render_prompt("fn")
|
||||
|
||||
def test_error_on_duplicate_prompts(self):
|
||||
"""Test error on duplicate prompts."""
|
||||
|
||||
def fn() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
manager = PromptManager(duplicate_behavior=DuplicateBehavior.ERROR)
|
||||
prompt = Prompt.from_function(fn)
|
||||
manager.add_prompt(prompt)
|
||||
|
||||
with pytest.raises(ValueError, match="Prompt already exists"):
|
||||
manager.add_prompt(prompt)
|
||||
|
||||
def test_replace_duplicate_prompts(self):
|
||||
"""Test replacing duplicate prompts."""
|
||||
|
||||
def fn1() -> str:
|
||||
return "Original"
|
||||
|
||||
def fn2() -> str:
|
||||
return "Replacement"
|
||||
|
||||
manager = PromptManager(duplicate_behavior=DuplicateBehavior.REPLACE)
|
||||
prompt1 = Prompt.from_function(fn1, name="test_prompt")
|
||||
prompt2 = Prompt.from_function(fn2, name="test_prompt")
|
||||
|
||||
manager.add_prompt(prompt1)
|
||||
manager.add_prompt(prompt2)
|
||||
|
||||
# Should have replaced the first prompt with the second
|
||||
stored_prompt = manager.get_prompt("test_prompt")
|
||||
assert stored_prompt == prompt2
|
||||
|
||||
|
||||
class TestPromptTags:
|
||||
"""Test functionality related to prompt tags."""
|
||||
|
|
|
|||
|
|
@ -11,7 +11,6 @@ from fastmcp.resources import (
|
|||
ResourceManager,
|
||||
ResourceTemplate,
|
||||
)
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -61,19 +60,24 @@ class TestResourceManager:
|
|||
|
||||
def test_warn_on_duplicate_resources(self, temp_file: Path, caplog):
|
||||
"""Test warning on duplicate resources."""
|
||||
manager = ResourceManager(duplicate_behavior=DuplicateBehavior.WARN)
|
||||
manager = ResourceManager(duplicate_behavior="warn")
|
||||
|
||||
resource = FileResource(
|
||||
uri=FileUrl(f"file://{temp_file}"),
|
||||
name="test",
|
||||
name="test_resource",
|
||||
path=temp_file,
|
||||
)
|
||||
|
||||
manager.add_resource(resource)
|
||||
manager.add_resource(resource)
|
||||
|
||||
assert "Resource already exists" in caplog.text
|
||||
# Should have the resource
|
||||
assert len(manager.list_resources()) == 1
|
||||
|
||||
def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog):
|
||||
"""Test disabling warning on duplicate resources."""
|
||||
manager = ResourceManager(duplicate_behavior=DuplicateBehavior.IGNORE)
|
||||
manager = ResourceManager(duplicate_behavior="ignore")
|
||||
resource = FileResource(
|
||||
uri=FileUrl(f"file://{temp_file}"),
|
||||
name="test",
|
||||
|
|
@ -85,12 +89,14 @@ class TestResourceManager:
|
|||
|
||||
def test_error_on_duplicate_resources(self, temp_file: Path):
|
||||
"""Test error on duplicate resources."""
|
||||
manager = ResourceManager(duplicate_behavior=DuplicateBehavior.ERROR)
|
||||
manager = ResourceManager(duplicate_behavior="error")
|
||||
|
||||
resource = FileResource(
|
||||
uri=FileUrl(f"file://{temp_file}"),
|
||||
name="test",
|
||||
name="test_resource",
|
||||
path=temp_file,
|
||||
)
|
||||
|
||||
manager.add_resource(resource)
|
||||
|
||||
with pytest.raises(ValueError, match="Resource already exists"):
|
||||
|
|
@ -98,27 +104,153 @@ class TestResourceManager:
|
|||
|
||||
def test_replace_duplicate_resources(self, temp_file: Path):
|
||||
"""Test replacing duplicate resources."""
|
||||
manager = ResourceManager(duplicate_behavior=DuplicateBehavior.REPLACE)
|
||||
manager = ResourceManager(duplicate_behavior="replace")
|
||||
|
||||
resource1 = FileResource(
|
||||
uri=FileUrl(f"file://{temp_file}"),
|
||||
name="test1",
|
||||
name="original",
|
||||
path=temp_file,
|
||||
)
|
||||
|
||||
resource2 = FileResource(
|
||||
uri=FileUrl(f"file://{temp_file}"),
|
||||
name="test2", # Different name
|
||||
name="replacement",
|
||||
path=temp_file,
|
||||
)
|
||||
|
||||
manager.add_resource(resource1)
|
||||
manager.add_resource(resource2)
|
||||
|
||||
# Should have replaced the first resource with the second
|
||||
# Should have replaced with the new resource
|
||||
resources = manager.list_resources()
|
||||
assert len(resources) == 1
|
||||
assert resources[0].name == "test2"
|
||||
assert resources[0].name == "replacement"
|
||||
|
||||
def test_ignore_duplicate_resources(self, temp_file: Path):
|
||||
"""Test ignoring duplicate resources."""
|
||||
manager = ResourceManager(duplicate_behavior="ignore")
|
||||
|
||||
resource1 = FileResource(
|
||||
uri=FileUrl(f"file://{temp_file}"),
|
||||
name="original",
|
||||
path=temp_file,
|
||||
)
|
||||
|
||||
resource2 = FileResource(
|
||||
uri=FileUrl(f"file://{temp_file}"),
|
||||
name="replacement",
|
||||
path=temp_file,
|
||||
)
|
||||
|
||||
manager.add_resource(resource1)
|
||||
result = manager.add_resource(resource2)
|
||||
|
||||
# Should keep the original
|
||||
resources = manager.list_resources()
|
||||
assert len(resources) == 1
|
||||
assert resources[0].name == "original"
|
||||
# Result should be the original resource
|
||||
assert result.name == "original"
|
||||
|
||||
def test_warn_on_duplicate_templates(self, caplog):
|
||||
"""Test warning on duplicate templates."""
|
||||
manager = ResourceManager(duplicate_behavior="warn")
|
||||
|
||||
def template_fn(id: str) -> str:
|
||||
return f"Template {id}"
|
||||
|
||||
template = ResourceTemplate.from_function(
|
||||
fn=template_fn,
|
||||
uri_template="test://{id}",
|
||||
name="test_template",
|
||||
)
|
||||
|
||||
manager.add_template(template)
|
||||
manager.add_template(template)
|
||||
|
||||
assert "Resource already exists" in caplog.text
|
||||
# Should have the template
|
||||
assert len(manager.list_templates()) == 1
|
||||
|
||||
def test_error_on_duplicate_templates(self):
|
||||
"""Test error on duplicate templates."""
|
||||
manager = ResourceManager(duplicate_behavior="error")
|
||||
|
||||
def template_fn(id: str) -> str:
|
||||
return f"Template {id}"
|
||||
|
||||
template = ResourceTemplate.from_function(
|
||||
fn=template_fn,
|
||||
uri_template="test://{id}",
|
||||
name="test_template",
|
||||
)
|
||||
|
||||
manager.add_template(template)
|
||||
|
||||
with pytest.raises(ValueError, match="Resource already exists"):
|
||||
manager.add_template(template)
|
||||
|
||||
def test_replace_duplicate_templates(self):
|
||||
"""Test replacing duplicate templates."""
|
||||
manager = ResourceManager(duplicate_behavior="replace")
|
||||
|
||||
def original_fn(id: str) -> str:
|
||||
return f"Original {id}"
|
||||
|
||||
def replacement_fn(id: str) -> str:
|
||||
return f"Replacement {id}"
|
||||
|
||||
template1 = ResourceTemplate.from_function(
|
||||
fn=original_fn,
|
||||
uri_template="test://{id}",
|
||||
name="original",
|
||||
)
|
||||
|
||||
template2 = ResourceTemplate.from_function(
|
||||
fn=replacement_fn,
|
||||
uri_template="test://{id}",
|
||||
name="replacement",
|
||||
)
|
||||
|
||||
manager.add_template(template1)
|
||||
manager.add_template(template2)
|
||||
|
||||
# Should have replaced with the new template
|
||||
templates = manager.list_templates()
|
||||
assert len(templates) == 1
|
||||
assert templates[0].name == "replacement"
|
||||
|
||||
def test_ignore_duplicate_templates(self):
|
||||
"""Test ignoring duplicate templates."""
|
||||
manager = ResourceManager(duplicate_behavior="ignore")
|
||||
|
||||
def original_fn(id: str) -> str:
|
||||
return f"Original {id}"
|
||||
|
||||
def replacement_fn(id: str) -> str:
|
||||
return f"Replacement {id}"
|
||||
|
||||
template1 = ResourceTemplate.from_function(
|
||||
fn=original_fn,
|
||||
uri_template="test://{id}",
|
||||
name="original",
|
||||
)
|
||||
|
||||
template2 = ResourceTemplate.from_function(
|
||||
fn=replacement_fn,
|
||||
uri_template="test://{id}",
|
||||
name="replacement",
|
||||
)
|
||||
|
||||
manager.add_template(template1)
|
||||
result = manager.add_template(template2)
|
||||
|
||||
# Should keep the original
|
||||
templates = manager.list_templates()
|
||||
assert len(templates) == 1
|
||||
assert templates[0].name == "original"
|
||||
# Result should be the original template
|
||||
assert result.name == "original"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_resource(self, temp_file: Path):
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ import pytest
|
|||
from pydantic import BaseModel
|
||||
|
||||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.settings import DuplicateBehavior
|
||||
from fastmcp.tools import ToolManager
|
||||
from fastmcp.tools.tool import Tool
|
||||
|
||||
|
|
@ -88,15 +87,17 @@ class TestAddTools:
|
|||
|
||||
def test_warn_on_duplicate_tools(self, caplog):
|
||||
"""Test warning on duplicate tools."""
|
||||
manager = ToolManager(duplicate_behavior="warn")
|
||||
|
||||
def f(x: int) -> int:
|
||||
def test_fn(x: int) -> int:
|
||||
return x
|
||||
|
||||
manager = ToolManager(duplicate_behavior=DuplicateBehavior.WARN)
|
||||
manager.add_tool_from_fn(f)
|
||||
with caplog.at_level(logging.WARNING):
|
||||
manager.add_tool_from_fn(f)
|
||||
assert "Tool already exists: f" in caplog.text
|
||||
manager.add_tool_from_fn(test_fn, name="test_tool")
|
||||
manager.add_tool_from_fn(test_fn, name="test_tool")
|
||||
|
||||
assert "Tool already exists: test_tool" in caplog.text
|
||||
# Should have the tool
|
||||
assert manager.get_tool("test_tool") is not None
|
||||
|
||||
def test_disable_warn_on_duplicate_tools(self, caplog):
|
||||
"""Test disabling warning on duplicate tools."""
|
||||
|
|
@ -104,7 +105,7 @@ class TestAddTools:
|
|||
def f(x: int) -> int:
|
||||
return x
|
||||
|
||||
manager = ToolManager(duplicate_behavior=DuplicateBehavior.IGNORE)
|
||||
manager = ToolManager(duplicate_behavior="ignore")
|
||||
manager.add_tool_from_fn(f)
|
||||
with caplog.at_level(logging.WARNING):
|
||||
manager.add_tool_from_fn(f)
|
||||
|
|
@ -112,18 +113,19 @@ class TestAddTools:
|
|||
|
||||
def test_error_on_duplicate_tools(self):
|
||||
"""Test error on duplicate tools."""
|
||||
manager = ToolManager(duplicate_behavior="error")
|
||||
|
||||
def f(x: int) -> int:
|
||||
def test_fn(x: int) -> int:
|
||||
return x
|
||||
|
||||
manager = ToolManager(duplicate_behavior=DuplicateBehavior.ERROR)
|
||||
manager.add_tool_from_fn(f)
|
||||
manager.add_tool_from_fn(test_fn, name="test_tool")
|
||||
|
||||
with pytest.raises(ValueError, match="Tool already exists"):
|
||||
manager.add_tool_from_fn(f)
|
||||
with pytest.raises(ValueError, match="Tool already exists: test_tool"):
|
||||
manager.add_tool_from_fn(test_fn, name="test_tool")
|
||||
|
||||
def test_replace_duplicate_tools(self):
|
||||
"""Test replacing duplicate tools."""
|
||||
manager = ToolManager(duplicate_behavior="replace")
|
||||
|
||||
def original_fn(x: int) -> int:
|
||||
return x
|
||||
|
|
@ -131,20 +133,33 @@ class TestAddTools:
|
|||
def replacement_fn(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
manager = ToolManager(duplicate_behavior=DuplicateBehavior.REPLACE)
|
||||
manager.add_tool_from_fn(original_fn, name="test_tool")
|
||||
replacement_tool = manager.add_tool_from_fn(replacement_fn, name="test_tool")
|
||||
manager.add_tool_from_fn(replacement_fn, name="test_tool")
|
||||
|
||||
# Should have replaced the first tool with the second
|
||||
stored_tool = manager.get_tool("test_tool")
|
||||
assert stored_tool is not None
|
||||
assert stored_tool == replacement_tool
|
||||
# Should have replaced with the new function
|
||||
tool = manager.get_tool("test_tool")
|
||||
assert tool is not None
|
||||
assert tool.fn.__name__ == "replacement_fn"
|
||||
|
||||
# The name should still be the same
|
||||
assert stored_tool.name == "test_tool"
|
||||
def test_ignore_duplicate_tools(self):
|
||||
"""Test ignoring duplicate tools."""
|
||||
manager = ToolManager(duplicate_behavior="ignore")
|
||||
|
||||
# But the function is different
|
||||
assert stored_tool.fn.__name__ == "replacement_fn"
|
||||
def original_fn(x: int) -> int:
|
||||
return x
|
||||
|
||||
def replacement_fn(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
manager.add_tool_from_fn(original_fn, name="test_tool")
|
||||
result = manager.add_tool_from_fn(replacement_fn, name="test_tool")
|
||||
|
||||
# Should keep the original
|
||||
tool = manager.get_tool("test_tool")
|
||||
assert tool is not None
|
||||
assert tool.fn.__name__ == "original_fn"
|
||||
# Result should be the original tool
|
||||
assert result.fn.__name__ == "original_fn"
|
||||
|
||||
|
||||
class TestToolTags:
|
||||
|
|
@ -630,7 +645,7 @@ class TestCustomToolNames:
|
|||
assert target_manager.get_tool("prefix/source_fn") is None
|
||||
|
||||
def test_replace_tool_keeps_original_name(self):
|
||||
"""Test that replacing a tool with DuplicateBehavior.REPLACE keeps the original name."""
|
||||
"""Test that replacing a tool with "replace" keeps the original name."""
|
||||
|
||||
def original_fn(x: int) -> int:
|
||||
return x
|
||||
|
|
@ -639,7 +654,7 @@ class TestCustomToolNames:
|
|||
return x * 2
|
||||
|
||||
# Create a manager with REPLACE behavior
|
||||
manager = ToolManager(duplicate_behavior=DuplicateBehavior.REPLACE)
|
||||
manager = ToolManager(duplicate_behavior="replace")
|
||||
|
||||
# Add the original tool
|
||||
original_tool = manager.add_tool_from_fn(original_fn, name="test_tool")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue