Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -5,4 +5,4 @@
"""

__project__ = "oracle.oci-compute-mcp-server"
__version__ = "1.0.2"
__version__ = "1.1.0"
101 changes: 101 additions & 0 deletions src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
91 changes: 88 additions & 3 deletions src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,11 @@
Image,
Instance,
Response,
VnicAttachment,
map_image,
map_instance,
map_response,
map_vnic_attachment,
)
from pydantic import Field

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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] = []

Expand All @@ -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

Expand Down Expand Up @@ -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()

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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"
2 changes: 1 addition & 1 deletion src/oci-compute-mcp-server/pyproject.toml
Original file line number Diff line number Diff line change
@@ -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"
Expand Down
Loading