OCI MCP Server

""" Tools for managing OCI Cost Management and Usage resources. """ import logging from typing import Dict, List, Any, Optional from datetime import datetime, timedelta import oci logger = logging.getLogger(__name__) def get_cost_usage_summary(usage_api_client: oci.usage_api.UsageapiClient, tenant_id: str, time_usage_started: str, time_usage_ended: str, granularity: str = "DAILY") -> List[Dict[str, Any]]: """ Get cost and usage summary for a tenancy. Args: usage_api_client: OCI UsageApi client tenant_id: OCID of the tenancy time_usage_started: Start time in ISO format (YYYY-MM-DD) time_usage_ended: End time in ISO format (YYYY-MM-DD) granularity: Granularity of the data (DAILY, MONTHLY) Returns: List of cost and usage summaries """ try: request_summarized_usages_details = oci.usage_api.models.RequestSummarizedUsagesDetails( tenant_id=tenant_id, time_usage_started=time_usage_started, time_usage_ended=time_usage_ended, granularity=granularity, is_aggregate_by_time=True ) usage_response = oci.pagination.list_call_get_all_results( usage_api_client.request_summarized_usages, request_summarized_usages_details=request_summarized_usages_details ) summaries = [] for item in usage_response.data.items: summaries.append({ "time_usage_started": str(item.time_usage_started), "time_usage_ended": str(item.time_usage_ended), "computed_amount": item.computed_amount, "computed_quantity": item.computed_quantity, "currency": item.currency, "service": item.service, "resource_name": item.resource_name, "compartment_name": item.compartment_name, "compartment_id": item.compartment_id, "unit": item.unit, }) logger.info(f"Retrieved {len(summaries)} usage summaries for tenancy {tenant_id}") return summaries except Exception as e: logger.exception(f"Error getting cost usage summary: {e}") raise def get_cost_by_service(usage_api_client: oci.usage_api.UsageapiClient, tenant_id: str, time_usage_started: str, time_usage_ended: str) -> List[Dict[str, Any]]: """ Get cost breakdown by service. Args: usage_api_client: OCI UsageApi client tenant_id: OCID of the tenancy time_usage_started: Start time in ISO format (YYYY-MM-DD) time_usage_ended: End time in ISO format (YYYY-MM-DD) Returns: List of costs grouped by service """ try: request_summarized_usages_details = oci.usage_api.models.RequestSummarizedUsagesDetails( tenant_id=tenant_id, time_usage_started=time_usage_started, time_usage_ended=time_usage_ended, granularity="DAILY", group_by=["service"] ) usage_response = oci.pagination.list_call_get_all_results( usage_api_client.request_summarized_usages, request_summarized_usages_details=request_summarized_usages_details ) # Aggregate by service service_costs = {} for item in usage_response.data.items: service = item.service if service not in service_costs: service_costs[service] = { "service": service, "total_cost": 0.0, "currency": item.currency, "unit": item.unit, } service_costs[service]["total_cost"] += float(item.computed_amount) if item.computed_amount else 0.0 result = list(service_costs.values()) logger.info(f"Retrieved cost breakdown for {len(result)} services") return result except Exception as e: logger.exception(f"Error getting cost by service: {e}") raise def get_cost_by_compartment(usage_api_client: oci.usage_api.UsageapiClient, tenant_id: str, time_usage_started: str, time_usage_ended: str) -> List[Dict[str, Any]]: """ Get cost breakdown by compartment. Args: usage_api_client: OCI UsageApi client tenant_id: OCID of the tenancy time_usage_started: Start time in ISO format (YYYY-MM-DD) time_usage_ended: End time in ISO format (YYYY-MM-DD) Returns: List of costs grouped by compartment """ try: request_summarized_usages_details = oci.usage_api.models.RequestSummarizedUsagesDetails( tenant_id=tenant_id, time_usage_started=time_usage_started, time_usage_ended=time_usage_ended, granularity="DAILY", group_by=["compartmentName"] ) usage_response = oci.pagination.list_call_get_all_results( usage_api_client.request_summarized_usages, request_summarized_usages_details=request_summarized_usages_details ) # Aggregate by compartment compartment_costs = {} for item in usage_response.data.items: compartment = item.compartment_name if compartment not in compartment_costs: compartment_costs[compartment] = { "compartment_name": compartment, "compartment_id": item.compartment_id, "total_cost": 0.0, "currency": item.currency, } compartment_costs[compartment]["total_cost"] += float(item.computed_amount) if item.computed_amount else 0.0 result = list(compartment_costs.values()) logger.info(f"Retrieved cost breakdown for {len(result)} compartments") return result except Exception as e: logger.exception(f"Error getting cost by compartment: {e}") raise def list_budgets(budget_client: oci.budget.BudgetClient, compartment_id: str) -> List[Dict[str, Any]]: """ List all budgets in a compartment. Args: budget_client: OCI Budget client compartment_id: OCID of the compartment Returns: List of budgets with their details """ try: budgets_response = oci.pagination.list_call_get_all_results( budget_client.list_budgets, compartment_id ) budgets = [] for budget in budgets_response.data: budgets.append({ "id": budget.id, "display_name": budget.display_name, "compartment_id": budget.compartment_id, "target_compartment_id": budget.target_compartment_id, "amount": budget.amount, "reset_period": budget.reset_period, "lifecycle_state": budget.lifecycle_state, "alert_rule_count": budget.alert_rule_count, "time_created": str(budget.time_created), "actual_spend": budget.actual_spend, "forecasted_spend": budget.forecasted_spend, "time_spend_computed": str(budget.time_spend_computed) if budget.time_spend_computed else None, }) logger.info(f"Found {len(budgets)} budgets in compartment {compartment_id}") return budgets except Exception as e: logger.exception(f"Error listing budgets: {e}") raise def get_budget(budget_client: oci.budget.BudgetClient, budget_id: str) -> Dict[str, Any]: """ Get details of a specific budget. Args: budget_client: OCI Budget client budget_id: OCID of the budget Returns: Details of the budget """ try: budget = budget_client.get_budget(budget_id).data budget_details = { "id": budget.id, "display_name": budget.display_name, "description": budget.description, "compartment_id": budget.compartment_id, "target_compartment_id": budget.target_compartment_id, "target_type": budget.target_type, "targets": budget.targets, "amount": budget.amount, "reset_period": budget.reset_period, "budget_processing_period_start_offset": budget.budget_processing_period_start_offset, "processing_period_type": budget.processing_period_type, "lifecycle_state": budget.lifecycle_state, "alert_rule_count": budget.alert_rule_count, "version": budget.version, "actual_spend": budget.actual_spend, "forecasted_spend": budget.forecasted_spend, "time_spend_computed": str(budget.time_spend_computed) if budget.time_spend_computed else None, "time_created": str(budget.time_created), "time_updated": str(budget.time_updated), } logger.info(f"Retrieved details for budget {budget_id}") return budget_details except Exception as e: logger.exception(f"Error getting budget details: {e}") raise