fix(mcp): keep one precedence when both virtual tools share a name

`instrument()` already handled a host that configures
`missing_capability_tool_name` and `collect_feedback.tool_name` to the
same string: it drops the feedback tool, keeps missing-capability
precedence, and warns "duplicate". The `PostHogMCP` dispatcher did
neither.

`prepare_tool_list` fell through to its "name already taken" branch, so
the feedback descriptor went unadvertised with no warning.
`prepare_tool_call` then set `is_feedback` and `is_missing_capability`
both true, and the README's dispatch snippet tests `is_feedback` first,
so every missing-capability call became a bogus $mcp_feedback capture --
the inverse of the precedence the other path sets on purpose.

Also folds four copies of the ownership-probe warning into one helper,
and names the three collision variants, so a typo is a type error rather
than silently picking the "blocked" wording.

Generated-By: PostHog Desktop
Task-Id: 7047b34a-b562-40ab-ae65-8b0f24700e57

Author: gesh

posthog/mcp/_instrument_fastmcp.py

@@ -40,6 +40,7 @@
     resolve_session_and_client,
     start_tool_call_lifecycle,
     start_tools_list_lifecycle,
+    warn_ownership_lookup_failed,
 )
 from ._internal import MCPAnalyticsData
 from ._model_parameters import (
@@ -326,10 +327,7 @@ def _name_owned_by_real_tool(server: Any, name: str) -> Optional[bool]:
     try:
         return tool_manager.get_tool(name) is not None
     except Exception as err:  # noqa: BLE001 - analytics must not break the call
-        log(
-            f'Warning: could not determine whether "{name}" is a real tool of '
-            f"yours; delegating the call to your server - {err}"
-        )
+        warn_ownership_lookup_failed(name, err)
         return None
 
 

posthog/mcp/_instrument_lowlevel.py

@@ -41,6 +41,7 @@
     resolve_virtual_tool_injection,
     start_tool_call_lifecycle,
     start_tools_list_lifecycle,
+    warn_ownership_lookup_failed,
 )
 from ._internal import MCPAnalyticsData
 from ._model_parameters import request_meta_from_context
