From 6e309e135413dccc41cdf657352d54f4d29bbe86 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 15 Apr 2025 09:33:23 -0400 Subject: [PATCH] Fix bug with duplicate behavior == ignore --- docs/servers/fastmcp.mdx | 4 +- docs/servers/prompts.mdx | 2 +- docs/servers/resources.mdx | 2 +- docs/servers/resources_backup.mdx | 2 +- docs/servers/tools.mdx | 2 +- src/fastmcp/prompts/prompt_manager.py | 27 ++-- src/fastmcp/resources/resource_manager.py | 39 ++++-- src/fastmcp/settings.py | 14 +- src/fastmcp/tools/tool_manager.py | 26 +++- tests/prompts/test_prompt_manager.py | 112 +++++++++++----- tests/resources/test_resource_manager.py | 154 ++++++++++++++++++++-- tests/tools/test_tool_manager.py | 65 +++++---- 12 files changed, 333 insertions(+), 116 deletions(-) diff --git a/docs/servers/fastmcp.mdx b/docs/servers/fastmcp.mdx index c86f22ed5..aec7f6231 100644 --- a/docs/servers/fastmcp.mdx +++ b/docs/servers/fastmcp.mdx @@ -314,12 +314,12 @@ from fastmcp.settings import DuplicateBehavior 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 diff --git a/docs/servers/prompts.mdx b/docs/servers/prompts.mdx index 6a25d8467..d1ffafe12 100644 --- a/docs/servers/prompts.mdx +++ b/docs/servers/prompts.mdx @@ -209,7 +209,7 @@ 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() diff --git a/docs/servers/resources.mdx b/docs/servers/resources.mdx index ec8d031d5..0cf9af926 100644 --- a/docs/servers/resources.mdx +++ b/docs/servers/resources.mdx @@ -301,7 +301,7 @@ 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") diff --git a/docs/servers/resources_backup.mdx b/docs/servers/resources_backup.mdx index 82bf58254..59db96787 100644 --- a/docs/servers/resources_backup.mdx +++ b/docs/servers/resources_backup.mdx @@ -250,7 +250,7 @@ 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") diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx index ec8633083..e04756574 100644 --- a/docs/servers/tools.mdx +++ b/docs/servers/tools.mdx @@ -315,7 +315,7 @@ 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() diff --git a/src/fastmcp/prompts/prompt_manager.py b/src/fastmcp/prompts/prompt_manager.py index 40251e1ca..30f971589 100644 --- a/src/fastmcp/prompts/prompt_manager.py +++ b/src/fastmcp/prompts/prompt_manager.py @@ -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( diff --git a/src/fastmcp/resources/resource_manager.py b/src/fastmcp/resources/resource_manager.py index d33aa6f54..ae79c12db 100644 --- a/src/fastmcp/resources/resource_manager.py +++ b/src/fastmcp/resources/resource_manager.py @@ -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( @@ -114,16 +125,17 @@ class ResourceManager: ) existing = self._resources.get(str(resource.uri)) if existing: - if self.duplicate_behavior == DuplicateBehavior.WARN: + if self.duplicate_behavior == "warn": logger.warning(f"Resource already exists: {resource.uri}") self._resources[str(resource.uri)] = resource - elif self.duplicate_behavior == DuplicateBehavior.REPLACE: + elif self.duplicate_behavior == "replace": self._resources[str(resource.uri)] = resource - elif self.duplicate_behavior == DuplicateBehavior.ERROR: + elif self.duplicate_behavior == "error": raise ValueError(f"Resource already exists: {resource.uri}") - elif self.duplicate_behavior == DuplicateBehavior.IGNORE: - pass - self._resources[str(resource.uri)] = resource + elif self.duplicate_behavior == "ignore": + return existing + else: + self._resources[str(resource.uri)] = resource return resource def add_template_from_fn( @@ -167,16 +179,17 @@ class ResourceManager: ) existing = self._templates.get(str(template.uri_template)) if existing: - if self.duplicate_behavior == DuplicateBehavior.WARN: + if self.duplicate_behavior == "warn": logger.warning(f"Resource already exists: {template.uri_template}") self._templates[str(template.uri_template)] = template - elif self.duplicate_behavior == DuplicateBehavior.REPLACE: + elif self.duplicate_behavior == "replace": self._templates[str(template.uri_template)] = template - elif self.duplicate_behavior == DuplicateBehavior.ERROR: + elif self.duplicate_behavior == "error": raise ValueError(f"Resource already exists: {template.uri_template}") - elif self.duplicate_behavior == DuplicateBehavior.IGNORE: - pass - self._templates[template.uri_template] = template + elif self.duplicate_behavior == "ignore": + return existing + else: + self._templates[template.uri_template] = template return template async def get_resource(self, uri: AnyUrl | str) -> Resource | None: diff --git a/src/fastmcp/settings.py b/src/fastmcp/settings.py index 09f7e4fdc..41b3fcb89 100644 --- a/src/fastmcp/settings.py +++ b/src/fastmcp/settings.py @@ -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, diff --git a/src/fastmcp/tools/tool_manager.py b/src/fastmcp/tools/tool_manager.py index 6e2c1bcde..764804058 100644 --- a/src/fastmcp/tools/tool_manager.py +++ b/src/fastmcp/tools/tool_manager.py @@ -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( diff --git a/tests/prompts/test_prompt_manager.py b/tests/prompts/test_prompt_manager.py index b96994a0c..e53148987 100644 --- a/tests/prompts/test_prompt_manager.py +++ b/tests/prompts/test_prompt_manager.py @@ -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.""" diff --git a/tests/resources/test_resource_manager.py b/tests/resources/test_resource_manager.py index 1ee31a62c..284ac1055 100644 --- a/tests/resources/test_resource_manager.py +++ b/tests/resources/test_resource_manager.py @@ -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): diff --git a/tests/tools/test_tool_manager.py b/tests/tools/test_tool_manager.py index 9a83860f9..c86e8b0fb 100644 --- a/tests/tools/test_tool_manager.py +++ b/tests/tools/test_tool_manager.py @@ -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")