diff --git a/nocobase_crm_mcp/nb_client.py b/nocobase_crm_mcp/nb_client.py
index ec19f72..17b827a 100644
--- a/nocobase_crm_mcp/nb_client.py
+++ b/nocobase_crm_mcp/nb_client.py
@@ -89,6 +89,34 @@ class NbClient:
         self._append_repeated(command, "--except", except_fields)
         return self._run_json(command)
 
+    def create_record(
+        self,
+        resource: str,
+        values: Mapping[str, Any],
+        whitelist: Sequence[str],
+    ) -> dict[str, Any]:
+        if not values:
+            raise ValueError("values must not be empty")
+        command = self._resource_command("create", resource)
+        self._append_json(command, "--values", values)
+        self._append_repeated(command, "--whitelist", whitelist)
+        return self._run_json(command)
+
+    def update_record(
+        self,
+        resource: str,
+        record_id: Any,
+        values: Mapping[str, Any],
+        whitelist: Sequence[str],
+    ) -> dict[str, Any]:
+        if not values:
+            raise ValueError("values must not be empty")
+        command = self._resource_command("update", resource)
+        self._append_json_or_scalar(command, "--filter-by-tk", record_id)
+        self._append_json(command, "--values", values)
+        self._append_repeated(command, "--whitelist", whitelist)
+        return self._run_json(command)
+
     def query_records(
         self,
         resource: str,
diff --git a/tests/test_nb_client.py b/tests/test_nb_client.py
index 1a4122c..f0245a0 100644
--- a/tests/test_nb_client.py
+++ b/tests/test_nb_client.py
@@ -86,6 +86,52 @@ class NbClientTests(unittest.TestCase):
         self.assertIn("--orders", commands[1])
         self.assertIn("--timezone", commands[1])
 
+    def test_create_record_uses_values_and_whitelist(self) -> None:
+        commands: list[list[str]] = []
+
+        def runner(cmd, **kwargs):
+            commands.append(cmd)
+            return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{"id":7}}', stderr="")
+
+        client = NbClient(env_name="nb-mcp", runner=runner)
+        client.create_record("t_projects", {"f_project_name": "EV inverter"}, ["f_project_name"])
+
+        self.assertEqual(commands[0][:4], ["nb", "api", "resource", "create"])
+        self.assertEqual(
+            json.loads(commands[0][commands[0].index("--values") + 1]),
+            {"f_project_name": "EV inverter"},
+        )
+        self.assertEqual(commands[0][commands[0].index("--whitelist") + 1], "f_project_name")
+
+    def test_update_record_uses_filter_by_tk(self) -> None:
+        commands: list[list[str]] = []
+
+        def runner(cmd, **kwargs):
+            commands.append(cmd)
+            return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{"id":7}}', stderr="")
+
+        client = NbClient(env_name="nb-mcp", runner=runner)
+        client.update_record("t_projects", 7, {"client_ID": 3}, ["client_ID"])
+
+        self.assertEqual(commands[0][:4], ["nb", "api", "resource", "update"])
+        self.assertEqual(commands[0][commands[0].index("--filter-by-tk") + 1], "7")
+
+    def test_record_writes_reject_empty_values_without_running_command(self) -> None:
+        commands: list[list[str]] = []
+
+        def runner(cmd, **kwargs):
+            commands.append(cmd)
+            return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{}}', stderr="")
+
+        client = NbClient(env_name="nb-mcp", runner=runner)
+
+        with self.assertRaises(ValueError):
+            client.create_record("t_projects", {}, ["f_project_name"])
+        with self.assertRaises(ValueError):
+            client.update_record("t_projects", 7, {}, ["f_project_name"])
+
+        self.assertEqual(commands, [])
+
     def test_list_collection_fields_uses_data_modeling_command(self) -> None:
         commands: list[list[str]] = []