Add screenshot and screen record functionality

Author: akshitsinha

.gitignore

@@ -8,3 +8,6 @@ wheels/
 
 # Virtual environments
 .venv
+
+# Vscode
+.vscode/

pyproject.toml

@@ -6,4 +6,7 @@ readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
     "fastmcp>=2.4.0",
+    "mss>=10.0.0",
+    "python-ffmpeg>=2.0.12",
+    "screeninfo>=0.8.1",
 ]

src/devices/preview.py

@@ -0,0 +1,5 @@
+from fastmcp import FastMCP
+
+
+def register_tools(app: FastMCP) -> None:
+    pass

src/devices/printer.py

@@ -5,6 +5,7 @@
 import subprocess
 import re
 import tempfile
+import os
 
 
 class Printer:
@@ -103,6 +104,42 @@ async def print_file(
             finally:
                 temp_file.close()
 
+    @app.tool(
+        name="print_as_pdf",
+        description="Print a file as PDF to a specified device location",
+        tags=["printer"],
+    )
+    async def print_as_pdf(
+        file_data: Annotated[
+            bytes, Field(description="Binary data of the PDF file to be saved")
+        ],
+        file_format: Annotated[str, Field(description="File format (must be 'pdf')")],
+        output_path: Annotated[
+            str, Field(description="Full path where the PDF should be saved")
+        ],
+    ) -> Dict[str, Any]:
+        if file_format.lower() != "pdf":
+            return {"success": False, "error": "File format must be 'pdf'"}
+
+        try:
+            output_dir = os.path.dirname(output_path)
+            if output_dir and not os.path.exists(output_dir):
+                os.makedirs(output_dir, exist_ok=True)
+
+            with open(output_path, "wb") as pdf_file:
+                pdf_file.write(file_data)
+
+            return {
+                "success": True,
+                "output_path": output_path,
+                "file_size": len(file_data),
+            }
+
+        except OSError as e:
+            return {"success": False, "error": f"Failed to save PDF: {str(e)}"}
+        except Exception as e:
+            return {"success": False, "error": f"Unexpected error: {str(e)}"}
+
     @app.tool(
         name="get_print_job",
         description="Get information about a print job",

src/devices/screen.py

@@ -1,74 +1,209 @@
 from typing import Dict, List, Optional, Any
 from fastmcp import FastMCP
-
-
-class Screen:
-    def __init__(self, device_id: str, name: str):
-        self.device_id = device_id
-        self.name = name
+from screeninfo import get_monitors
+from datetime import datetime
+from mss import mss
+import tempfile
+import os
+import time
+from typing import Annotated
+from pydantic import Field
+from ffmpeg.asyncio import FFmpeg
+import shutil
 
 
 def register_tools(app: FastMCP) -> None:
     @app.tool(
-        name="list_displays", description="List all displays connected to the system"
+        name="list_displays",
+        description="List all displays connected to the system",
+        tags=["screen"],
     )
     async def list_displays() -> List[Dict[str, Any]]:
-        displays = [
-            {
-                "device_id": "display0",
-                "name": "Built-in Retina Display",
-                "resolution": "2560x1600",
-                "is_primary": True,
-            },
-            {
-                "device_id": "display1",
-                "name": "External HDMI Monitor",
-                "resolution": "3840x2160",
-                "is_primary": False,
-            },
-        ]
+        displays = []
+        monitors = get_monitors()
+
+        for i, monitor in enumerate(monitors):
+            displays.append(
+                {
+                    "device_id": f"display{i}",
+                    "name": f"Display {i}",
+                    "resolution": f"{monitor.width}x{monitor.height}",
+                    "is_primary": monitor.is_primary,
+                    "x": monitor.x,
+                    "y": monitor.y,
+                }
+            )
+
         return displays
 
-    @app.tool(name="capture_screen", description="Capture a screenshot from a display")
-    async def capture_screen(
-        device_id: Optional[str] = None,
-        save_path: Optional[str] = None,
-        region: Optional[str] = None,
+    @app.tool(
+        name="capture_screenshot",
+        description="Capture a screenshot from a display",
+        tags=["screen"],
+    )
+    async def capture_screenshot(
+        device_id: Annotated[
+            Optional[str],
+            Field(
+                description="The display identifier in format 'displayN' where N is the display index (e.g., 'display0', 'display1')",
+                default="display0",
+            ),
+        ] = "display0",
+        save_path: Annotated[
+            Optional[str],
+            Field(
+                description="The file path with file name where the screenshot should be saved. If None, a temporary file with timestamp will be created automatically"
+            ),
+        ] = None,
     ) -> Dict[str, Any]:
-        return {
-            "success": True,
-            "file_path": save_path or "/tmp/mcp-peripherals/screenshot.png",
-            "resolution": region or "Full Screen (2560x1600)",
-            "timestamp": "2025-05-23T12:34:56",
-            "device_id": device_id or "display0",
-        }
-
-    @app.tool(name="record_screen", description="Start recording a screen")
+        try:
+            display_index = 0
+            if device_id and device_id.startswith("display"):
+                try:
+                    display_index = int(device_id.replace("display", ""))
+                except ValueError:
+                    return {
+                        "success": False,
+                        "error": f"Invalid device_id format: {device_id}. Expected format: 'displayN'",
+                    }
+
+            monitors = get_monitors()
+            if not monitors:
+                return {"success": False, "error": "No displays found"}
+
+            if display_index >= len(monitors):
+                return {
+                    "success": False,
+                    "error": f"Display {device_id} not found. Available displays: 0-{len(monitors) - 1}",
+                }
+
+            with mss() as sct:
+                if save_path is None:
+                    temp_dir = tempfile.gettempdir()
+                    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+                    save_path = os.path.join(temp_dir, f"screenshot_{timestamp}.png")
+                else:
+                    if (
+                        os.path.isdir(save_path)
+                        or save_path.endswith("/")
+                        or save_path.endswith("\\")
+                    ):
+                        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+                        save_path = os.path.join(
+                            save_path, f"screenshot_{timestamp}.png"
+                        )
+                    else:
+                        if not os.path.splitext(save_path)[1]:
+                            save_path = save_path + ".png"
+
+                    os.makedirs(os.path.dirname(save_path), exist_ok=True)
+
+                sct.shot(mon=display_index + 1, output=save_path)
+                return {
+                    "success": True,
+                    "file_path": save_path,
+                }
+
+        except Exception as e:
+            return {"success": False, "error": f"Screenshot capture failed: {str(e)}"}
+
+    @app.tool(
+        name="record_screen", description="Start recording a screen", tags=["screen"]
+    )
     async def record_screen(
-        device_id: Optional[str] = None,
-        save_path: Optional[str] = None,
-        region: Optional[str] = None,
-        fps: int = 30,
-        audio: bool = False,
+        device_id: Annotated[
+            Optional[str],
+            Field(
+                description="The display identifier in format 'displayN' where N is the display index (e.g., 'display0', 'display1')",
+                default="display0",
+            ),
+        ] = "display0",
+        save_path: Annotated[
+            Optional[str],
+            Field(
+                description="The file path with or without file name where the video should be saved. If None, a temporary file with timestamp will be created automatically"
+            ),
+        ] = None,
+        duration: Annotated[
+            Optional[int],
+            Field(
+                description="Duration of the recording in seconds. If None, defaults to 10 seconds",
+                default=10,
+            ),
+        ] = 10,
+        fps: Annotated[
+            Optional[float],
+            Field(
+                description="Frames per second for the recording. If None, defaults to 15 fps",
+                default=15.0,
+            ),
+        ] = 15.0,
     ) -> Dict[str, Any]:
-        return {
-            "success": True,
-            "recording_id": "screen123456",
-            "file_path": save_path or "/tmp/mcp-peripherals/screen_recording.mp4",
-            "start_time": "2025-05-23T12:34:56",
-            "device_id": device_id or "display0",
-            "region": region or "Full Screen",
-            "fps": fps,
-            "with_audio": audio,
-        }
-
-    @app.tool(name="stop_screen_recording", description="Stop recording a screen")
-    async def stop_screen_recording(recording_id: str) -> Dict[str, Any]:
-        return {
-            "success": True,
-            "recording_id": recording_id,
-            "file_path": "/tmp/mcp-peripherals/screen_recording.mp4",
-            "duration": "00:02:34",
-            "file_size": "42.7 MB",
-            "resolution": "2560x1600",
-        }
+        try:
+            display_index = 0
+            if device_id and device_id.startswith("display"):
+                try:
+                    display_index = int(device_id.replace("display", ""))
+                except ValueError:
+                    return {
+                        "success": False,
+                        "error": f"Invalid device_id format: {device_id}. Expected format: 'displayN'",
+                    }
+
+            monitors = get_monitors()
+            if not monitors:
+                return {"success": False, "error": "No displays found"}
+
+            if display_index >= len(monitors):
+                return {
+                    "success": False,
+                    "error": f"Display {device_id} not found. Available displays: 0-{len(monitors) - 1}",
+                }
+
+            if save_path is None:
+                temp_dir = tempfile.gettempdir()
+                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+                save_path = os.path.join(temp_dir, f"screen_recording_{timestamp}.mp4")
+            else:
+                if (
+                    os.path.isdir(save_path)
+                    or save_path.endswith("/")
+                    or save_path.endswith("\\")
+                ):
+                    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+                    save_path = os.path.join(
+                        save_path, f"screen_recording_{timestamp}.mp4"
+                    )
+                else:
+                    if not os.path.splitext(save_path)[1]:
+                        save_path = save_path + ".mp4"
+
+                os.makedirs(os.path.dirname(save_path), exist_ok=True)
+
+            frames_dir = tempfile.mkdtemp()
+
+            with mss() as sct:
+                start_time = time.time()
+                frame_count = 0
+
+                while time.time() - start_time < duration:
+                    sct.shot(
+                        mon=display_index + 1,
+                        output=f"{frames_dir}/frame_{frame_count:06d}.png",
+                    )
+
+                    frame_count += 1
+
+            ffmpeg = (
+                FFmpeg()
+                .option("y")
+                .input(f"{frames_dir}/frame_%06d.png", framerate=fps)
+                .output(save_path, vcodec="libx264", pix_fmt="yuv420p")
+            )
+
+            await ffmpeg.execute()
+            shutil.rmtree(frames_dir)
+            return {"success": True, "file_path": save_path}
+
+        except Exception as e:
+            return {"success": False, "error": f"Screen recording failed: {str(e)}"}

src/main.py

@@ -1,5 +1,6 @@
 from server import create_app
 from config import get_settings
+import asyncio
 
 
 def main():
@@ -9,4 +10,4 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())