Claude IPC MCP

Instructions

Implementation Reference

• src/claude_ipc_server.py:971-983 (registration)

  Registration of the 'check' MCP tool including its input schema requiring 'instance_id'.

  Tool(
      name="check",
      description="Check for new messages",
      inputSchema={
          "type": "object",
          "properties": {
              "instance_id": {
                  "type": "string",
                  "description": "Your instance ID"
              }
          },
          "required": ["instance_id"]
      }

• src/claude_ipc_server.py:1140-1161 (handler)

  MCP tool handler for 'check' that delegates to the internal message broker via BrokerClient and formats the retrieved messages for display.

  elif name == "check":
      if not current_session_token:
          return [TextContent(type="text", text="Error: Not registered. Please register first.")]
          
      response = BrokerClient.send_request({
          "action": "check",
          "instance_id": arguments["instance_id"],
          "session_token": current_session_token
      })
      
      if response["status"] == "ok" and response.get("messages"):
          formatted = "New messages:\n"
          for msg in response["messages"]:
              formatted += f"\nFrom: {msg['from']}\n"
              formatted += f"Time: {msg['timestamp']}\n"
              formatted += f"Content: {msg['message']['content']}\n"
              if msg['message'].get('data'):
                  formatted += f"Data: {json.dumps(msg['message']['data'], indent=2)}\n"
          return [TextContent(type="text", text=formatted)]
      else:
          return [TextContent(type="text", text="No new messages")]

• src/claude_ipc_server.py:735-768 (helper)

  Internal MessageBroker handler for 'check' action: retrieves, clears unread messages for the instance, and marks them as read in the SQLite database.

  elif action == "check":
      # instance_id already validated and set from session
      instance_id = request["instance_id"]
      
      # Resolve name through forwarding if needed
      resolved_id = self._resolve_name(instance_id)
      
      if resolved_id not in self.queues:
          return {"status": "ok", "messages": []}
          
      messages = self.queues[resolved_id]
      self.queues[resolved_id] = []
      
      # Mark messages as read in database
      if self.db_path and messages:
          try:
              conn = sqlite3.connect(self.db_path)
              cursor = conn.cursor()
              
              # Get message IDs to mark as read
              for msg in messages:
                  cursor.execute('''
                      UPDATE messages 
                      SET read_flag = 1 
                      WHERE to_id = ? AND timestamp = ? AND read_flag = 0
                  ''', (resolved_id, msg.get("timestamp")))
              
              conn.commit()
              conn.close()
          except Exception as e:
              logger.error(f"Failed to mark messages as read: {e}")
      
      return {"status": "ok", "messages": messages}