fix(mcp): inject virtual tools once, on the first tools/list page

.sampo/changesets/mcp-virtual-tool-first-page.md

@@ -0,0 +1,7 @@
+---
+pypi/posthog: patch
+---
+
+MCP analytics now adds its virtual tools (`get_more_tools`, `send_feedback`) to the first `tools/list` page only, rather than to every page and to the last page respectively.
+
+If one of your own tools already uses a virtual tool's name, PostHog warns and names the option that renames its own — `missing_capability_tool_name`, or `collect_feedback`'s `tool_name`. Warnings also reach the `posthog.mcp` logger, so you see them without setting the `logger` option.

posthog/mcp/README.md

@@ -134,8 +134,8 @@ and adds no tools.
 The tool covers what `report_missing` covers (as feedback_type
 `missing_capability`), so new integrations should enable only one of the two.
 
-If a real tool already uses the name, the SDK logs a warning, does not inject
-the virtual tool, and never intercepts the real tool.
+See [Virtual tools on a paginated `tools/list`](#virtual-tools-on-a-paginated-toolslist)
+for name collisions and the page rule.
 
 Use the object form to rename the tool, declare host-specific fields, or route
 reports to a real backend:
@@ -184,6 +184,70 @@ if call.is_feedback:
 `on_feedback` is ignored on this path — the dispatcher routes reports itself via
 `call.feedback_report`.
 
+On this path you own the page rule, because you pass the switches per call. Pass
+them for the first page only, and pass your own tool as `original_tool` so a real
+tool by a virtual tool's name wins:
+
+```python
+first_page = request.params.get("cursor") is None
+tools = posthog.prepare_tool_list(
+    page_tools, report_missing=first_page, collect_feedback=first_page
+)
+
+call = posthog.prepare_tool_call(
+    tool_name, raw_args, original_tool=my_tools.get(tool_name)
+)
+```
+
+## Virtual tools on a paginated `tools/list`
+
+`instrument()` advertises up to two tools of its own: `get_more_tools`
+(`report_missing=True`) and `send_feedback` (`collect_feedback=True`).
+
+A client concatenates every `tools/list` page into one list, so each virtual tool
+is appended to the **first page only** — the page every client reads, including
+clients that never follow `nextCursor`. "First page" means a request with no
+cursor; an empty string is a valid opaque cursor, so `cursor: ""` is a
+continuation page.
+
+### Tool name collisions
+
+Your tools win when the SDK can see them. Rename the SDK's to keep both:
+
+```python
+MCPAnalyticsOptions(
+    report_missing=True,
+    missing_capability_tool_name="find_posthog_tools",
+    collect_feedback=CollectFeedbackOptions(tool_name="tell_posthog"),
+)
+```
+
+- A real tool using the name **on the first page** wins: the SDK warns, injects
+  nothing, and never intercepts yours.
+- A real tool that appears **only on a later page** is shadowed — page one cannot
+  see page two, so the SDK's tool is already advertised and calls to the name
+  reach it. The SDK warns when that page is served. `@posthog/mcp` is the same.
+- Configuring **both** virtual tools with one name advertises only
+  `get_more_tools`, and warns.
+
+Warnings go to the `logger` option *and* the `posthog.mcp` standard-library
+logger, so a default-configured host sees them on stderr.
+
+At call time the SDK checks ownership again, covering the process that serves a
+call without having served a listing — the ordinary multi-pod case. FastMCP and
+v2 `MCPServer` are asked via their tool registry; a raw low-level server has
+none, so the SDK calls your own `tools/list` handler, once per call to a virtual
+tool's name and never for ordinary traffic. If that check cannot answer — a
+registry lookup or a listing handler raised, or you registered `tools/list`
+after `instrument()` — the call is delegated to your server rather than
+intercepted, and the reason is logged: guessing the other way would swallow a
+real tool of yours silently.
+
+This check is the only ownership signal used at call time, so a server that
+serves **different tool sets to different callers** from one instrumented
+instance is handled correctly: nothing is carried over from whichever listing
+happened to be served last.
+
 ## Stateless / multi-pod servers
 
 A stateless MCP server issues no session id, so `$session_id` fragments across pods

posthog/mcp/_conversation_id.py

@@ -78,20 +78,27 @@ def resolve_conversation_id(
     enabled: bool,
     args: Any,
     tool_name: Optional[str],
-    missing_capability_tool_name: str,
+    missing_capability_tool_name: Optional[str],
     feedback_tool_name: Optional[str] = None,
 ) -> Tuple[Optional[str], bool]:
     """Return ``(conversation_id, minted)``. Disabled, get_more_tools, or
     send_feedback → ``(None, False)``; agent echoed a handle we could have minted
     → ``(value, False)``; anything else (omitted, or a value the agent made up)
     → ``(new uuid, True)``.
 
+    Either virtual tool's name arrives as ``None`` when that tool is disabled,
+    so a real application tool by the same name mints and echoes a handle like
+    any other tool's.
+
     Lowercased on the way in: the shape test is case-insensitive but the hash
     behind ``$session_id`` is not, so an uppercased echo (some hosts normalise
     uuids) would land in a different session than the call that minted it."""
     if (
         not enabled
-        or tool_name == missing_capability_tool_name
+        or (
+            missing_capability_tool_name is not None
+            and tool_name == missing_capability_tool_name
+        )
         or (feedback_tool_name is not None and tool_name == feedback_tool_name)
     ):
         return None, False

posthog/mcp/_instrument_fastmcp.py

@@ -30,17 +30,17 @@
 from ._instrument_lowlevel import _wrap_resource_requests
 from ._instrumentation import (
     _to_jsonable,
-    append_get_more_tools,
-    append_send_feedback,
+    apply_virtual_tool_injection,
     collect_listed_tools,
     extract_tools,
-    listing_has_next_page,
+    is_first_listing_page,
     mutate_tool_schema,
-    refresh_feedback_shadow,
     request_to_dict,
+    resolve_virtual_tool_injection,
     resolve_session_and_client,
     start_tool_call_lifecycle,
     start_tools_list_lifecycle,
+    warn_ownership_lookup_failed,
 )
 from ._internal import MCPAnalyticsData
 from ._model_parameters import (
@@ -50,7 +50,7 @@
 )
 from ._output_instructions import mirror_instructions_into_structured_content
 from .logger import log
-from .tools import get_more_tools_result_text, resolve_missing_capability_tool_name
+from .tools import get_more_tools_result_text
 
 _WRAPPED_FLAG = "__posthog_mcp_wrapped__"
 
@@ -115,15 +115,15 @@ async def wrapped(
             },
         )
 
