feat: volume attachments

src/openstack_mcp_server/tools/compute_tools.py

@@ -45,6 +45,8 @@ def register_tools(self, mcp: FastMCP):
         mcp.tool()(self.action_server)
         mcp.tool()(self.update_server)
         mcp.tool()(self.delete_server)
+        mcp.tool()(self.attach_volume)
+        mcp.tool()(self.detach_volume)
 
     def get_servers(self) -> list[Server]:
         """
@@ -125,7 +127,7 @@ def get_flavors(self) -> list[Flavor]:
             flavor_list.append(Flavor(**flavor))
         return flavor_list
 
-    def action_server(self, id: str, action: ServerActionEnum) -> None:
+    def action_server(self, id: str, action: str) -> None:
         """
         Perform an action on a Compute server.
 
@@ -151,19 +153,19 @@ def action_server(self, id: str, action: ServerActionEnum) -> None:
         conn = get_openstack_conn()
 
         action_methods = {
-            ServerActionEnum.PAUSE: conn.compute.pause_server,
-            ServerActionEnum.UNPAUSE: conn.compute.unpause_server,
-            ServerActionEnum.SUSPEND: conn.compute.suspend_server,
-            ServerActionEnum.RESUME: conn.compute.resume_server,
-            ServerActionEnum.LOCK: conn.compute.lock_server,
-            ServerActionEnum.UNLOCK: conn.compute.unlock_server,
-            ServerActionEnum.RESCUE: conn.compute.rescue_server,
-            ServerActionEnum.UNRESCUE: conn.compute.unrescue_server,
-            ServerActionEnum.START: conn.compute.start_server,
-            ServerActionEnum.STOP: conn.compute.stop_server,
-            ServerActionEnum.SHELVE: conn.compute.shelve_server,
-            ServerActionEnum.SHELVE_OFFLOAD: conn.compute.shelve_offload_server,
-            ServerActionEnum.UNSHELVE: conn.compute.unshelve_server,
+            ServerActionEnum.PAUSE.value: conn.compute.pause_server,
+            ServerActionEnum.UNPAUSE.value: conn.compute.unpause_server,
+            ServerActionEnum.SUSPEND.value: conn.compute.suspend_server,
+            ServerActionEnum.RESUME.value: conn.compute.resume_server,
+            ServerActionEnum.LOCK.value: conn.compute.lock_server,
+            ServerActionEnum.UNLOCK.value: conn.compute.unlock_server,
+            ServerActionEnum.RESCUE.value: conn.compute.rescue_server,
+            ServerActionEnum.UNRESCUE.value: conn.compute.unrescue_server,
+            ServerActionEnum.START.value: conn.compute.start_server,
+            ServerActionEnum.STOP.value: conn.compute.stop_server,
+            ServerActionEnum.SHELVE.value: conn.compute.shelve_server,
+            ServerActionEnum.SHELVE_OFFLOAD.value: conn.compute.shelve_offload_server,
+            ServerActionEnum.UNSHELVE.value: conn.compute.unshelve_server,
         }
 
         if action not in action_methods:
