[CHORE] Add VNIC tools

Signed-off-by: will.shope <[email protected]>

Author: shopewf

src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/__init__.py

@@ -5,4 +5,4 @@
 """
 
 __project__ = "oracle.oci-compute-mcp-server"
-__version__ = "1.0.2"
+__version__ = "1.1.0"

src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/models.py

@@ -781,3 +781,104 @@ def map_response(resp: oci.response.Response) -> Response | None:
 
 
 # endregion
+
+# region VnicAttachment
+
+
+class VnicAttachment(BaseModel):
+    """
+    Pydantic model mirroring the fields of oci.core.models.VnicAttachment.
+    """
+
+    availability_domain: Optional[str] = Field(
+        None,
+        description="The availability domain of the instance. Example: `Uocm:PHX-AD-1`",
+    )
+    compartment_id: Optional[str] = Field(
+        None,
+        description=(
+            "The OCID of the compartment the VNIC attachment is in, which is the "
+            "same compartment the instance is in."
+        ),
+    )
+    display_name: Optional[str] = Field(
+        None,
+        description=(
+            "A user-friendly name. Does not have to be unique, and it's changeable. "
+            "Avoid entering confidential information."
+        ),
+    )
+    id: Optional[str] = Field(None, description="The OCID of the VNIC attachment.")
+    instance_id: Optional[str] = Field(None, description="The OCID of the instance.")
+    lifecycle_state: Optional[
+        Literal["ATTACHING", "ATTACHED", "DETACHING", "DETACHED", "UNKNOWN_ENUM_VALUE"]
+    ] = Field(None, description="The current state of the VNIC attachment.")
+    nic_index: Optional[int] = Field(
+        None,
+        description=(
+            "Which physical network interface card (NIC) the VNIC uses. Certain bare "
+            "metal instance shapes have two active physical NICs (0 and 1). If you add "
+            "a secondary VNIC to one of these instances, you can specify which NIC "
+            "the VNIC will use. For more information, see Virtual Network Interface "
+            "Cards (VNICs)."
+        ),
+    )
+    subnet_id: Optional[str] = Field(
+        None, description="The OCID of the subnet to create the VNIC in."
+    )
+    vlan_id: Optional[str] = Field(
+        None,
+        description=(
+            "The OCID of the VLAN to create the VNIC in. Creating the VNIC in a VLAN "
+            "(instead of a subnet) is possible only if you are an Oracle Cloud VMware "
+            "Solution customer. See Vlan. An error is returned if the instance already "
+            "has a VNIC attached to it from this VLAN."
+        ),
+    )
+    time_created: Optional[datetime] = Field(
+        None,
+        description=(
+            "The date and time the VNIC attachment was created, in the format defined "
+            "by RFC3339. Example: `2016-08-25T21:10:29.600Z`"
+        ),
+    )
+    vlan_tag: Optional[int] = Field(
+        None,
+        description=(
+            "The Oracle-assigned VLAN tag of the attached VNIC. Available after the "
+            "attachment process is complete. However, if the VNIC belongs to a VLAN "
+            "as part of the Oracle Cloud VMware Solution, the `vlanTag` value is "
+            "instead the value of the `vlanTag` attribute for the VLAN. See Vlan. "
+            "Example: `0`"
+        ),
+    )
+    vnic_id: Optional[str] = Field(
+        None,
+        description=(
+            "The OCID of the VNIC. Available after the attachment process is "
+            "complete."
+        ),
+    )
+
+
+def map_vnic_attachment(va: oci.core.models.VnicAttachment) -> VnicAttachment:
+    """
+    Convert an oci.core.models.VnicAttachment to oracle.oci_compute_mcp_server.models.VnicAttachment.
+    """
+    return VnicAttachment(
+        availability_domain=getattr(va, "availability_domain", None),
+        compartment_id=getattr(va, "compartment_id", None),
+        display_name=getattr(va, "display_name", None),
+        id=getattr(va, "id", None),
+        instance_id=getattr(va, "instance_id", None),
+        lifecycle_state=getattr(va, "lifecycle_state", None),
+        nic_index=getattr(va, "nic_index", None),
+        subnet_id=getattr(va, "subnet_id", None),
+        vlan_id=getattr(va, "vlan_id", None),
+        time_created=getattr(va, "time_created", None),
+        vlan_tag=getattr(va, "vlan_tag", None),
+        vnic_id=getattr(va, "vnic_id", None),
+    )
+
+
+# endregion

src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/server.py

@@ -20,9 +20,11 @@
     Image,
     Instance,
     Response,
+    VnicAttachment,
     map_image,
     map_instance,
     map_response,
+    map_vnic_attachment,
 )
 from pydantic import Field
 
@@ -88,7 +90,7 @@ def list_instances(
                 "limit": limit,
             }
 
-            if lifecycle_state is not None:
+            if lifecycle_state:
                 kwargs["lifecycle_state"] = lifecycle_state
 
             response = client.list_instances(**kwargs)
@@ -272,6 +274,11 @@ def list_images(
     operating_system: Optional[str] = Field(
         None, description="The operating system to filter with"
     ),
+    limit: Optional[int] = Field(
+        None,
+        description="The maximum amount of resources to return. If None, there is no limit.",
+        ge=1,
+    ),
 ) -> list[Image]:
     images: list[Image] = []
 
@@ -282,8 +289,14 @@ def list_images(
         has_next_page = True
         next_page: str = None
 
-        while has_next_page:
-            response = client.list_images(compartment_id=compartment_id, page=next_page)
+        while has_next_page and (limit is None or len(images) < limit):
+            kwargs = {
+                "compartment_id": compartment_id,
+                "page": next_page,
+                "limit": limit,
+            }
+
+            response = client.list_images(**kwargs)
             has_next_page = response.has_next_page
             next_page = response.next_page if hasattr(response, "next_page") else None
 
@@ -345,6 +358,78 @@ def instance_action(
         raise e
 
 
+@mcp.tool(
+    description="List vnic attachments in a given compartment and/or on a given instance. "
+)
+def list_vnic_attachments(
+    compartment_id: str = Field(
+        ...,
+        description="The OCID of the compartment. "
+        "If an instance_id is passed in, but no compartment_id is passed in,"
+        "then the compartment OCID of the instance may be used as a default.",
+    ),
+    instance_id: Optional[str] = Field(None, description="The OCID of the instance"),
+    limit: Optional[int] = Field(
+        None,
+        description="The maximum amount of resources to return. If None, there is no limit.",
+        ge=1,
+    ),
+) -> list[VnicAttachment]:
+    vnic_attachments: list[VnicAttachment] = []
+
+    try:
+        client = get_compute_client()
+
+        response: oci.response.Response = None
+        has_next_page = True
+        next_page: str = None
+
+        while has_next_page and (limit is None or len(vnic_attachments) < limit):
+            kwargs = {
+                "compartment_id": compartment_id,
+                "page": next_page,
+                "limit": limit,
+            }
+
+            if instance_id:
+                kwargs["instance_id"] = instance_id
+
+            response = client.list_vnic_attachments(**kwargs)
+            has_next_page = response.has_next_page
+            next_page = response.next_page if hasattr(response, "next_page") else None
+
+            data: list[oci.core.models.VnicAttachment] = response.data
+
+            for d in data:
+                vnic_attachments.append(map_vnic_attachment(d))
+
+        logger.info(f"Found {len(vnic_attachments)} Vnic Attachments")
+        return vnic_attachments
+
+    except Exception as e:
+        logger.error(f"Error in list_vnic_attachments tool: {str(e)}")
+        raise e
+
+
+@mcp.tool(description="Get Vnic Attachment with a given OCID")
+def get_vnic_attachment(
+    vnic_attachment_id: str = Field(..., description="The OCID of the vnic attachment")
+) -> VnicAttachment:
+    try:
+        client = get_compute_client()
+
+        response: oci.response.Response = client.get_vnic_attachment(
+            vnic_attachment_id=vnic_attachment_id
+        )
+        data: oci.core.models.VnicAttachment = response.data
+        logger.info("Found Vnic Attachment")
+        return map_vnic_attachment(data)
+
+    except Exception as e:
+        logger.error(f"Error in get_vnic_attachment tool: {str(e)}")
+        raise e
+
+
 def main() -> None:
     mcp.run()
 

src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_tools.py

@@ -227,3 +227,55 @@ async def test_instance_action(self, mock_get_client):
 
             assert result["id"] == "instance1"
             assert result["lifecycle_state"] == "STOPPING"
+
+    @pytest.mark.asyncio
+    @patch("oracle.oci_compute_mcp_server.server.get_compute_client")
+    async def test_list_vnic_attachments(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_get_client.return_value = mock_client
+
+        mock_list_response = create_autospec(oci.response.Response)
+        mock_list_response.data = [
+            oci.core.models.VnicAttachment(
+                id="vnicattachment1",
+                display_name="VNIC attachment 1",
+                lifecycle_state="ATTACHED",
+            )
+        ]
+        mock_list_response.has_next_page = False
+        mock_list_response.next_page = None
+        mock_client.list_vnic_attachments.return_value = mock_list_response
+
+        async with Client(mcp) as client:
+            call_tool_result = await client.call_tool(
+                "list_vnic_attachments",
+                {
+                    "compartment_id": "test_compartment",
+                },
+            )
+            result = call_tool_result.structured_content["result"]
+
+            assert len(result) == 1
+            assert result[0]["id"] == "vnicattachment1"
+
+    @pytest.mark.asyncio
+    @patch("oracle.oci_compute_mcp_server.server.get_compute_client")
+    async def test_get_vnic_attachment(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_get_client.return_value = mock_client
+
+        mock_get_response = create_autospec(oci.response.Response)
+        mock_get_response.data = oci.core.models.VnicAttachment(
+            id="vnicattachment1",
+            display_name="VNIC attachment 1",
+            lifecycle_state="ATTACHED",
+        )
+        mock_client.get_vnic_attachment.return_value = mock_get_response
+
+        async with Client(mcp) as client:
+            call_tool_result = await client.call_tool(
+                "get_vnic_attachment", {"vnic_attachment_id": "vnicattachment1"}
+            )
+            result = call_tool_result.structured_content
+
+            assert result["id"] == "vnicattachment1"

src/oci-compute-mcp-server/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oracle.oci-compute-mcp-server"
-version = "1.0.2"
+version = "1.1.0"
 description = "OCI Compute Service MCP server"
 readme = "README.md"
 requires-python = ">=3.13"