Modified get_action_status to work with action_name and not action_type

Author: stex2005

.gitignore

@@ -178,4 +178,7 @@ pyrightconfig.json
 .vs/
 
 # camera
-/camera/*
+camera/*
+
+# tests
+tests/

server.py

@@ -80,26 +80,26 @@ def list_verified_robot_specifications() -> dict:
     )
 )
 def connect_to_robot(
-    ip: Optional[str] = None,
-    port: Optional[Union[int, str]] = None,
+    ip: str = ROSBRIDGE_IP,
+    port: Union[int, str] = ROSBRIDGE_PORT,
     ping_timeout: float = 2.0,
     port_timeout: float = 2.0,
 ) -> dict:
     """
     Connect to a robot by setting the IP and port for the WebSocket connection, then testing connectivity.
 
     Args:
-        ip (Optional[str]): The IP address of the rosbridge server. Defaults to "127.0.0.1" (localhost).
-        port (Optional[int]): The port number of the rosbridge server. Defaults to 9090.
+        ip (str): The IP address of the rosbridge server. Defaults to "127.0.0.1" (localhost).
+        port (int): The port number of the rosbridge server. Defaults to 9090.
         ping_timeout (float): Timeout for ping in seconds. Default = 2.0.
         port_timeout (float): Timeout for port check in seconds. Default = 2.0.
 
     Returns:
         dict: Connection status with ping and port check results.
     """
     # Set default values if None
-    actual_ip = ip if ip is not None else "127.0.0.1"
-    actual_port = int(port) if port is not None else 9090
+    actual_ip = str(ip).strip() if ip else ROSBRIDGE_IP
+    actual_port = int(port) if port else ROSBRIDGE_PORT
 
     # Set the IP and port
     ws_manager.set_ip(actual_ip, actual_port)
@@ -2112,12 +2112,7 @@ def get_action_details(action_type: str) -> dict:
     if not action_type or not action_type.strip():
         return {"error": "Action type cannot be empty"}
 
-    result = {
-        "action_type": action_type, 
-        "goal": {}, 
-        "result": {}, 
-        "feedback": {}
-    }
+    result = {"action_type": action_type, "goal": {}, "result": {}, "feedback": {}}
 
     # Get goal, result, and feedback details in a single WebSocket context
     with ws_manager:
@@ -2131,7 +2126,12 @@ def get_action_details(action_type: str) -> dict:
         }
 
         goal_response = ws_manager.request(goal_message)
-        if goal_response and isinstance(goal_response, dict) and "values" in goal_response and "error" not in goal_response:
+        if (
+            goal_response
+            and isinstance(goal_response, dict)
+            and "values" in goal_response
+            and "error" not in goal_response
+        ):
             typedefs = goal_response["values"].get("typedefs", [])
             if typedefs:
                 for typedef in typedefs:
@@ -2141,24 +2141,24 @@ def get_action_details(action_type: str) -> dict:
                     examples = typedef.get("examples", [])
                     const_names = typedef.get("constnames", [])
                     const_values = typedef.get("constvalues", [])
-                    
+
                     fields = {}
                     field_details = {}
                     for i, (name, ftype) in enumerate(zip(field_names, field_types)):
                         fields[name] = ftype
                         field_details[name] = {
                             "type": ftype,
                             "array_length": field_array_len[i] if i < len(field_array_len) else -1,
-                            "example": examples[i] if i < len(examples) else None
+                            "example": examples[i] if i < len(examples) else None,
                         }
-                    
+
                     result["goal"] = {
-                        "fields": fields, 
+                        "fields": fields,
                         "field_count": len(fields),
                         "field_details": field_details,
                         "message_type": typedef.get("type", ""),
                         "examples": examples,
-                        "constants": dict(zip(const_names, const_values)) if const_names else {}
+                        "constants": dict(zip(const_names, const_values)) if const_names else {},
                     }
 
         # Get result details using action-specific service
@@ -2171,7 +2171,12 @@ def get_action_details(action_type: str) -> dict:
         }
 
         result_response = ws_manager.request(result_message)
-        if result_response and isinstance(result_response, dict) and "values" in result_response and "error" not in result_response:
+        if (
+            result_response
+            and isinstance(result_response, dict)
+            and "values" in result_response
+            and "error" not in result_response
+        ):
             typedefs = result_response["values"].get("typedefs", [])
             if typedefs:
                 for typedef in typedefs:
@@ -2181,24 +2186,24 @@ def get_action_details(action_type: str) -> dict:
                     examples = typedef.get("examples", [])
                     const_names = typedef.get("constnames", [])
                     const_values = typedef.get("constvalues", [])
-                    
+
                     fields = {}
                     field_details = {}
                     for i, (name, ftype) in enumerate(zip(field_names, field_types)):
                         fields[name] = ftype
                         field_details[name] = {
                             "type": ftype,
                             "array_length": field_array_len[i] if i < len(field_array_len) else -1,
-                            "example": examples[i] if i < len(examples) else None
+                            "example": examples[i] if i < len(examples) else None,
                         }
-                    
+
                     result["result"] = {
-                        "fields": fields, 
+                        "fields": fields,
                         "field_count": len(fields),
                         "field_details": field_details,
                         "message_type": typedef.get("type", ""),
                         "examples": examples,
-                        "constants": dict(zip(const_names, const_values)) if const_names else {}
+                        "constants": dict(zip(const_names, const_values)) if const_names else {},
                     }
 
         # Get feedback details using action-specific service
@@ -2211,7 +2216,12 @@ def get_action_details(action_type: str) -> dict:
         }
 
         feedback_response = ws_manager.request(feedback_message)
-        if feedback_response and isinstance(feedback_response, dict) and "values" in feedback_response and "error" not in feedback_response:
+        if (
+            feedback_response
+            and isinstance(feedback_response, dict)
+            and "values" in feedback_response
+            and "error" not in feedback_response
+        ):
             typedefs = feedback_response["values"].get("typedefs", [])
             if typedefs:
                 for typedef in typedefs:
@@ -2221,24 +2231,24 @@ def get_action_details(action_type: str) -> dict:
                     examples = typedef.get("examples", [])
                     const_names = typedef.get("constnames", [])
                     const_values = typedef.get("constvalues", [])
-                    
+
                     fields = {}
                     field_details = {}
                     for i, (name, ftype) in enumerate(zip(field_names, field_types)):
                         fields[name] = ftype
                         field_details[name] = {
                             "type": ftype,
                             "array_length": field_array_len[i] if i < len(field_array_len) else -1,
-                            "example": examples[i] if i < len(examples) else None
+                            "example": examples[i] if i < len(examples) else None,
                         }
-                    
+
                     result["feedback"] = {
-                        "fields": fields, 
+                        "fields": fields,
                         "field_count": len(fields),
                         "field_details": field_details,
                         "message_type": typedef.get("type", ""),
                         "examples": examples,
-                        "constants": dict(zip(const_names, const_values)) if const_names else {}
+                        "constants": dict(zip(const_names, const_values)) if const_names else {},
                     }
 
     # Check if we got any data
@@ -2250,33 +2260,33 @@ def get_action_details(action_type: str) -> dict:
 
 @mcp.tool(
     description=(
-        "Get action status for a specific action type. Works only with ROS 2.\n"
+        "Get action status for a specific action name. Works only with ROS 2.\n"
         "Example:\n"
-        "get_action_status('action_tutorials_interfaces/action/Fibonacci')"
+        "get_action_status('/fibonacci')"
     )
 )
-def get_action_status(action_type: str) -> dict:
+def get_action_status(action_name: str) -> dict:
     """
-    Get action status for a specific action type. Works only with ROS 2.
+    Get action status for a specific action name. Works only with ROS 2.
 
     Args:
-        action_type (str): The action type (e.g., 'action_tutorials_interfaces/action/Fibonacci')
+        action_name (str): The action name (e.g., '/fibonacci')
 
     Returns:
         dict: Contains action status information including active goals and their status.
     """
     # Validate input
-    if not action_type or not action_type.strip():
-        return {"error": "Action type cannot be empty"}
+    if not action_name or not action_name.strip():
+        return {"error": "Action name cannot be empty"}
+
+    # Ensure action name starts with /
+    if not action_name.startswith("/"):
+        action_name = f"/{action_name}"
 
-    # Extract action name from action type
-    # For example: 'action_tutorials_interfaces/action/Fibonacci' -> '/fibonacci'
-    action_name = f"/{action_type.split('/')[-1].lower()}"
-    
     # Try to get action status by subscribing to the status topic
     status_topic = f"{action_name}/_action/status"
     status_msg_type = "action_msgs/msg/GoalStatusArray"
-    
+
     try:
         # Subscribe to action status topic
         with ws_manager:
@@ -2286,75 +2296,79 @@ def get_action_status(action_type: str) -> dict:
                 "type": status_msg_type,
                 "id": f"get_action_status_{action_name.replace('/', '_')}",
             }
-            
+
             send_error = ws_manager.send(message)
             if send_error:
                 return {
-                    "action_type": action_type,
                     "action_name": action_name,
                     "success": False,
                     "error": f"Failed to subscribe to status topic: {send_error}",
                 }
-            
+
             # Wait for status message
             response = ws_manager.receive(timeout=3.0)
             if not response:
                 return {
-                    "action_type": action_type,
                     "action_name": action_name,
                     "success": False,
                     "error": "No response from action status topic",
                 }
-            
+
             response_data = json.loads(response)
-            
+
             if response_data.get("op") == "status" and response_data.get("level") == "error":
-                return {"error": f"Action status error: {response_data.get('msg', 'Unknown error')}"}
-            
+                return {
+                    "error": f"Action status error: {response_data.get('msg', 'Unknown error')}"
+                }
+
             if "msg" not in response_data or "status_list" not in response_data["msg"]:
                 return {
-                    "action_type": action_type,
                     "action_name": action_name,
                     "success": True,
                     "active_goals": [],
                     "goal_count": 0,
-                    "note": f"No active goals found for action {action_name}"
+                    "note": f"No active goals found for action {action_name}",
                 }
-            
+
             status_list = response_data["msg"]["status_list"]
             status_map = {
-                0: "STATUS_UNKNOWN", 1: "STATUS_ACCEPTED", 2: "STATUS_EXECUTING",
-                3: "STATUS_CANCELING", 4: "STATUS_SUCCEEDED", 5: "STATUS_CANCELED", 6: "STATUS_ABORTED"
+                0: "STATUS_UNKNOWN",
+                1: "STATUS_ACCEPTED",
+                2: "STATUS_EXECUTING",
+                3: "STATUS_CANCELING",
+                4: "STATUS_SUCCEEDED",
+                5: "STATUS_CANCELED",
+                6: "STATUS_ABORTED",
             }
-            
+
             active_goals = []
             for status_item in status_list:
                 goal_info = status_item.get("goal_info", {})
                 goal_id = goal_info.get("goal_id", {}).get("uuid", "unknown")
                 status = status_item.get("status", -1)
                 stamp = goal_info.get("stamp", {})
-                
-                active_goals.append({
-                    "goal_id": goal_id,
-                    "status": status,
-                    "status_text": status_map.get(status, "UNKNOWN"),
-                    "timestamp": f"{stamp.get('sec', 0)}.{stamp.get('nanosec', 0)}"
-                })
-            
+
+                active_goals.append(
+                    {
+                        "goal_id": goal_id,
+                        "status": status,
+                        "status_text": status_map.get(status, "UNKNOWN"),
+                        "timestamp": f"{stamp.get('sec', 0)}.{stamp.get('nanosec', 0)}",
+                    }
+                )
+
             return {
-                "action_type": action_type,
                 "action_name": action_name,
                 "success": True,
                 "active_goals": active_goals,
                 "goal_count": len(active_goals),
-                "note": f"Found {len(active_goals)} active goal(s) for action {action_name}"
+                "note": f"Found {len(active_goals)} active goal(s) for action {action_name}",
             }
-                
+
     except json.JSONDecodeError as e:
         return {"error": f"Failed to parse status response: {str(e)}"}
     except Exception as e:
         return {
-            "action_type": action_type,
             "action_name": action_name,
             "success": False,
             "error": f"Failed to get action status: {str(e)}",

utils/robot_specifications/local_rosbridge.yaml

@@ -0,0 +1,13 @@
+# Robot configuration template
+# Replace this text with specific instructions and information about your robot.
+# Include details on how to control it, important topics, and any special commands.
+# For example:
+# - Control commands (e.g., movement, actions)
+# - Sensor topics (e.g., camera, lidar)
+# - Safety guidelines
+# - Any custom messages or services used by the robot
+
+name: local_rosbridge
+alias: turtlesim # Optional alias for easier reference
+type: sim # or 'real' for simulation
+prompts: The local_rosbridge is a generic local turtlesim robot.