@@ -214,3 +216,28 @@ def delete_server(self, id: str) -> None:
         """
         conn = get_openstack_conn()
         conn.compute.delete_server(id)
+
+    def attach_volume(
+        self, server_id: str, volume_id: str, device: str | None = None
+    ) -> None:
+        """
+        Attach a volume to a Compute server.
+
+        :param server_id: The UUID of the server.
+        :param volume_id: The UUID of the volume to attach.
+        :param device: Name of the device such as, /dev/vdb. If you specify this parameter, the device must not exist in the guest operating system.
+        """
+        conn = get_openstack_conn()
+        conn.compute.create_volume_attachment(
+            server_id, volume_id=volume_id, device=device
+        )
+
+    def detach_volume(self, server_id: str, volume_id: str) -> None:
+        """
+        Detach a volume from a Compute server.
+
+        :param server_id: The UUID of the server.
+        :param volume_id: The UUID of the volume to detach.
+        """
+        conn = get_openstack_conn()
+        conn.compute.delete_volume_attachment(server_id, volume_id)

src/openstack_mcp_server/tools/response/compute.py

@@ -20,6 +20,10 @@ class IPAddress(BaseModel):
 
         model_config = ConfigDict(validate_by_name=True)
 
+    class VolumeAttachment(BaseModel):
+        id: str
+        delete_on_termination: bool
+
     class SecurityGroup(BaseModel):
         name: str
 
@@ -35,6 +39,7 @@ class SecurityGroup(BaseModel):
     security_groups: list[SecurityGroup] | None = None
     accessIPv4: str | None = None
     accessIPv6: str | None = None
+    attached_volumes: list[VolumeAttachment] | None = Field(default=None)
 
 
 class Flavor(BaseModel):

tests/tools/test_compute_tools.py

@@ -268,9 +268,11 @@ def test_register_tools(self):
                 call(compute_tools.action_server),
                 call(compute_tools.update_server),
                 call(compute_tools.delete_server),
+                call(compute_tools.attach_volume),
+                call(compute_tools.detach_volume),
             ],
         )
-        assert mock_tool_decorator.call_count == 7
+        assert mock_tool_decorator.call_count == 9
 
     def test_compute_tools_instantiation(self):
         """Test ComputeTools can be instantiated."""
@@ -555,3 +557,90 @@ def test_delete_server_not_found(self, mock_get_openstack_conn):
             compute_tools.delete_server(server_id)
 
         mock_conn.compute.delete_server.assert_called_once_with(server_id)
+
+    def test_attach_volume_success(self, mock_get_openstack_conn):
+        """Test attaching a volume to a server successfully."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "test-server-id"
+        volume_id = "test-volume-id"
+
+        mock_conn.compute.create_volume_attachment.return_value = None
+
+        compute_tools = ComputeTools()
+        result = compute_tools.attach_volume(server_id, volume_id)
+
+        assert result is None
+        mock_conn.compute.create_volume_attachment.assert_called_once_with(
+            server_id, volume_id=volume_id, device=None
+        )
+
+    def test_attach_volume_with_device(self, mock_get_openstack_conn):
+        """Test attaching a volume to a server with a specific device."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "test-server-id"
+        volume_id = "test-volume-id"
+        device = "/dev/vdb"
+
+        mock_conn.compute.create_volume_attachment.return_value = None
+
+        compute_tools = ComputeTools()
+        result = compute_tools.attach_volume(server_id, volume_id, device)
+
+        assert result is None
+        mock_conn.compute.create_volume_attachment.assert_called_once_with(
+            server_id, volume_id=volume_id, device=device
+        )
+
+    def test_attach_volume_exception(self, mock_get_openstack_conn):
+        """Test attaching a volume when exception occurs."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "test-server-id"
+        volume_id = "test-volume-id"
+
+        mock_conn.compute.create_volume_attachment.side_effect = (
+            NotFoundException()
+        )
+
+        compute_tools = ComputeTools()
+
+        with pytest.raises(NotFoundException):
+            compute_tools.attach_volume(server_id, volume_id)
+
+        mock_conn.compute.create_volume_attachment.assert_called_once_with(
+            server_id, volume_id=volume_id, device=None
+        )
+
+    def test_detach_volume_success(self, mock_get_openstack_conn):
+        """Test detaching a volume from a server successfully."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "test-server-id"
+        volume_id = "test-volume-id"
+
+        mock_conn.compute.delete_volume_attachment.return_value = None
+
+        compute_tools = ComputeTools()
+        result = compute_tools.detach_volume(server_id, volume_id)
+
+        assert result is None
+        mock_conn.compute.delete_volume_attachment.assert_called_once_with(
+            server_id, volume_id
+        )
+
+    def test_detach_volume_exception(self, mock_get_openstack_conn):
+        """Test detaching a volume when exception occurs."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "test-server-id"
+        volume_id = "test-volume-id"
+
+        mock_conn.compute.delete_volume_attachment.side_effect = (
+            NotFoundException()
+        )
+
+        compute_tools = ComputeTools()
+
+        with pytest.raises(NotFoundException):
+            compute_tools.detach_volume(server_id, volume_id)
+
+        mock_conn.compute.delete_volume_attachment.assert_called_once_with(
+            server_id, volume_id
+        )