ruff fix

Author: haixuanTao

src/reachy_mini_conversation_demo/mcp_server.py

@@ -9,11 +9,11 @@
 import json
 import logging
 import time
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import cv2
-import mcp.types as types
 import numpy as np
+from mcp import types
 from mcp.server import NotificationOptions, Server
 
 # MCP Server imports
@@ -43,8 +43,8 @@ def __init__(
         self,
         reachy_mini: ReachyMini,
         movement_manager: MovementManager,
-        camera: Optional[cv2.VideoCapture] = None,
-        vision_manager: Optional[VisionManager] = None,
+        camera: cv2.VideoCapture | None = None,
+        vision_manager: VisionManager | None = None,
         camera_retry_attempts: int = 5,
         camera_retry_delay_s: float = 0.10,
         vision_timeout_s: float = 8.0,
@@ -62,7 +62,7 @@ def __init__(
 
 # Helper functions
 def _read_frame(
-    cap: cv2.VideoCapture, attempts: int = 5, delay_s: float = 0.1
+    cap: cv2.VideoCapture, attempts: int = 5, delay_s: float = 0.1,
 ) -> np.ndarray:
     """Read a frame from the camera with retries."""
     trials, frame, ret = 0, None, False
@@ -77,7 +77,7 @@ def _read_frame(
     return frame
 
 
-def _execute_motion(config: ReachyMCPConfig, target: Any) -> Dict[str, Any]:
+def _execute_motion(config: ReachyMCPConfig, target: Any) -> dict[str, Any]:
     """Apply motion to reachy_mini and update movement_manager state."""
     movement_manager = config.movement_manager
     movement_manager.moving_start = time.monotonic()
@@ -111,7 +111,7 @@ def _setup_handlers(self):
         """Set up MCP server handlers"""
 
         @self.server.list_tools()
-        async def handle_list_tools() -> List[Tool]:
+        async def handle_list_tools() -> list[Tool]:
             """List available tools"""
             return [
                 Tool(
@@ -124,7 +124,7 @@ async def handle_list_tools() -> List[Tool]:
                                 "type": "string",
                                 "enum": ["left", "right", "up", "down", "front"],
                                 "description": "Direction to move the head",
-                            }
+                            },
                         },
                         "required": ["direction"],
                     },
@@ -138,7 +138,7 @@ async def handle_list_tools() -> List[Tool]:
                             "question": {
                                 "type": "string",
                                 "description": "Question to ask about the captured image",
-                            }
+                            },
                         },
                         "required": ["question"],
                     },
@@ -147,26 +147,25 @@ async def handle_list_tools() -> List[Tool]:
 
         @self.server.call_tool()
         async def handle_call_tool(
-            name: str, arguments: dict
-        ) -> List[types.TextContent | types.ImageContent | types.EmbeddedResource]:
+            name: str, arguments: dict,
+        ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
             """Handle tool calls"""
             if name == "move_head":
                 return await self._handle_move_head(arguments)
-            elif name == "camera":
+            if name == "camera":
                 return await self._handle_camera(arguments)
-            else:
-                raise ValueError(f"Unknown tool: {name}")
+            raise ValueError(f"Unknown tool: {name}")
 
         @self.server.list_resources()
-        async def handle_list_resources() -> List[Resource]:
+        async def handle_list_resources() -> list[Resource]:
             """List available resources"""
             return [
                 Resource(
                     uri="reachy://status",
                     name="Robot Status",
                     description="Current status of the Reachy Mini robot",
                     mimeType="application/json",
-                )
+                ),
             ]
 
         @self.server.read_resource()
@@ -178,34 +177,33 @@ async def handle_read_resource(uri: str) -> str:
                     "camera_available": self.config.camera is not None,
                     "vision_available": self.config.vision_manager is not None,
                     "current_head_pose": getattr(
-                        self.config.movement_manager, "current_head_pose", None
+                        self.config.movement_manager, "current_head_pose", None,
                     ),
                     "is_moving": self._is_robot_moving(),
                 }
                 return json.dumps(status, indent=2)
-            else:
-                raise ValueError(f"Unknown resource: {uri}")
+            raise ValueError(f"Unknown resource: {uri}")
 
     def _is_robot_moving(self) -> bool:
         """Check if robot is currently moving"""
         movement_manager = self.config.movement_manager
         if not hasattr(movement_manager, "moving_start") or not hasattr(
-            movement_manager, "moving_for"
+            movement_manager, "moving_for",
         ):
             return False
 
         elapsed = time.monotonic() - movement_manager.moving_start
         return elapsed < movement_manager.moving_for
 
