fix(mcp): send $mcp_initialize only for an initialize request

.changeset/mcp-initialize-only-on-initialize.md

@@ -0,0 +1,5 @@
+---
+"posthog-ruby": patch
+---
+
+`PostHog::MCP` now sends `$mcp_initialize` only for a client's `initialize` request. A session's first `tools/call`, listing or resource read without an `initialize` (stateless HTTP or a lone call) no longer emits one, so fewer `$mcp_initialize` events will arrive.

lib/posthog/mcp/instrumentation.rb

@@ -249,7 +249,7 @@ def dispatch_initialize
           yield
         rescue StandardError => e
           safely do
-            session_id = prepare_request(@request, skip_initialize: true)
+            session_id = prepare_request(@request)
             record_initialize(session_id, client_name, client_version, requested_version, error: e)
           end
           raise
@@ -258,7 +258,7 @@ def dispatch_initialize
         safely do
           negotiated = (result.is_a?(Hash) ? fetch(result, :protocolVersion) : nil) || requested_version
           minted = mint_session_token(client_name, client_version, negotiated, requested_version)
-          session_id = prepare_request(@request, skip_initialize: true, token: minted)
+          session_id = prepare_request(@request, token: minted)
           record_initialize(session_id, client_name, client_version, negotiated, response: result)
         end
         result
