chore(identity): Remove unnecessary variable(#31)

- Delete unnecessary variable in test code.

Author: S0okJu

.github/workflows/ci.yml

@@ -0,0 +1,56 @@
+name: CI
+
+on:
+  pull_request:
+    branches: [ develop ]
+  push:
+    branches: [ develop ]
+
+jobs:
+  lint-and-test:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v3
+
+    - name: Set up Python
+      run: uv python install
+
+    - name: Install dependencies
+      run: uv sync --all-groups
+
+    - name: Run ruff check
+      run: uv run ruff check
+
+    - name: Run ruff format check
+      run: uv run ruff format --check .
+
+    - name: Run tests with pytest
+      run: uv run pytest
+
+
+  test-matrix:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v3
+
+    - name: Set up Python ${{ matrix.python-version }}
+      run: uv python install ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      run: uv sync --all-groups
+
+    - name: Run tests
+      run: uv run pytest

src/openstack_mcp_server/tools/__init__.py

@@ -6,6 +6,7 @@ def register_tool(mcp: FastMCP):
     Register Openstack MCP tools.
     """
 
+    from .block_storage_tools import BlockStorageTools
     from .compute_tools import ComputeTools
     from .identity_tools import IdentityTools
     from .image_tools import ImageTools
@@ -15,3 +16,4 @@ def register_tool(mcp: FastMCP):
     ImageTools().register_tools(mcp)
     IdentityTools().register_tools(mcp)
     NetworkTools().register_tools(mcp)
+    BlockStorageTools().register_tools(mcp)

src/openstack_mcp_server/tools/block_storage_tools.py

@@ -0,0 +1,185 @@
+from fastmcp import FastMCP
+
+from .base import get_openstack_conn
+from .response.block_storage import (
+    Volume,
+    VolumeAttachment,
+)
+
+
+class BlockStorageTools:
+    """
+    A class to encapsulate Block Storage-related tools and utilities.
+    """
+
+    def register_tools(self, mcp: FastMCP):
+        """
+        Register Block Storage-related tools with the FastMCP instance.
+        """
+        mcp.tool()(self.get_volumes)
+        mcp.tool()(self.get_volume_details)
+        mcp.tool()(self.create_volume)
+        mcp.tool()(self.delete_volume)
+        mcp.tool()(self.extend_volume)
+
+    def get_volumes(self) -> list[Volume]:
+        """
+        Get the list of Block Storage volumes.
+
+        :return: A list of Volume objects representing the volumes.
+        """
+        conn = get_openstack_conn()
+
+        # List the volumes
+        volume_list = []
+        for volume in conn.block_storage.volumes():
+            attachments = []
+            for attachment in volume.attachments or []:
+                attachments.append(
+                    VolumeAttachment(
+                        server_id=attachment.get("server_id"),
+                        device=attachment.get("device"),
+                        attachment_id=attachment.get("id"),
+                    ),
+                )
+
+            volume_list.append(
+                Volume(
+                    id=volume.id,
+                    name=volume.name,
+                    status=volume.status,
+                    size=volume.size,
+                    volume_type=volume.volume_type,
+                    availability_zone=volume.availability_zone,
+                    created_at=str(volume.created_at)
+                    if volume.created_at
+                    else None,
+                    is_bootable=volume.is_bootable,
+                    is_encrypted=volume.is_encrypted,
+                    description=volume.description,
+                    attachments=attachments,
+                ),
+            )
+
+        return volume_list
+
+    def get_volume_details(self, volume_id: str) -> Volume:
+        """
+        Get detailed information about a specific volume.
+
+        :param volume_id: The ID of the volume to get details for
+        :return: A Volume object with detailed information
+        """
+        conn = get_openstack_conn()
+
+        volume = conn.block_storage.get_volume(volume_id)
+
+        attachments = []
+        for attachment in volume.attachments or []:
+            attachments.append(
+                VolumeAttachment(
+                    server_id=attachment.get("server_id"),
+                    device=attachment.get("device"),
+                    attachment_id=attachment.get("id"),
+                ),
+            )
+
+        return Volume(
+            id=volume.id,
+            name=volume.name,
+            status=volume.status,
+            size=volume.size,
+            volume_type=volume.volume_type,
+            availability_zone=volume.availability_zone,
+            created_at=str(volume.created_at),
+            is_bootable=volume.is_bootable,
+            is_encrypted=volume.is_encrypted,
+            description=volume.description,
+            attachments=attachments,
+        )
+
+    def create_volume(
+        self,
+        name: str,
+        size: int,
+        description: str | None = None,
+        volume_type: str | None = None,
+        availability_zone: str | None = None,
+        bootable: bool | None = None,
+        image: str | None = None,
+    ) -> Volume:
+        """
+        Create a new volume.
+
+        :param name: Name for the new volume
+        :param size: Size of the volume in GB
+        :param description: Optional description for the volume
+        :param volume_type: Optional volume type
+        :param availability_zone: Optional availability zone
+        :param bootable: Optional flag to make the volume bootable
+        :param image: Optional Image name, ID or object from which to create
+        :return: The created Volume object
+        """
+        conn = get_openstack_conn()
+
+        volume_kwargs = {
+            "name": name,
+        }
+
+        if description is not None:
+            volume_kwargs["description"] = description
+        if volume_type is not None:
+            volume_kwargs["volume_type"] = volume_type
+        if availability_zone is not None:
+            volume_kwargs["availability_zone"] = availability_zone
+
+        volume = conn.block_storage.create_volume(
+            size=size,
+            image=image,
+            bootable=bootable,
+            **volume_kwargs,
+        )
+
+        volume_obj = Volume(
+            id=volume.id,
+            name=volume.name,
+            status=volume.status,
+            size=volume.size,
+            volume_type=volume.volume_type,
+            availability_zone=volume.availability_zone,
+            created_at=str(volume.created_at),
+            is_bootable=volume.is_bootable,
+            is_encrypted=volume.is_encrypted,
+            description=volume.description,
+            attachments=[],
+        )
+
+        return volume_obj
+
+    def delete_volume(self, volume_id: str, force: bool = False) -> None:
+        """
+        Delete a volume.
+
+        :param volume_id: The ID of the volume to delete
+        :param force: Whether to force delete the volume
+        :return: None
+        """
+        conn = get_openstack_conn()
+
+        conn.block_storage.delete_volume(
+            volume_id,
+            force=force,
+            ignore_missing=False,
+        )
+
+    def extend_volume(self, volume_id: str, new_size: int) -> None:
+        """
+        Extend a volume to a new size.
+
+        :param volume_id: The ID of the volume to extend
+        :param new_size: The new size in GB (must be larger than current size)
+        :return: None
+        """
+        conn = get_openstack_conn()
+
+        conn.block_storage.extend_volume(volume_id, new_size)

src/openstack_mcp_server/tools/compute_tools.py

@@ -1,6 +1,11 @@
+from typing import Any
+
 from fastmcp import FastMCP
 
-from openstack_mcp_server.tools.response.compute import Server
+from openstack_mcp_server.tools.response.compute import (
+    Flavor,
+    Server,
+)
 
 from .base import get_openstack_conn
 
@@ -14,38 +19,86 @@ def register_tools(self, mcp: FastMCP):
         """
         Register Compute-related tools with the FastMCP instance.
         """
+        mcp.tool()(self.get_servers)
+        mcp.tool()(self.get_server)
+        mcp.tool()(self.create_server)
+        mcp.tool()(self.get_flavors)
 
-        mcp.tool()(self.get_compute_servers)
-        mcp.tool()(self.get_compute_server)
-
-    def get_compute_servers(self) -> list[Server]:
+    def get_servers(self) -> list[Server]:
         """
-        Get the list of Compute servers by invoking the registered tool.
+        Get the list of Compute servers.
 
-        :return: A list of Server objects representing the Compute servers.
+        :return: A list of Server objects.
         """
-        # Initialize connection
         conn = get_openstack_conn()
-
-        # List the servers
         server_list = []
         for server in conn.compute.servers():
-            server_list.append(
-                Server(name=server.name, id=server.id, status=server.status),
-            )
+            server_list.append(Server(**server))
 
         return server_list
 
-    def get_compute_server(self, id: str) -> Server:
+    def get_server(self, id: str) -> Server:
         """
-        Get a specific Compute server by invoking the registered tool.
+        Get a specific Compute server.
 
         :param id: The ID of the server to retrieve.
-        :return: A Server object representing the Compute server.
+        :return: A Server object.
         """
-        # Initialize connection
         conn = get_openstack_conn()
-
-        # Get a specific server (for example, the first one)
         server = conn.compute.get_server(id)
-        return Server(name=server.name, id=server.id, status=server.status)
+        return Server(**server)
+
+    def create_server(
+        self,
+        name: str,
+        image: str,
+        flavor: int,
+        network: str,
+        key_name: str | None = None,
+        security_groups: list[str] | None = None,
+        user_data: str | None = None,
+    ) -> Server:
+        """
+        Create a new Compute server.
+
+        :param name: The name of the server.
+        :param image: The ID of the image to use.
+        :param flavor: The ID of the flavor to use.
+        :param network: The ID of the network to attach.
+        :param key_name: The name of the key pair to use.
+        :param security_groups: A list of security group names to attach.
+        :param user_data: User data to pass to the server.
+        :return: A Server object
+        """
+        conn = get_openstack_conn()
+        server_params: dict[str, Any] = {
+            "name": name,
+            "flavorRef": flavor,
+            "imageRef": image,
+            "networks": [{"uuid": network}],
+            "key_name": key_name,
+            "security_groups": security_groups,
+            "user_data": user_data,
+        }
+        server_params = {
+            k: v for k, v in server_params.items() if v is not None
+        }
+
+        resp = conn.compute.create_server(**server_params)
+        # NOTE: The create_server method returns a server object with minimal information.
+        # To get the full server details, we need to fetch it again.
+        server = conn.compute.get_server(resp.id)
+
+        return Server(**server)
+
+    def get_flavors(self) -> list[Flavor]:
+        """
+        Get flavors (server hardware configurations).
+
+        :return: A list of Flavor objects.
+        """
+        conn = get_openstack_conn()
+        flavor_list = []
+        for flavor in conn.compute.flavors():
+            flavor_list.append(Flavor(**flavor))
+        return flavor_list