feat: add browser session recording via CDP screencast (start/stop API)

cdp_use/client.py

@@ -4,6 +4,7 @@
 import re
 import time
 from typing import TYPE_CHECKING, Any, Dict, Optional
+from cdp_use.recorder import Recorder
 
 import websockets
 
@@ -400,3 +401,14 @@ async def emit_event(
         produced by your application rather than the browser.
         """
         return await self._event_registry.handle_event(method, params or {}, session_id)
+
+    async def start_recording(self, output_dir: str):
+        """
+        Start recording browser session frames.
+
+        :param output_dir: Directory to save frames
+        :return: Recorder instance
+        """
+        recorder = Recorder(self, output_dir)
+        await recorder.start()
+        return recorder

cdp_use/recorder.py

@@ -0,0 +1,85 @@
+import os
+import base64
+import asyncio
+from typing import Any
+
+
+class Recorder:
+    """
+    Records browser frames using CDP screencast and saves them as JPEG images.
+    """
+
+    def __init__(self, client: Any, output_dir: str):
+        self.client = client
+        self.output_dir = output_dir
+        self.frame_count = 0
+        self._running = False
+        self._queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
+        self._worker_task: asyncio.Task | None = None
+
+    async def start(self) -> None:
+        """
+        Start recording frames.
+        """
+        os.makedirs(self.output_dir, exist_ok=True)
+        self._running = True
+
+        async def worker():
+            while self._running or not self._queue.empty():
+                event = await self._queue.get()
+
+                self.frame_count += 1
+
+                # Decode base64 frame into binary image
+                image_data = base64.b64decode(event["data"])
+
+                filename = os.path.join(
+                    self.output_dir, f"frame_{self.frame_count:04d}.jpg"
+                )
+
+                with open(filename, "wb") as f:
+                    f.write(image_data)
+
+                print(f"Saved {filename}")
+
+                # Acknowledge frame so Chrome continues sending frames
+                await self.client.send.Page.screencastFrameAck({
+                    "sessionId": event["sessionId"]
+                })
+
+                self._queue.task_done()
+
+        def on_frame(event: dict, session_id: str) -> None:
+            if self._running:
+                self._queue.put_nowait(event)
+
+        self.client.register.Page.screencastFrame(on_frame)
+
+        self._worker_task = asyncio.create_task(worker())
+
+        await self.client.send.Page.enable()
+
+        await self.client.send.Page.startScreencast({
+            "format": "jpeg",
+            "quality": 50,
+            "everyNthFrame": 1
+        })
+
+    async def stop(self) -> None:
+        """
+        Stop recording and finalize frame saving.
+        """
+        self._running = False
+
+        await self.client.send.Page.stopScreencast()
+
+        await self._queue.join()
+
+        if self._worker_task:
+            self._worker_task.cancel()
+            try:
+                await self._worker_task
+            except asyncio.CancelledError:
+                pass
+
+        print(f"Recording saved to {self.output_dir}")

examples/record.py

@@ -0,0 +1,22 @@
+import asyncio
+from cdp_use.client import CDPClient
+
+
+async def main():
+    ws_url = "ws://localhost:9222/devtools/page/XXXX"
+
+    async with CDPClient(ws_url) as client:
+        # Start recording browser session
+        recorder = await client.start_recording("recording_output")
+
+        try:
+            # Perform actions while recording
+            await client.send.Page.navigate({"url": "https://youtube.com"})
+            await asyncio.sleep(5)
+        finally:
+            # Ensure recording is properly stopped
+            await recorder.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())