feat(mcp): capture model identifiers (#927)

* feat(mcp): capture model identifiers

Capture model identifiers from recognized client metadata, with an SDK-owned self-report fallback for other clients. Preserve source provenance and fail closed when the application owns the field.\n\nVerify both MCP Python SDK 1.x and 2.x, including a 2026-07-28 wire-level call.

* fix(mcp): preserve model capture opt-in and tool ownership

Gate custom-dispatcher model resolution behind capture_model and copy
object tools before model injection so repeated listings preserve ownership.
Publish the completed ownership map without an intermediate empty state.

Keep additionalProperties constraints in all analytics schema injectors.
Document adapter-specific requiredness rather than requiring fields that
standalone FastMCP strips before input validation. Share the eligibility
predicate and update prepare helper documentation and the changeset.

Verification:
- MCP v1 suite: 267 passed.
- MCP v2 suite: 245 passed, 13 expected skips.
- Ruff check and format check passed repository-wide.
- Repository mypy/baseline check passed (229 source files).
- Public API snapshot and git diff --check passed.
- Regression cases failed before the fix for disabled capture, object
  ownership and strict validation through a low-level tool-cache rebuild.

Full non-MCP test suite not rerun locally; CI covers the broader matrix.

* chore(mcp): address release note and test dependency feedback

Clarify that preserving strict schemas during context and conversation-ID injection also affects existing users with model capture disabled. Declare jsonschema>=4.0 in the test extra and refresh only its direct dependency entries in uv.lock.

Verification: python -m pytest posthog/test/mcp --timeout=30 -q passed (267 tests, MCP SDK 1.29.0); uv lock --check and git diff --check passed. Configured pre-commit checks skipped these non-Python files. CodeScene reported no applicable changes. MCP v2 was not rerun locally for this documentation and dependency-only update.

* fix(mcp): preserve positional constructor compatibility

Append model capture fields after the existing public dataclass fields. Existing positional MCPAnalyticsOptions calls keep their identity and callback bindings, and PreparedToolCall retains the fourth positional missing-capability flag.

Add four parameterized regression cases and regenerate the public API snapshot. The cases failed before the field reorder and pass afterward.

Verification: 271 MCP tests passed; public_api_snapshot and public_api_check passed; configured Ruff pre-commit checks passed; mypy baseline check found no issues in 230 source files; strict import and git diff checks passed. CodeScene checked both Python files with no issues. MCP v2 validation remains covered by CI.

Author: lucasheriques

.sampo/changesets/bold-king-mielikki.md

@@ -0,0 +1,8 @@
+---
+pypi/posthog: minor
+---
+
+Capture MCP model identifiers from client metadata or an SDK-owned self-report field.
+Model capture remains opt-in and preserves application-owned fields across repeated tool listings.
+
+MCP context and conversation-ID injection now preserve `additionalProperties: false` in tool schemas, including when model capture is disabled. Servers that validate these schemas now reject undeclared arguments that earlier SDK versions allowed. Declared analytics fields remain valid.

posthog/mcp/README.md

@@ -21,6 +21,75 @@ Request headers use the same identity and package version, so SDK Health can com
 Because `$lib` is a client-level identity, `instrument()` relabels every event sent by the client passed to it.
 Use a client dedicated to MCP analytics if the application also captures unrelated events.
 
+## Capture the calling model
+
+Model capture is off by default. Enable it for an instrumented MCP Python SDK 1.x or
+2.x server:
+
+```python
+from posthog.mcp import MCPAnalyticsOptions, instrument
+
+analytics = instrument(
+    server,
+    posthog,
+    MCPAnalyticsOptions(capture_model=True),
+)
+```
+
+The SDK records the best model identifier visible to the server as
+`$mcp_llm_model`. Recognized client metadata wins and sets
+`$mcp_llm_model_source` to `client_metadata`. The SDK also adds an `llm_model`
+string to each compatible tool schema as a fallback, recorded with source
+`self_reported`. It is required for custom dispatchers and the official high-level
+MCP SDK adapters. Raw low-level servers and standalone `fastmcp.FastMCP` advertise
+it as optional; the standalone adapter strips it before input validation, so
+requiring it would reject calls. Existing schema strictness is preserved when
+analytics fields are added.
+
+The recognized metadata path is Codex's `x-codex-turn-metadata.model` field in
+request `_meta`. Other clients, including Claude Code, use the self-report path
+until they expose a stable model field. Missing, blank, and `unknown` values are
+not recorded.
+
+MCP does not standardize or attest model identity. Both sources are unverified.
+Use them to compare tool behavior across models, not for billing or access
+control.
+
+Model self-reporting only runs when PostHog can prove it owns the injected field.
+If a tool already declares `llm_model`, or uses a root `$ref`, `oneOf`, `allOf`, or
+`anyOf` schema, PostHog leaves the schema and argument untouched. Client metadata
+can still be captured in those cases.
+
+For a custom dispatcher, use the same option on `PostHogMCP` and pass request
+metadata through explicitly:
+
+```python
+from posthog.mcp import PostHogMCP
+
+posthog = PostHogMCP("phc_...", capture_model=True)
+tools = posthog.prepare_tool_list(server_tools)
+original_tool = next(tool for tool in server_tools if tool["name"] == tool_name)
+call = posthog.prepare_tool_call(
+    tool_name,
+    raw_args,
+    request_meta=request.get("params", {}).get("_meta"),
+    original_tool=original_tool,
+)
+result = dispatch(tool_name, call.args)
+posthog.capture_tool_call(
+    tool_name,
+    llm_model=call.llm_model,
+    llm_model_source=call.llm_model_source,
+)
+```
+
+Passing `original_tool` keeps ownership accurate when `tools/list` and
+`tools/call` reach different server replicas. A persistent single-process
+dispatcher can omit it after calling `prepare_tool_list()`.
+Model injection copies tool objects instead of changing their original schemas.
+Always advertise the returned list and pass the original application tool to
+`prepare_tool_call()`. Repeatedly preparing the original list preserves ownership.
+
 ## Stateless / multi-pod servers
 
 A stateless MCP server issues no session id, so `$session_id` fragments across pods

posthog/mcp/__init__.py

@@ -73,6 +73,8 @@
 from .types import (
     CaptureEventData,
     MCPAnalyticsContextOptions,
+    MCPAnalyticsModelOptions,
+    MCPAnalyticsModelSource,
     MCPAnalyticsOptions,
     PreparedToolCall,
     UserIdentity,
@@ -85,6 +87,8 @@
     "PostHogMCP",
     "MCPAnalyticsOptions",
     "MCPAnalyticsContextOptions",
+    "MCPAnalyticsModelOptions",
+    "MCPAnalyticsModelSource",
     "UserIdentity",
     "CaptureEventData",
     "PreparedToolCall",

posthog/mcp/_capture.py

@@ -63,6 +63,8 @@ def capture_event(
         "response": event_input.get("response"),
         "user_intent": event_input.get("user_intent"),
         "user_intent_source": event_input.get("user_intent_source"),
+        "llm_model": event_input.get("llm_model"),
+        "llm_model_source": event_input.get("llm_model_source"),
         "is_error": event_input.get("is_error"),
         "error": event_input.get("error"),
         "error_type": event_input.get("error_type"),

posthog/mcp/_context_parameters.py

@@ -79,10 +79,7 @@ def add_context_parameter_to_schema(
     if not isinstance(schema.get("properties"), dict):
         schema["properties"] = {}
 
-    # additionalProperties: false would reject the injected context — remove it
-    # (the SDK adds this when converting Pydantic models to JSON Schema).
-    if schema.get("additionalProperties") is False:
-        schema.pop("additionalProperties", None)
+    # The declared context property is allowed even under additionalProperties: false.
 
     schema["properties"]["context"] = {
         "type": "string",

posthog/mcp/_conversation_id.py

@@ -57,8 +57,6 @@ def add_conversation_id_to_schema(
     schema = copy.deepcopy(schema)
     if not isinstance(schema.get("properties"), dict):
         schema["properties"] = {}
-    if schema.get("additionalProperties") is False:
-        schema.pop("additionalProperties", None)
     schema["properties"][CONVERSATION_ID_PARAM_NAME] = {
         "type": "string",
         "description": DEFAULT_CONVERSATION_ID_DESCRIPTION,

posthog/mcp/_instrument_fastmcp.py

@@ -39,6 +39,11 @@
     start_tools_list_lifecycle,
 )
 from ._internal import MCPAnalyticsData
+from ._model_parameters import (
+    can_inject_model_parameter,
+    is_capture_model_enabled,
+    request_meta_from_context,
+)
 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
@@ -90,6 +95,8 @@ async def wrapped(
             data,
             name=name,
             arguments=arguments,
+            request_meta=request_meta_from_context(_tool_call_request_context(context)),
+            allow_self_reported_model=_analytics_owns_model(server, data, name),
             mcp_session_id=mcp_session_id,
             token=token,
             client_name=client_name,
@@ -119,6 +126,8 @@ async def wrapped(
                 server, name, "conversation_id"
             ):
                 strip_keys.add("conversation_id")
+            if _analytics_owns_model(server, data, name):
+                strip_keys.add("llm_model")
             if strip_keys:
                 call_arguments = {
                     k: v for k, v in arguments.items() if k not in strip_keys
@@ -251,7 +260,7 @@ async def list_handler(req: Any) -> Any:
         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)
+                append_get_more_tools(result, missing_name, data)
                 names.append(missing_name)
 
         await lifecycle.record_result(
@@ -310,6 +319,16 @@ def _tool_owns_context(server: Any, name: str) -> bool:
     return _tool_owns_param(server, name, "context")
 
 
+def _analytics_owns_model(server: Any, data: MCPAnalyticsData, name: str) -> bool:
+    if not is_capture_model_enabled(data.options.capture_model):
+        return False
+    try:
+        tool = server._tool_manager.get_tool(name)
+        return can_inject_model_parameter(getattr(tool, "parameters", None))
+    except Exception:  # noqa: BLE001 - model analytics must never break dispatch
+        return data.tool_model_parameter_injected.get(name, False)
+
+
 def _tool_call_request_context(context: Any) -> Any:
     """The request context behind a FastMCP ``Context``, or ``None``.
 

posthog/mcp/_instrument_lowlevel.py

@@ -34,6 +34,7 @@
     start_tools_list_lifecycle,
 )
 from ._internal import MCPAnalyticsData
+from ._model_parameters import request_meta_from_context
 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
@@ -97,6 +98,10 @@ async def handler(req: Any) -> Any:
             data,
             name=name,
             arguments=arguments,
+            request_meta=request_meta_from_context(_request_context(server)),
+            allow_self_reported_model=data.tool_model_parameter_injected.get(
+                name, False
+            ),
             mcp_session_id=mcp_session_id,
             token=token,
             client_name=client_name,
@@ -127,7 +132,10 @@ async def handler(req: Any) -> Any:
         # tools/list and across stateless per-request server instances.
         if strip_injected and req.params.arguments:
             owned = await _tool_owned_injected_keys(high_level, name)
-            for key in ("context", "conversation_id"):
+            injected_keys = ["context", "conversation_id"]
+            if data.tool_model_parameter_injected.get(name, False):
+                injected_keys.append("llm_model")
+            for key in injected_keys:
                 if key not in owned:
                     req.params.arguments.pop(key, None)
 
@@ -289,7 +297,7 @@ async def handler(req: Any) -> Any:
         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)
+                append_get_more_tools(result, missing_name, data)
                 names.append(missing_name)
 
         await lifecycle.record_result(

posthog/mcp/_instrument_v2.py

@@ -47,6 +47,11 @@
     start_tools_list_lifecycle,
 )
 from ._internal import MCPAnalyticsData
+from ._model_parameters import (
+    can_inject_model_parameter,
+    is_capture_model_enabled,
+    request_meta_from_context,
+)
 from ._output_instructions import mirror_instructions_into_structured_content
 from .logger import log
 from .request_headers import get_request_headers
@@ -222,6 +227,18 @@ def _tool_owns_param_v2(high_level: Any, name: str, param: str) -> bool:
     return param in _tool_own_properties_v2(high_level, name)
 
 
+def _analytics_owns_model_v2(
+    high_level: Any, data: MCPAnalyticsData, name: str
+) -> bool:
+    if not is_capture_model_enabled(data.options.capture_model):
+        return False
+    try:
+        tool = high_level._tool_manager.get_tool(name)
+        return can_inject_model_parameter(getattr(tool, "parameters", None))
+    except Exception:  # noqa: BLE001 - model analytics must never break dispatch
+        return data.tool_model_parameter_injected.get(name, False)
+
+
 # --- high-level: ToolManager.call_tool seam --------------------------------------
 
 
@@ -249,6 +266,8 @@ async def wrapped(
             data,
             name=name,
             arguments=arguments,
+            request_meta=request_meta_from_context(ctx),
+            allow_self_reported_model=_analytics_owns_model_v2(server, data, name),
             mcp_session_id=mcp_session_id,
             token=token,
             client_name=client_name,
@@ -281,6 +300,8 @@ async def wrapped(
                 and "conversation_id" not in own_properties
             ):
                 strip_keys.add("conversation_id")
+            if _analytics_owns_model_v2(server, data, name):
+                strip_keys.add("llm_model")
             if strip_keys:
                 call_arguments = {
                     k: v for k, v in arguments.items() if k not in strip_keys
@@ -389,6 +410,10 @@ async def handler(ctx: Any, params: Any) -> Any:
             data,
             name=name,
             arguments=arguments,
+            request_meta=request_meta_from_context(ctx),
+            allow_self_reported_model=data.tool_model_parameter_injected.get(
+                name, False
+            ),
             mcp_session_id=mcp_session_id,
             token=token,
             client_name=client_name,
@@ -504,7 +529,7 @@ async def handler(ctx: Any, params: Any) -> Any:
         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_v2(result, missing_name)
+                _append_get_more_tools_v2(result, missing_name, data)
                 names.append(missing_name)
 
         await lifecycle.record_result(
@@ -520,14 +545,21 @@ async def handler(ctx: Any, params: Any) -> Any:
     _replace_handler(server, _LIST_METHOD, handler, entry.params_type)
 
 
-def _append_get_more_tools_v2(result: Any, name: str) -> None:
+def _append_get_more_tools_v2(result: Any, name: str, data: MCPAnalyticsData) -> None:
     descriptor = build_report_missing_descriptor(name)
     tool = mcp_types.Tool(
         name=descriptor["name"],
         description=descriptor["description"],
         input_schema=descriptor["inputSchema"],
         annotations=descriptor["annotations"],
     )
+    mutate_tool_schema(
+        data,
+        tool,
+        schema_attribute="input_schema",
+        owns_context=True,
+        context_required=True,
+    )
     tools_list = getattr(result, "tools", None)
     if isinstance(tools_list, list):
         tools_list.append(tool)