-    async def _handle_move_head(self, arguments: dict) -> List[types.TextContent]:
+    async def _handle_move_head(self, arguments: dict) -> list[types.TextContent]:
         """Handle head movement tool call"""
         direction = arguments.get("direction")
         if not direction:
             return [
                 types.TextContent(
                     type="text",
                     text=json.dumps({"error": "direction parameter is required"}),
-                )
+                ),
             ]
 
         logger.info("MCP Tool call: move_head direction=%s", direction)
@@ -220,8 +218,8 @@ async def _handle_move_head(self, arguments: dict) -> List[types.TextContent]:
 
             return [
                 types.TextContent(
-                    type="text", text=json.dumps({"status": f"Now looking {direction}"})
-                )
+                    type="text", text=json.dumps({"status": f"Now looking {direction}"}),
+                ),
             ]
 
         except Exception as e:
@@ -230,35 +228,35 @@ async def _handle_move_head(self, arguments: dict) -> List[types.TextContent]:
                 types.TextContent(
                     type="text",
                     text=json.dumps(
-                        {"error": f"Head movement failed: {type(e).__name__}: {e}"}
+                        {"error": f"Head movement failed: {type(e).__name__}: {e}"},
                     ),
-                )
+                ),
             ]
 
-    async def _handle_camera(self, arguments: dict) -> List[types.TextContent]:
+    async def _handle_camera(self, arguments: dict) -> list[types.TextContent]:
         """Handle camera tool call"""
         question = arguments.get("question", "").strip()
         if not question:
             return [
                 types.TextContent(
                     type="text",
                     text=json.dumps({"error": "question must be a non-empty string"}),
-                )
+                ),
             ]
 
         if not self.config.camera:
             return [
                 types.TextContent(
-                    type="text", text=json.dumps({"error": "Camera not available"})
-                )
+                    type="text", text=json.dumps({"error": "Camera not available"}),
+                ),
             ]
 
         if not self.config.vision_manager:
             return [
                 types.TextContent(
                     type="text",
                     text=json.dumps({"error": "Vision manager not available"}),
-                )
+                ),
             ]
 
         logger.info("MCP Tool call: camera question=%s", question[:120])
@@ -274,7 +272,7 @@ async def _handle_camera(self, arguments: dict) -> List[types.TextContent]:
 
             # Process with vision manager
             result = await asyncio.to_thread(
-                self.config.vision_manager.processor.process_image, frame, question
+                self.config.vision_manager.processor.process_image, frame, question,
             )
 
             if isinstance(result, dict) and "error" in result:
@@ -287,10 +285,10 @@ async def _handle_camera(self, arguments: dict) -> List[types.TextContent]:
                         {
                             "image_description": result
                             if isinstance(result, str)
-                            else str(result)
-                        }
+                            else str(result),
+                        },
                     ),
-                )
+                ),
             ]
 
         except Exception as e:
@@ -299,9 +297,9 @@ async def _handle_camera(self, arguments: dict) -> List[types.TextContent]:
                 types.TextContent(
                     type="text",
                     text=json.dumps(
-                        {"error": f"Camera capture failed: {type(e).__name__}: {e}"}
+                        {"error": f"Camera capture failed: {type(e).__name__}: {e}"},
                     ),
-                )
+                ),
             ]
 
     async def run(self, transport_type: str = "stdio"):
@@ -330,8 +328,8 @@ async def run(self, transport_type: str = "stdio"):
 def create_reachy_mcp_server(
     reachy_mini: ReachyMini,
     movement_manager: MovementManager,
-    camera: Optional[cv2.VideoCapture] = None,
-    vision_manager: Optional[VisionManager] = None,
+    camera: cv2.VideoCapture | None = None,
+    vision_manager: VisionManager | None = None,
     **config_kwargs,
 ) -> ReachyMCPServer:
     """Factory function to create a configured Reachy MCP Server"""
@@ -380,7 +378,7 @@ async def async_main():
         help="Transport method (default: stdio)",
     )
     parser.add_argument(
-        "--camera-index", type=int, default=0, help="Camera index (default: 0)"
+        "--camera-index", type=int, default=0, help="Camera index (default: 0)",
     )
 
     args = parser.parse_args()
@@ -439,7 +437,7 @@ async def async_main():
         # Initialize movement manager
         logger.info("Initializing movement manager...")
         movement_manager = MovementManager(
-            current_robot=current_robot, head_tracker=head_tracker, camera=camera
+            current_robot=current_robot, head_tracker=head_tracker, camera=camera,
         )
 
         # Create MCP server