fix: drop configurable dedupe from AggregateProvider, always warn (#3877)

CLAUDE.md

@@ -50,6 +50,8 @@ When modifying MCP functionality, changes typically need to be applied across al
 - **Resource Templates** (`src/resources/`)
 - **Prompts** (`src/prompts/`)
 
+**Before writing cross-component logic (dedupe, grouping, lookups, identity checks), read `FastMCPComponent` in `src/fastmcp/utilities/components.py`.** The base class defines the shared surface — `name`, `version`, `tags`, `meta`, and critically the `key` property which is the canonical MCP identity (encodes type, identifier, and version). Prefer `item.key` over ad-hoc `name or uri or uri_template` fallbacks; overrides in `Resource` and `ResourceTemplate` already handle URI-based identity, and `.key` includes the version suffix so variants of the same component don't falsely collide.
+
 ## Development Rules
 
 **Read `CONTRIBUTING.md` before opening issues or PRs.** It describes when PRs are appropriate, what we expect from enhancement proposals, and what we'll close without review.

src/fastmcp/server/providers/aggregate.py

@@ -74,7 +74,6 @@ class AggregateProvider(Provider):
         """
         super().__init__()
         self.providers: list[Provider] = list(providers or [])
-        self._on_duplicate: str = "warn"
 
     def add_provider(self, provider: Provider, *, namespace: str = "") -> None:
         """Add a provider with optional namespace.
@@ -109,12 +108,17 @@ class AggregateProvider(Provider):
     ) -> list[T]:
         """Collect successful list results, logging any exceptions.
 
-        Detects duplicate component names across providers and applies
-        the on_duplicate behavior (error/warn/replace/ignore).
+        Emits a warning when the same MCP identity is returned by more than
+        one provider — surfaces composition mistakes to the server author.
+        This is always a warning: cross-provider collisions happen at runtime
+        (sometimes dynamically), so an errorable/strict mode would give the
+        author no way to react and would crash list calls in production.
         """
         collected: list[T] = []
-        # Track (name, provider_index) to allow version variants from same provider
-        seen_names: dict[str, int] = {}  # name -> provider index of first occurrence
+        # FastMCPComponent.key encodes type, identifier, and version —
+        # so version variants of the same component are NOT reported as
+        # collisions (matching _get_highest_version_result behavior).
+        seen_keys: dict[str, int] = {}
         for i, result in enumerate(results):
             if isinstance(result, BaseException):
                 logger.warning(
@@ -123,27 +127,15 @@ class AggregateProvider(Provider):
                 )
                 continue
             for item in result:
-                # Extract identity key: name, uri, or uri_template
-                name = (
-                    getattr(item, "name", None)
-                    or getattr(item, "uri", None)
-                    or getattr(item, "uri_template", None)
-                )
-                if name is not None:
-                    name = str(name)
-                    if name in seen_names and seen_names[name] != i:
-                        # Duplicate from a DIFFERENT provider — flag it
-                        msg = (
-                            f"Duplicate {operation} component '{name}' "
+                key = getattr(item, "key", None)
+                if key is not None:
+                    first = seen_keys.setdefault(key, i)
+                    if first != i:
+                        logger.warning(
+                            f"Duplicate {operation} component {key!r} "
                             f"from provider {self.providers[i]} "
-                            f"(first seen from provider "
-                            f"{self.providers[seen_names[name]]})"
+                            f"(first seen from provider {self.providers[first]})"
                         )
-                        if self._on_duplicate == "error":
-                            raise ValueError(msg)
-                        elif self._on_duplicate == "warn":
-                            logger.warning(msg)
-                    seen_names.setdefault(name, i)
                 collected.append(item)
         return collected
 

src/fastmcp/utilities/components.py

@@ -147,6 +147,13 @@ class FastMCPComponent(FastMCPBaseModel):
 
         Subclasses should override this to use their specific identifier.
         Base implementation uses name.
+
+        Prefer `.key` over ad-hoc `name or uri or uri_template` logic for any
+        cross-component identity work (dedupe, grouping, collision detection,
+        lookup tables). It encodes type, identifier, and version, so variants
+        of the same component don't falsely collide with each other, and
+        cross-type identifiers (e.g. a tool and a resource both named "foo")
+        can't clash.
         """
         base_key = self.make_key(self.name)
         return f"{base_key}@{self.version or ''}"

tests/server/mount/test_mount.py

@@ -544,31 +544,17 @@ class TestPrefixConflictResolution:
 
 
 class TestCrossProviderDuplicateDetection:
-    """Test that on_duplicate catches duplicates across mounted providers."""
+    """Cross-provider collisions always log a warning — diagnostic signal only.
 
-    async def test_on_duplicate_error_raises_for_same_namespace(self):
-        """Mounting two servers with the same namespace and tool name raises."""
-        main = FastMCP("Main", on_duplicate="error")
-        sub1 = FastMCP("Sub1")
-        sub2 = FastMCP("Sub2")
+    `on_duplicate` is a registration-time setting for LocalProvider (two
+    decorators on the same server), not a knob for AggregateProvider
+    composition. Mounted-provider collisions happen at runtime (sometimes
+    dynamically), so an error mode would give the author no way to react.
+    """
 
-        @sub1.tool(name="greet")
-        def greet_v1() -> str:
-            return "from sub1"
-
-        @sub2.tool(name="greet")
-        def greet_v2() -> str:
-            return "from sub2"
-
-        main.mount(sub1, "ns")
-        main.mount(sub2, "ns")
-
-        with pytest.raises(ValueError, match="Duplicate"):
-            await main.list_tools()
-
-    async def test_on_duplicate_warn_logs_for_same_namespace(self, caplog):
-        """Mounting with on_duplicate='warn' logs a warning instead of raising."""
-        main = FastMCP("Main", on_duplicate="warn")
+    async def test_cross_provider_duplicate_warns(self, caplog):
+        """Two mounted providers exposing the same tool identity log a warning."""
+        main = FastMCP("Main")
         sub1 = FastMCP("Sub1")
         sub2 = FastMCP("Sub2")
 
@@ -589,8 +575,8 @@ class TestCrossProviderDuplicateDetection:
         assert any("Duplicate" in r.message for r in caplog.records)
 
     async def test_no_false_positive_for_different_names(self):
-        """Different tool names in the same namespace don't trigger duplicate."""
-        main = FastMCP("Main", on_duplicate="error")
+        """Different tool names in the same namespace don't trigger a warning."""
+        main = FastMCP("Main")
         sub1 = FastMCP("Sub1")
         sub2 = FastMCP("Sub2")