-        if lifecycle.is_missing_capability:
+        if lifecycle.is_missing_capability and (
+            _name_owned_by_real_tool(server, name) is False
+        ):
             await lifecycle.record_missing_capability()
             return [
                 mcp_types.TextContent(type="text", text=get_more_tools_result_text())
             ]
 
-        if lifecycle.is_feedback and not _feedback_name_owned_by_real_tool(
-            server, name
-        ):
+        if lifecycle.is_feedback and (_name_owned_by_real_tool(server, name) is False):
             reply = await lifecycle.record_feedback()
             return [mcp_types.TextContent(type="text", text=reply)]
 
@@ -206,6 +206,7 @@ def _inject_tool_schemas(server: Any, data: MCPAnalyticsData, tools: list) -> No
             schema_attribute="inputSchema",
             owns_context=_tool_owns_context(server, tool.name),
             context_required=True,
+            is_sdk_virtual_tool=False,
         )
 
 
@@ -229,10 +230,6 @@ async def list_handler(req: Any) -> Any:
         if req is None:
             result = await original(req)
             tools = extract_tools(result)
-            # Refresh the collision flag here too: this pass sees the real tool
-            # registry, so a real tool named like the feedback tool is detected
-            # before any client-facing listing.
-            refresh_feedback_shadow(data, tools)
             _inject_tool_schemas(server, data, tools)
             return result
 
@@ -272,19 +269,17 @@ async def list_handler(req: Any) -> Any:
         tools = extract_tools(result)
         # Empty is computed before adding the virtual missing-capability tool.
         names, empty = collect_listed_tools(data, tools)
-        feedback_name = refresh_feedback_shadow(data, tools)
+        injection = resolve_virtual_tool_injection(
+            data,
+            tools,
+            is_first_page=is_first_listing_page(getattr(req, "params", None)),
+        )
 
         _inject_tool_schemas(server, data, tools)
 
-        if data.options.report_missing:
-            missing_name = resolve_missing_capability_tool_name(data.options)
-            if not any(t.name == missing_name for t in tools):
-                append_get_more_tools(result, missing_name, data)
-                names.append(missing_name)
-
-        if feedback_name is not None and not listing_has_next_page(result):
-            append_send_feedback(result, data)
-            names.append(feedback_name)
+        result = apply_virtual_tool_injection(
+            result, injection, names, data, schema_field="inputSchema"
+        )
 
         await lifecycle.record_result(
             names=names,
@@ -322,14 +317,18 @@ def _inject_prompt_back(result: Any, conversation_id: str) -> Any:
     return result
 
 
-def _feedback_name_owned_by_real_tool(server: Any, name: str) -> bool:
-    """Live registry probe so a real tool by the feedback tool's name is never
-    shadowed even before the first listing refreshes the collision flag."""
-    try:
-        tool_manager = getattr(server, "_tool_manager", None)
-        return tool_manager is not None and tool_manager.get_tool(name) is not None
-    except Exception:  # noqa: BLE001 - unknown tool -> the name is not owned
+def _name_owned_by_real_tool(server: Any, name: str) -> Optional[bool]:
+    """Live registry probe, so a real tool by a virtual tool's name is never
+    shadowed. Tri-state like its low-level twin: ``None`` when the lookup failed
+    rather than answered, and callers must not intercept on it."""
+    tool_manager = getattr(server, "_tool_manager", None)
+    if tool_manager is None:
         return False
+    try:
+        return tool_manager.get_tool(name) is not None
+    except Exception as err:  # noqa: BLE001 - analytics must not break the call
+        warn_ownership_lookup_failed(name, err)
+        return None
 
 
 def _tool_owns_param(server: Any, name: str, param: str) -> bool: