OCI MCP Server

Instructions

Implementation Reference

• mcp_server_oci/mcp_server.py:475-482 (handler)

  The MCP tool handler function `mcp_list_db_systems` that executes the tool logic by calling the OCI database client via the imported helper function.

  @mcp.tool(name="list_db_systems")
  @mcp_tool_wrapper(
      start_msg="Listing DB Systems in compartment {compartment_id}...",
      error_prefix="Error listing DB Systems"
  )
  async def mcp_list_db_systems(ctx: Context, compartment_id: str) -> List[Dict[str, Any]]:
      """List DB Systems in a compartment."""
      return list_db_systems(oci_clients["database"], compartment_id)

• mcp_server_oci/mcp_server.py:475-475 (registration)

  MCP tool registration decorator specifying the tool name "list_db_systems".

  @mcp.tool(name="list_db_systems")

• mcp_server_oci/tools/dbsystems.py:13-42 (helper)

  Supporting helper function `list_db_systems` imported and called by the handler, which wraps the OCI SDK `list_db_systems` call with pagination and response formatting.

  def list_db_systems(database_client: oci.database.DatabaseClient, compartment_id: str) -> List[Dict[str, Any]]:
      """List DB Systems in a compartment."""
      try:
          resp = oci.pagination.list_call_get_all_results(
              database_client.list_db_systems,
              compartment_id=compartment_id,
          )
  
          items = []
          for d in resp.data:
              items.append({
                  "id": d.id,
                  "display_name": d.display_name,
                  "lifecycle_state": d.lifecycle_state,
                  "shape": d.shape,
                  "database_edition": getattr(d, "database_edition", None),
                  "availability_domain": getattr(d, "availability_domain", None),
                  "time_created": str(getattr(d, "time_created", "")),
                  "subnet_id": getattr(d, "subnet_id", None),
                  "compartment_id": d.compartment_id,
                  "node_count": getattr(d, "node_count", None),
                  "version": getattr(d, "version", None),
                  "cpu_core_count": getattr(d, "cpu_core_count", None),
                  "data_storage_size_in_gb": getattr(d, "data_storage_size_in_gb", None),
              })
          logger.info(f"Found {len(items)} DB Systems in compartment {compartment_id}")
          return items
      except Exception as e:
          logger.exception(f"Error listing DB Systems: {e}")
          raise