@@ -501,10 +502,7 @@ async def _name_owned_by_real_tool(
             # -- guessing that would swallow a real tool of theirs. What reaches
             # here is the visibility, transform and auth work layered on top of
             # the providers; a provider failure never does (see the docstring).
-            log(
-                f'Warning: could not determine whether "{name}" is a real tool of '
-                f"yours; delegating the call to your server - {err}"
-            )
+            warn_ownership_lookup_failed(name, err)
             return None
     # May be None: see `raw_listing_owns_tool_name`. Callers intercept only on a
     # definite False.

posthog/mcp/_instrument_v2.py

@@ -55,6 +55,7 @@
     resolve_virtual_tool_injection,
     start_tool_call_lifecycle,
     start_tools_list_lifecycle,
+    warn_ownership_lookup_failed,
 )
 from ._internal import MCPAnalyticsData
 from ._model_parameters import (
@@ -737,8 +738,5 @@ def _name_owned_by_real_tool_v2(high_level: Any, name: str) -> Optional[bool]:
     try:
         return high_level._tool_manager.get_tool(name) is not None
     except Exception as err:  # noqa: BLE001 - analytics must not break the call
-        log(
-            f'Warning: could not determine whether "{name}" is a real tool of '
-            f"yours; delegating the call to your server - {err}"
-        )
+        warn_ownership_lookup_failed(name, err)
         return None

posthog/mcp/_instrumentation.py

@@ -14,7 +14,7 @@
 import threading
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Literal, Optional, Set
 
 from ._capture import capture_event
 from ._context_parameters import (
@@ -59,6 +59,10 @@
 VIRTUAL_TOOL_MISSING_CAPABILITY = "missing_capability"
 VIRTUAL_TOOL_FEEDBACK = "feedback"
 
+# Why a collision happened, which picks the warning's wording. Named so a typo
+# is a type error instead of silently getting the "blocked" text.
+VirtualToolCollisionVariant = Literal["blocked", "duplicate", "shadowed"]
+
 # The option that renames each virtual tool, quoted verbatim in the collision
 # warnings.
 _VIRTUAL_TOOL_RENAME_OPTION = {
@@ -875,16 +879,26 @@ async def raw_listing_owns_tool_name(
     try:
         names = await probe(ctx)
     except Exception as err:  # noqa: BLE001 - analytics must not break the call
-        log(
-            f'Warning: could not determine whether "{name}" is a real tool of '
-            f"yours; delegating the call to your server - {err}"
-        )
+        warn_ownership_lookup_failed(name, err)
         return None
     return None if names is None else name in names
 
 
+def warn_ownership_lookup_failed(name: str, err: Exception) -> None:
+    """A tool-ownership lookup raised instead of answering. Every adapter's probe
+    reports it the same way and then returns ``None``, so the call is delegated
+    to the host rather than intercepted on a guess."""
+    log(
+        f'Warning: could not determine whether "{name}" is a real tool of '
+        f"yours; delegating the call to your server - {err}"
+    )
+
+
 def _warn_virtual_tool_collision(
-    data: MCPAnalyticsData, kind: str, name: str, variant: str
+    data: MCPAnalyticsData,
+    kind: str,
+    name: str,
+    variant: VirtualToolCollisionVariant,
 ) -> None:
     """Warn once per ``(kind, name, variant)`` for the life of the server's
     tracking state, so a client that re-lists tools on every turn doesn't flood
@@ -897,7 +911,11 @@ def _warn_virtual_tool_collision(
 
 
 def virtual_tool_collision_message(
-    kind: str, name: str, variant: str, *, rename_option: Optional[str] = None
+    kind: str,
+    name: str,
+    variant: VirtualToolCollisionVariant,
+    *,
+    rename_option: Optional[str] = None,
 ) -> str:
     """The warning text for a virtual-tool name collision. Always names the option
     that renames PostHog's tool -- a warning without its own remedy gets ignored.

posthog/mcp/posthog_mcp.py

@@ -27,6 +27,7 @@
 from ._instrumentation import (
     VIRTUAL_TOOL_FEEDBACK,
     VIRTUAL_TOOL_MISSING_CAPABILITY,
+    VirtualToolCollisionVariant,
     drain_pending_sync,
     fire_and_forget,
     virtual_tool_collision_message,
@@ -398,38 +399,42 @@ def prepare_tool_list(
         # straight back. Warning about that would be warning about ourselves.
         if report_missing:
             name = self._missing_capability_tool_name
-            if self._real_tool_owns_name(prepared, name):
+            existing = _find_tool(prepared, name)
+            if existing is not None and not self._is_sdk_virtual_tool(existing):
                 self._warn_virtual_tool_collision(
                     VIRTUAL_TOOL_MISSING_CAPABILITY,
                     name,
                     'PostHogMCP(missing_capability_tool_name="...")',
                 )
-            elif not any(_tool_name(t) == name for t in prepared):
+            elif existing is None:
                 prepared.append(build_report_missing_descriptor(name))
         if collect_feedback and self._collect_feedback is not None:
             name = self._feedback_tool_name
-            if self._real_tool_owns_name(prepared, name):
+            existing = _find_tool(prepared, name)
+            # Both virtual tools under one name: missing-capability wins in
+            # `prepare_tool_call`, so advertising this one too would dead-letter
+            # the feedback path. Same precedence as `instrument()`.
+            duplicate = name == self._missing_capability_tool_name
+            if duplicate or (
+                existing is not None and not self._is_sdk_virtual_tool(existing)
+            ):
                 self._warn_virtual_tool_collision(
                     VIRTUAL_TOOL_FEEDBACK,
                     name,
                     'PostHogMCP(collect_feedback=CollectFeedbackOptions(tool_name="..."))',
+                    variant="duplicate" if duplicate else "blocked",
                 )
-            elif not any(_tool_name(t) == name for t in prepared):
+            elif existing is None:
                 prepared.append(get_feedback_tool_descriptor(self._collect_feedback))
         prepared = self._inject_models(prepared)
         return prepared
 
-    def _real_tool_owns_name(self, prepared: List[Any], name: str) -> bool:
-        """Whether a *host* tool in this listing owns ``name``. Our own
-        descriptor doesn't count: a host may re-prepare an already-prepared
-        list, and that is not a collision to warn about."""
-        return any(
-            _tool_name(tool) == name and not self._is_sdk_virtual_tool(tool)
-            for tool in prepared
-        )
-
     def _warn_virtual_tool_collision(
-        self, kind: str, name: str, rename_option: str
+        self,
+        kind: str,
+        name: str,
+        rename_option: str,
+        variant: VirtualToolCollisionVariant = "blocked",
     ) -> None:
         """Warn once per ``(kind, name)`` for this client's lifetime, so a host
         that prepares a listing on every request doesn't flood the log."""
@@ -439,7 +444,7 @@ def _warn_virtual_tool_collision(
         self._warned_virtual_tool_collisions.add(key)
         warn(
             virtual_tool_collision_message(
-                kind, name, "blocked", rename_option=rename_option
+                kind, name, variant, rename_option=rename_option
             )
         )
 
@@ -489,9 +494,13 @@ def prepare_tool_call(
         # the real tool wins — the stateless twin of the ownership check
         # instrument() runs. Without it the name match stands, and the
         # documented remedy for a collision is renaming PostHog's tool.
+        # The name check against missing-capability keeps one precedence when
+        # both tools share a name: without it both flags are true, and a
+        # dispatcher testing `is_feedback` first misroutes every call.
         is_feedback = (
             self._collect_feedback is not None
             and name == self._feedback_tool_name
+            and name != self._missing_capability_tool_name
             and original_tool is None
         )
         # Same guard for the missing-capability tool. Unlike feedback it has no
@@ -690,6 +699,12 @@ def _tool_name(tool: Any) -> Optional[str]:
     return getattr(tool, "name", None)
 
 
+def _find_tool(prepared: List[Any], name: str) -> Optional[Any]:
+    """The listed tool using ``name``, or ``None``. One pass answers both "is
+    the name taken" and "is the tool holding it ours"."""
+    return next((tool for tool in prepared if _tool_name(tool) == name), None)
+
+
 def _tool_schema(tool: Any) -> Optional[Dict[str, Any]]:
     if isinstance(tool, dict):
         schema = tool.get("inputSchema")

posthog/test/mcp/test_feedback.py

@@ -948,6 +948,31 @@ async def test_posthogmcp_prepare_tool_call_without_opt_in_never_flags():
     assert call.is_feedback is False and call.feedback_report is None
 
 
+async def test_posthogmcp_one_name_for_both_virtual_tools(caplog):
+    # Both virtual tools configured with the same name. Missing-capability wins
+    # every call path, so advertising the feedback tool too would dead-letter it
+    # — the stateless twin of the "duplicate" drop `instrument()` performs.
+    client, _ = make_client(
+        missing_capability_tool_name="assist",
+        collect_feedback=CollectFeedbackOptions(tool_name="assist"),
+    )
+    tools = [{"name": "search", "inputSchema": {"type": "object", "properties": {}}}]
+
+    with caplog.at_level("WARNING", logger="posthog.mcp"):
+        prepared = client.prepare_tool_list(
+            tools, report_missing=True, collect_feedback=True
+        )
+
+    assert [t["name"] for t in prepared] == ["search", "assist"]
+    assert any("both" in r.message for r in caplog.records if r.name == "posthog.mcp")
+
+    # A dispatcher testing `is_feedback` first must not swallow the call: the
+    # name belongs to missing-capability, so only that flag is set.
+    call = client.prepare_tool_call("assist", {"context": "need csv export"})
+    assert call.is_missing_capability is True
+    assert call.is_feedback is False and call.feedback_report is None
+
+
 async def test_posthogmcp_original_tool_wins_name_collision():
     # A host whose own list holds a real `send_feedback` tool passes it as
     # `original_tool`; the call then dispatches as a real tool call instead of