OCI MCP Server

""" Tools for managing OCI Storage resources. """ import logging from typing import Dict, List, Any, Optional import oci logger = logging.getLogger(__name__) def list_buckets(object_storage_client: oci.object_storage.ObjectStorageClient, compartment_id: str, namespace_name: str) -> List[Dict[str, Any]]: """ List all buckets in a compartment. Args: object_storage_client: OCI ObjectStorage client compartment_id: OCID of the compartment namespace_name: Object Storage namespace name Returns: List of buckets with their details """ try: buckets_response = oci.pagination.list_call_get_all_results( object_storage_client.list_buckets, namespace_name, compartment_id ) buckets = [] for bucket in buckets_response.data: buckets.append({ "name": bucket.name, "namespace": bucket.namespace, "compartment_id": bucket.compartment_id, "created_by": bucket.created_by, "time_created": str(bucket.time_created), "etag": bucket.etag, }) logger.info(f"Found {len(buckets)} buckets in compartment {compartment_id}") return buckets except Exception as e: logger.exception(f"Error listing buckets: {e}") raise def get_bucket(object_storage_client: oci.object_storage.ObjectStorageClient, namespace_name: str, bucket_name: str) -> Dict[str, Any]: """ Get details of a specific bucket. Args: object_storage_client: OCI ObjectStorage client namespace_name: Object Storage namespace name bucket_name: Name of the bucket Returns: Details of the bucket """ try: bucket = object_storage_client.get_bucket(namespace_name, bucket_name).data bucket_details = { "name": bucket.name, "namespace": bucket.namespace, "compartment_id": bucket.compartment_id, "created_by": bucket.created_by, "time_created": str(bucket.time_created), "etag": bucket.etag, "public_access_type": bucket.public_access_type, "storage_tier": bucket.storage_tier, "object_events_enabled": bucket.object_events_enabled, "versioning": bucket.versioning, "replication_enabled": bucket.replication_enabled, "is_read_only": bucket.is_read_only, "object_lifecycle_policy_etag": bucket.object_lifecycle_policy_etag, } logger.info(f"Retrieved details for bucket {bucket_name}") return bucket_details except Exception as e: logger.exception(f"Error getting bucket details: {e}") raise def list_volumes(block_storage_client: oci.core.BlockstorageClient, compartment_id: str) -> List[Dict[str, Any]]: """ List all block volumes in a compartment. Args: block_storage_client: OCI BlockStorage client compartment_id: OCID of the compartment Returns: List of volumes with their details """ try: volumes_response = oci.pagination.list_call_get_all_results( block_storage_client.list_volumes, compartment_id ) volumes = [] for volume in volumes_response.data: volumes.append({ "id": volume.id, "display_name": volume.display_name, "compartment_id": volume.compartment_id, "availability_domain": volume.availability_domain, "size_in_mbs": volume.size_in_mbs, "size_in_gbs": volume.size_in_gbs, "lifecycle_state": volume.lifecycle_state, "time_created": str(volume.time_created), "volume_group_id": volume.volume_group_id, "is_hydrated": volume.is_hydrated, "vpus_per_gb": volume.vpus_per_gb, "is_auto_tune_enabled": volume.is_auto_tune_enabled, "auto_tuned_vpus_per_gb": volume.auto_tuned_vpus_per_gb, }) logger.info(f"Found {len(volumes)} volumes in compartment {compartment_id}") return volumes except Exception as e: logger.exception(f"Error listing volumes: {e}") raise def get_volume(block_storage_client: oci.core.BlockstorageClient, volume_id: str) -> Dict[str, Any]: """ Get details of a specific volume. Args: block_storage_client: OCI BlockStorage client volume_id: OCID of the volume Returns: Details of the volume """ try: volume = block_storage_client.get_volume(volume_id).data volume_details = { "id": volume.id, "display_name": volume.display_name, "compartment_id": volume.compartment_id, "availability_domain": volume.availability_domain, "size_in_mbs": volume.size_in_mbs, "size_in_gbs": volume.size_in_gbs, "lifecycle_state": volume.lifecycle_state, "time_created": str(volume.time_created), "volume_group_id": volume.volume_group_id, "is_hydrated": volume.is_hydrated, "vpus_per_gb": volume.vpus_per_gb, "is_auto_tune_enabled": volume.is_auto_tune_enabled, "auto_tuned_vpus_per_gb": volume.auto_tuned_vpus_per_gb, "kms_key_id": volume.kms_key_id, "source_details": { "type": volume.source_details.type if volume.source_details else None, "id": volume.source_details.id if volume.source_details else None, } if volume.source_details else None, } logger.info(f"Retrieved details for volume {volume_id}") return volume_details except Exception as e: logger.exception(f"Error getting volume details: {e}") raise def list_boot_volumes(block_storage_client: oci.core.BlockstorageClient, availability_domain: str, compartment_id: str) -> List[Dict[str, Any]]: """ List all boot volumes in a compartment and availability domain. Args: block_storage_client: OCI BlockStorage client availability_domain: Availability domain name compartment_id: OCID of the compartment Returns: List of boot volumes with their details """ try: boot_volumes_response = oci.pagination.list_call_get_all_results( block_storage_client.list_boot_volumes, availability_domain, compartment_id ) boot_volumes = [] for boot_volume in boot_volumes_response.data: boot_volumes.append({ "id": boot_volume.id, "display_name": boot_volume.display_name, "compartment_id": boot_volume.compartment_id, "availability_domain": boot_volume.availability_domain, "size_in_mbs": boot_volume.size_in_mbs, "size_in_gbs": boot_volume.size_in_gbs, "lifecycle_state": boot_volume.lifecycle_state, "time_created": str(boot_volume.time_created), "is_hydrated": boot_volume.is_hydrated, "vpus_per_gb": boot_volume.vpus_per_gb, "is_auto_tune_enabled": boot_volume.is_auto_tune_enabled, "auto_tuned_vpus_per_gb": boot_volume.auto_tuned_vpus_per_gb, }) logger.info(f"Found {len(boot_volumes)} boot volumes in compartment {compartment_id}") return boot_volumes except Exception as e: logger.exception(f"Error listing boot volumes: {e}") raise def get_boot_volume(block_storage_client: oci.core.BlockstorageClient, boot_volume_id: str) -> Dict[str, Any]: """ Get details of a specific boot volume. Args: block_storage_client: OCI BlockStorage client boot_volume_id: OCID of the boot volume Returns: Details of the boot volume """ try: boot_volume = block_storage_client.get_boot_volume(boot_volume_id).data boot_volume_details = { "id": boot_volume.id, "display_name": boot_volume.display_name, "compartment_id": boot_volume.compartment_id, "availability_domain": boot_volume.availability_domain, "size_in_mbs": boot_volume.size_in_mbs, "size_in_gbs": boot_volume.size_in_gbs, "lifecycle_state": boot_volume.lifecycle_state, "time_created": str(boot_volume.time_created), "is_hydrated": boot_volume.is_hydrated, "vpus_per_gb": boot_volume.vpus_per_gb, "is_auto_tune_enabled": boot_volume.is_auto_tune_enabled, "auto_tuned_vpus_per_gb": boot_volume.auto_tuned_vpus_per_gb, "kms_key_id": boot_volume.kms_key_id, "source_details": { "type": boot_volume.source_details.type if boot_volume.source_details else None, "id": boot_volume.source_details.id if boot_volume.source_details else None, } if boot_volume.source_details else None, } logger.info(f"Retrieved details for boot volume {boot_volume_id}") return boot_volume_details except Exception as e: logger.exception(f"Error getting boot volume details: {e}") raise def list_file_systems(file_storage_client: oci.file_storage.FileStorageClient, compartment_id: str, availability_domain: str) -> List[Dict[str, Any]]: """ List all file systems in a compartment and availability domain. Args: file_storage_client: OCI FileStorage client compartment_id: OCID of the compartment availability_domain: Availability domain name Returns: List of file systems with their details """ try: file_systems_response = oci.pagination.list_call_get_all_results( file_storage_client.list_file_systems, compartment_id, availability_domain ) file_systems = [] for file_system in file_systems_response.data: file_systems.append({ "id": file_system.id, "display_name": file_system.display_name, "compartment_id": file_system.compartment_id, "availability_domain": file_system.availability_domain, "lifecycle_state": file_system.lifecycle_state, "time_created": str(file_system.time_created), "metered_bytes": file_system.metered_bytes, "is_clone_parent": file_system.is_clone_parent, "is_hydrated": file_system.is_hydrated, "lifecycle_details": file_system.lifecycle_details, "kms_key_id": file_system.kms_key_id, }) logger.info(f"Found {len(file_systems)} file systems in compartment {compartment_id}") return file_systems except Exception as e: logger.exception(f"Error listing file systems: {e}") raise def get_file_system(file_storage_client: oci.file_storage.FileStorageClient, file_system_id: str) -> Dict[str, Any]: """ Get details of a specific file system. Args: file_storage_client: OCI FileStorage client file_system_id: OCID of the file system Returns: Details of the file system """ try: file_system = file_storage_client.get_file_system(file_system_id).data file_system_details = { "id": file_system.id, "display_name": file_system.display_name, "compartment_id": file_system.compartment_id, "availability_domain": file_system.availability_domain, "lifecycle_state": file_system.lifecycle_state, "time_created": str(file_system.time_created), "metered_bytes": file_system.metered_bytes, "is_clone_parent": file_system.is_clone_parent, "is_hydrated": file_system.is_hydrated, "lifecycle_details": file_system.lifecycle_details, "kms_key_id": file_system.kms_key_id, "source_details": { "parent_file_system_id": file_system.source_details.parent_file_system_id if file_system.source_details else None, "source_snapshot_id": file_system.source_details.source_snapshot_id if file_system.source_details else None, } if file_system.source_details else None, } logger.info(f"Retrieved details for file system {file_system_id}") return file_system_details except Exception as e: logger.exception(f"Error getting file system details: {e}") raise