OCI MCP Server

""" Tools for managing OCI network resources. """ import logging from typing import Dict, List, Any, Optional import oci logger = logging.getLogger(__name__) def list_vcns(network_client: oci.core.VirtualNetworkClient, compartment_id: str) -> List[Dict[str, Any]]: """ List all Virtual Cloud Networks (VCNs) in a compartment. Args: network_client: OCI VirtualNetwork client compartment_id: OCID of the compartment Returns: List of VCNs with their details """ try: # List all VCNs in the compartment vcns_response = oci.pagination.list_call_get_all_results( network_client.list_vcns, compartment_id, ) # Format the VCNs vcns = [] for vcn in vcns_response.data: vcns.append({ "id": vcn.id, "name": vcn.display_name, "lifecycle_state": vcn.lifecycle_state, "cidr_block": vcn.cidr_block, "time_created": str(vcn.time_created), "compartment_id": vcn.compartment_id, "dns_label": vcn.dns_label, "default_dhcp_options_id": vcn.default_dhcp_options_id, "default_route_table_id": vcn.default_route_table_id, "default_security_list_id": vcn.default_security_list_id, }) logger.info(f"Found {len(vcns)} VCNs in compartment {compartment_id}") return vcns except Exception as e: logger.exception(f"Error listing VCNs: {e}") raise def get_vcn(network_client: oci.core.VirtualNetworkClient, vcn_id: str) -> Dict[str, Any]: """ Get details of a specific VCN. Args: network_client: OCI VirtualNetwork client vcn_id: OCID of the VCN Returns: Details of the VCN """ try: # Get the VCN details vcn = network_client.get_vcn(vcn_id).data # Format the VCN details vcn_details = { "id": vcn.id, "name": vcn.display_name, "lifecycle_state": vcn.lifecycle_state, "cidr_block": vcn.cidr_block, "time_created": str(vcn.time_created), "compartment_id": vcn.compartment_id, "dns_label": vcn.dns_label, "default_dhcp_options_id": vcn.default_dhcp_options_id, "default_route_table_id": vcn.default_route_table_id, "default_security_list_id": vcn.default_security_list_id, } # Add IPv6 CIDR blocks if available if hasattr(vcn, 'ipv6_cidr_blocks') and vcn.ipv6_cidr_blocks: vcn_details["ipv6_cidr_blocks"] = vcn.ipv6_cidr_blocks logger.info(f"Retrieved details for VCN {vcn_id}") return vcn_details except Exception as e: logger.exception(f"Error getting VCN details: {e}") raise def list_subnets(network_client: oci.core.VirtualNetworkClient, compartment_id: str, vcn_id: Optional[str] = None) -> List[Dict[str, Any]]: """ List all subnets in a compartment, optionally filtered by VCN. Args: network_client: OCI VirtualNetwork client compartment_id: OCID of the compartment vcn_id: Optional OCID of the VCN to filter by Returns: List of subnets with their details """ try: # List all subnets in the compartment, optionally filtered by VCN if vcn_id: subnets_response = oci.pagination.list_call_get_all_results( network_client.list_subnets, compartment_id, vcn_id=vcn_id ) logger.info(f"Listing subnets in compartment {compartment_id} and VCN {vcn_id}") else: # If no VCN ID provided, we need to list all subnets in all VCNs vcns = list_vcns(network_client, compartment_id) subnets_response_data = [] for vcn in vcns: vcn_subnets = oci.pagination.list_call_get_all_results( network_client.list_subnets, compartment_id, vcn_id=vcn["id"] ).data subnets_response_data.extend(vcn_subnets) # Create a custom response object to match the structure returned by list_call_get_all_results class CustomResponse: def __init__(self, data): self.data = data subnets_response = CustomResponse(subnets_response_data) logger.info(f"Listing all subnets in compartment {compartment_id} across all VCNs") # Format the subnets subnets = [] for subnet in subnets_response.data: subnet_details = { "id": subnet.id, "name": subnet.display_name, "lifecycle_state": subnet.lifecycle_state, "cidr_block": subnet.cidr_block, "availability_domain": subnet.availability_domain, "compartment_id": subnet.compartment_id, "vcn_id": subnet.vcn_id, "route_table_id": subnet.route_table_id, "dhcp_options_id": subnet.dhcp_options_id, "security_list_ids": subnet.security_list_ids, "time_created": str(subnet.time_created), "prohibit_public_ip_on_vnic": subnet.prohibit_public_ip_on_vnic, } # Add IPv6 CIDR block if available if hasattr(subnet, 'ipv6_cidr_block') and subnet.ipv6_cidr_block: subnet_details["ipv6_cidr_block"] = subnet.ipv6_cidr_block subnets.append(subnet_details) logger.info(f"Found {len(subnets)} subnets") return subnets except Exception as e: logger.exception(f"Error listing subnets: {e}") raise def get_subnet(network_client: oci.core.VirtualNetworkClient, subnet_id: str) -> Dict[str, Any]: """ Get details of a specific subnet. Args: network_client: OCI VirtualNetwork client subnet_id: OCID of the subnet Returns: Details of the subnet """ try: # Get the subnet details subnet = network_client.get_subnet(subnet_id).data # Format the subnet details subnet_details = { "id": subnet.id, "name": subnet.display_name, "lifecycle_state": subnet.lifecycle_state, "cidr_block": subnet.cidr_block, "availability_domain": subnet.availability_domain, "compartment_id": subnet.compartment_id, "vcn_id": subnet.vcn_id, "route_table_id": subnet.route_table_id, "dhcp_options_id": subnet.dhcp_options_id, "security_list_ids": subnet.security_list_ids, "time_created": str(subnet.time_created), "prohibit_public_ip_on_vnic": subnet.prohibit_public_ip_on_vnic, } # Add IPv6 CIDR block if available if hasattr(subnet, 'ipv6_cidr_block') and subnet.ipv6_cidr_block: subnet_details["ipv6_cidr_block"] = subnet.ipv6_cidr_block logger.info(f"Retrieved details for subnet {subnet_id}") return subnet_details except Exception as e: logger.exception(f"Error getting subnet details: {e}") raise def list_vnics(compute_client: oci.core.ComputeClient, network_client: oci.core.VirtualNetworkClient, compartment_id: str, instance_id: Optional[str] = None) -> List[Dict[str, Any]]: """ List all VNICs in a compartment, optionally filtered by instance. Args: compute_client: OCI Compute client network_client: OCI VirtualNetwork client compartment_id: OCID of the compartment instance_id: Optional OCID of the instance to filter by Returns: List of VNICs with their details """ try: # Get VNIC attachments if instance_id: vnic_attachments = oci.pagination.list_call_get_all_results( compute_client.list_vnic_attachments, compartment_id, instance_id=instance_id ).data logger.info(f"Listing VNICs for instance {instance_id} in compartment {compartment_id}") else: vnic_attachments = oci.pagination.list_call_get_all_results( compute_client.list_vnic_attachments, compartment_id ).data logger.info(f"Listing all VNICs in compartment {compartment_id}") # Get VNIC details vnics = [] for attachment in vnic_attachments: try: vnic = network_client.get_vnic(attachment.vnic_id).data vnic_details = { "id": vnic.id, "display_name": vnic.display_name, "hostname_label": vnic.hostname_label, "is_primary": vnic.is_primary, "lifecycle_state": vnic.lifecycle_state, "mac_address": vnic.mac_address, "private_ip": vnic.private_ip, "public_ip": vnic.public_ip, "subnet_id": vnic.subnet_id, "time_created": str(vnic.time_created), "compartment_id": vnic.compartment_id, "attachment_id": attachment.id, "instance_id": attachment.instance_id, "attachment_lifecycle_state": attachment.lifecycle_state, } # Add IPv6 addresses if available if hasattr(vnic, 'ipv6_addresses') and vnic.ipv6_addresses: vnic_details["ipv6_addresses"] = vnic.ipv6_addresses vnics.append(vnic_details) except Exception as vnic_error: logger.warning(f"Error getting VNIC details for attachment {attachment.id}: {vnic_error}") # Add basic info from attachment vnics.append({ "attachment_id": attachment.id, "vnic_id": attachment.vnic_id, "instance_id": attachment.instance_id, "lifecycle_state": attachment.lifecycle_state, "time_created": str(attachment.time_created), "compartment_id": attachment.compartment_id, "error": str(vnic_error) }) logger.info(f"Found {len(vnics)} VNICs") return vnics except Exception as e: logger.exception(f"Error listing VNICs: {e}") raise def get_vnic(network_client: oci.core.VirtualNetworkClient, vnic_id: str) -> Dict[str, Any]: """ Get details of a specific VNIC. Args: network_client: OCI VirtualNetwork client vnic_id: OCID of the VNIC Returns: Details of the VNIC """ try: # Get the VNIC details vnic = network_client.get_vnic(vnic_id).data # Format the VNIC details vnic_details = { "id": vnic.id, "display_name": vnic.display_name, "hostname_label": vnic.hostname_label, "is_primary": vnic.is_primary, "lifecycle_state": vnic.lifecycle_state, "mac_address": vnic.mac_address, "private_ip": vnic.private_ip, "public_ip": vnic.public_ip, "subnet_id": vnic.subnet_id, "time_created": str(vnic.time_created), "compartment_id": vnic.compartment_id, } # Add IPv6 addresses if available if hasattr(vnic, 'ipv6_addresses') and vnic.ipv6_addresses: vnic_details["ipv6_addresses"] = vnic.ipv6_addresses logger.info(f"Retrieved details for VNIC {vnic_id}") return vnic_details except Exception as e: logger.exception(f"Error getting VNIC details: {e}") raise