@@ -360,7 +360,6 @@ def record_tools_list(session_id, names:, response: nil, empty: false, error: ni
       end
 
       def record_initialize(session_id, client_name, client_version, protocol_version, response: nil, error: nil)
-        @data.mark_session_initialized(session_id)
         event = base_event(EventType::MCP_INITIALIZE, session_id, @request)
         event['client_name'] = client_name
         event['client_version'] = client_version
@@ -442,8 +441,8 @@ def prime_session(request, conversation_id)
         @scope[:session_id] = session_id if @scope.is_a?(Hash)
       end
 
-      # Resolve the session id, run identify, then lazily emit initialize.
-      def prepare_request(request, conversation_id: nil, skip_initialize: false, token: nil)
+      # Resolve the session id and run identify.
+      def prepare_request(request, conversation_id: nil, token: nil)
         token ||= @token
         session_id, source = Session.resolve(@data, mcp_session_id(token), token: token,
                                                                            conversation_id: conversation_id)
@@ -463,30 +462,9 @@ def prepare_request(request, conversation_id: nil, skip_initialize: false, token
           @scope[:actor] = actor if @scope.is_a?(Hash)
           self.class.capture_event(@data, identify_event, actor: actor) if identify_event
         end
-        maybe_emit_initialize(session_id, request) unless skip_initialize
         session_id
       end
 
-      def maybe_emit_initialize(session_id, request)
-        # Claiming is atomic: two threads opening the same session concurrently
-        # must not both get past the check and emit an initialize each.
-        return unless @data.claim_session_initialized(session_id)
-
-        name, version = client_identity(request)
-        event = {
-          'event_type' => EventType::MCP_INITIALIZE,
-          'session_id' => session_id,
-          'client_name' => name,
-          'client_version' => version,
-          'protocol_version' => protocol_version,
-          'timestamp' => Time.now.utc
-        }
-        props = resolve_event_properties({ method: 'initialize', params: {} })
-        event['properties'] = props unless props.nil?
-        TransportIdentity.stamp(event, @headers)
-        self.class.capture_event(@data, event, actor: actor_for(session_id))
-      end
-
       # Era is decided by the version the client *asked for*:
       # a client declaring the 2026-07-28 revision or later must not be answered
       # with an `Mcp-Session-Id`, even though this gem counter-offers a legacy version.

lib/posthog/mcp/tracking_data.rb

@@ -7,8 +7,6 @@ module MCP
     #
     # @api private
     class TrackingData
-      MAX_INITIALIZED_SESSIONS = 1000
-
       attr_reader :options, :sink, :server_name, :server_version, :identified_sessions,
                   :tool_descriptions, :tool_categories, :tool_output_instructions, :tool_owned_params
       attr_accessor :session_id, :session_source, :last_mcp_session_id, :last_activity, :warned_no_stateless_session,
@@ -40,7 +38,6 @@ def initialize(options:, sink:, server_name: nil, server_version: nil)
         # Which injected argument names the analytics layer owns per tool (i.e.
         # the tool does not declare them itself).
         @tool_owned_params = {}
-        @initialized_sessions = {}
       end
 
       def synchronize(&block)
@@ -51,31 +48,6 @@ def synchronize(&block)
         end
       end
 
-      def mark_session_initialized(session_id)
-        synchronize do
-          @initialized_sessions.delete(session_id)
-          @initialized_sessions[session_id] = true
-          @initialized_sessions.shift while @initialized_sessions.length > MAX_INITIALIZED_SESSIONS
-        end
-      end
-
-      def session_initialized?(session_id)
-        synchronize { @initialized_sessions.key?(session_id) }
-      end
-
-      # Claim the one lazy initialize event a session gets. Check and mark are a
-      # single critical section, so when concurrent first requests race only the
-      # caller that inserted the key sees `true` and emits `$mcp_initialize`.
-      def claim_session_initialized(session_id)
-        synchronize do
-          next false if @initialized_sessions.key?(session_id)
-
-          @initialized_sessions[session_id] = true
-          @initialized_sessions.shift while @initialized_sessions.length > MAX_INITIALIZED_SESSIONS
-          true
-        end
-      end
-
       def remember_tool(name, description: nil, category: nil, owned_params: nil, output_instructions: nil)
         synchronize do
           @tool_descriptions[name] = description if description.is_a?(String) && !description.empty?

spec/posthog/mcp/identity_and_session_spec.rb

@@ -141,25 +141,3 @@
     expect(described_class.resolve(data, nil, token: token)[0]).to eq('ses_tok')
   end
 end
-
-RSpec.describe PostHog::MCP::TrackingData do
-  let(:data) { described_class.new(options: PostHog::MCP::Options.new, sink: nil) }
-
-  it 'lets exactly one racing caller claim a session initialize' do
-    # Check and mark must be one critical section: two threads opening the same
-    # session concurrently would otherwise both emit an $mcp_initialize.
-    50.times do |round|
-      session_id = "ses_#{round}"
-      gate = Queue.new
-      threads = Array.new(12) do
-        Thread.new do
-          gate.pop
-          data.claim_session_initialized(session_id)
-        end
-      end
-      12.times { gate << true }
-      expect(threads.map(&:value).count(true)).to eq(1)
-      expect(data.session_initialized?(session_id)).to be(true)
-    end
-  end
-end

spec/posthog/mcp/instrument_spec.rb

@@ -329,12 +329,12 @@ def self.call(message: nil, **) = MCP::Tool::Response.new([{ type: 'text', text:
       response = server.handle(rpc(1, 'tools/call', { name: 'boom', arguments: { context: 'break it' } }))
       expect(response[:error][:code]).to eq(-32_603)
       events = drain_events(client)
-      expect(events.map { |e| e[:event] }).to eq(['$mcp_initialize', '$mcp_tool_call', '$exception'])
-      call = events[1][:properties]
+      expect(events.map { |e| e[:event] }).to eq(['$mcp_tool_call', '$exception'])
+      call = events[0][:properties]
       expect(call).to include('$mcp_is_error' => true, '$mcp_error_type' => 'ArgumentError',
                               '$mcp_error_message' => 'explode with token [redacted]')
       expect(call).not_to have_key('$mcp_response')
-      sibling = events[2][:properties]
+      sibling = events[1][:properties]
       expect(sibling['$exception_list'].map { |e| e['type'] }).to eq(%w[MCP::Server::RequestHandlerError ArgumentError])
       expect(sibling).to include('$mcp_tool_name' => 'boom', '$exception_level' => 'error')
       expect(sibling).not_to have_key('$mcp_source')
@@ -344,8 +344,8 @@ def self.call(message: nil, **) = MCP::Tool::Response.new([{ type: 'text', text:
       described_class.instrument(server, client, enable_exception_autocapture: false, enable_conversation_id: false)
       server.handle(rpc(1, 'tools/call', { name: 'soft_fail', arguments: {} }))
       events = drain_events(client)
-      expect(events.map { |e| e[:event] }).to eq(['$mcp_initialize', '$mcp_tool_call'])
-      expect(events[1][:properties]).to include('$mcp_is_error' => true, '$mcp_error_type' => 'Error',
+      expect(events.map { |e| e[:event] }).to eq(['$mcp_tool_call'])
+      expect(events[0][:properties]).to include('$mcp_is_error' => true, '$mcp_error_type' => 'Error',
                                                 '$mcp_error_message' => 'tool failed badly')
     end
 
@@ -363,12 +363,11 @@ def self.call(message: nil, **) = MCP::Tool::Response.new([{ type: 'text', text:
       server.handle(rpc(4, 'resources/read', { uri: 'file:///readme' }))
       events = drain_events(client)
       names = events.map { |e| e[:event] }
-      expect(names).to eq(['$mcp_initialize', '$mcp_prompts_list', '$mcp_prompt_get', '$mcp_resources_list',
-                           '$mcp_resource_read'])
-      expect(events[2][:properties]['$mcp_resource_name']).to eq('greet')
-      expect(events[2][:properties]).not_to have_key('$mcp_tool_name')
-      expect(events[4][:properties]['$mcp_resource_name']).to eq('file:///readme')
-      expect(events[4][:properties]['$mcp_response']['contents'][0]['text']).to eq('hello')
+      expect(names).to eq(['$mcp_prompts_list', '$mcp_prompt_get', '$mcp_resources_list', '$mcp_resource_read'])
+      expect(events[1][:properties]['$mcp_resource_name']).to eq('greet')
+      expect(events[1][:properties]).not_to have_key('$mcp_tool_name')
+      expect(events[3][:properties]['$mcp_resource_name']).to eq('file:///readme')
+      expect(events[3][:properties]['$mcp_response']['contents'][0]['text']).to eq('hello')
     end
 
     it 'redacts binary prompt messages and resource blobs on the way out' do
@@ -502,8 +501,8 @@ def stringify_result_keys(target)
       described_class.instrument(server, client, before_send: before_send)
       server.handle(rpc(1, 'tools/call', { name: 'boom', arguments: { context: 'secret intent' } }))
       events = drain_events(client)
-      expect(events.map { |e| e[:event] }).to eq(['$mcp_initialize', '$mcp_tool_call'])
-      expect(events[1][:properties]).not_to have_key('$mcp_intent')
+      expect(events.map { |e| e[:event] }).to eq(['$mcp_tool_call'])
+      expect(events[0][:properties]).not_to have_key('$mcp_intent')
     end
 
     it 'drops telemetry when before_send raises without breaking the tool response' do
@@ -578,7 +577,7 @@ def stringify_result_keys(target)
       events = drain_events(client)
       # The hook runs after truncation, so without a second pass the batch would
       # drop these whole for exceeding the per-message limit.
-      expect(events.map { |e| e[:event] }).to eq(['$mcp_initialize', '$mcp_tool_call'])
+      expect(events.map { |e| e[:event] }).to eq(['$mcp_tool_call'])
       events.each do |event|
         expect(JSON.generate(event).bytesize).to be <= PostHog::Defaults::Message::MAX_BYTES
         expect(event[:properties]['bloat']).to start_with('😀')
@@ -862,5 +861,49 @@ def stringify_result_keys(target)
       end
     end
   end
+
+  describe '$mcp_initialize' do
+    before do
+      server.define_prompt(name: 'greet', description: 'g', arguments: []) do |_args, **|
+        MCP::Prompt::Result.new(description: 'x', messages: [])
+      end
+      server.define_resource(uri: 'file:///readme', name: 'readme', mime_type: 'text/plain') do
+        [{ uri: 'file:///readme', text: 'hello' }]
+      end
+      described_class.instrument(server, client)
+    end
+
+    def event_names
+      drain_events(client).map { |event| event[:event] }
+    end
+
+    [
+      ['tools/call', { name: 'echo', arguments: { message: 'hi' } }, '$mcp_tool_call'],
+      ['tools/list', nil, '$mcp_tools_list'],
+      ['prompts/list', nil, '$mcp_prompts_list'],
+      ['prompts/get', { name: 'greet', arguments: {} }, '$mcp_prompt_get'],
+      ['resources/list', nil, '$mcp_resources_list'],
+      ['resources/read', { uri: 'file:///readme' }, '$mcp_resource_read']
+    ].each do |method, params, event_name|
+      it "is not sent for a lone #{method} without an initialize" do
+        server.handle(rpc(1, method, params))
+        expect(event_names).to eq([event_name])
+      end
+
+      it "is sent once, by the initialize, before a #{method}" do
+        server.handle(initialize_request)
+        server.handle(rpc(2, method, params))
+        expect(event_names).to eq(['$mcp_initialize', event_name])
+      end
+    end
+
+    it 'is not sent when concurrent first calls open one session without an initialize' do
+      threads = Array.new(12) do |i|
+        Thread.new { server.handle(rpc(i, 'tools/call', { name: 'echo', arguments: { message: "m#{i}" } })) }
+      end
+      threads.each(&:join)
+      expect(event_names).to eq(['$mcp_tool_call'] * 12)
+    end
+  end
 end
 # rubocop:enable Layout/LineLength