feat(mcp): capture model identifiers

.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)