Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -194,3 +194,6 @@ cython_debug/
# refer to https://docs.cursor.com/context/ignore-files
.cursorignore
.cursorindexingignore

# Openstack Config Files
clouds.yaml
3 changes: 3 additions & 0 deletions src/openstack_mcp_server/tools/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from fastmcp import FastMCP

from openstack_mcp_server.tools.connection import ConnectionManager


def register_tool(mcp: FastMCP):
"""
Expand All @@ -17,3 +19,4 @@ def register_tool(mcp: FastMCP):
IdentityTools().register_tools(mcp)
NetworkTools().register_tools(mcp)
BlockStorageTools().register_tools(mcp)
ConnectionManager().register_tools(mcp)
27 changes: 8 additions & 19 deletions src/openstack_mcp_server/tools/base.py
Original file line number Diff line number Diff line change
@@ -1,27 +1,16 @@
import openstack
from openstack_mcp_server.tools.connection import ConnectionManager

from openstack import connection

from openstack_mcp_server import config
_connection_manager = ConnectionManager()


class OpenStackConnectionManager:
"""OpenStack Connection Manager"""

_connection: connection.Connection | None = None

@classmethod
def get_connection(cls) -> connection.Connection:
"""OpenStack Connection"""
if cls._connection is None:
openstack.enable_logging(debug=config.MCP_DEBUG_MODE)
cls._connection = openstack.connect(cloud=config.MCP_CLOUD_NAME)
return cls._connection
def get_openstack_conn():
return _connection_manager.get_connection()


_openstack_connection_manager = OpenStackConnectionManager()
def set_openstack_cloud_name(cloud_name: str) -> None:
_connection_manager.set_cloud_name(cloud_name)


def get_openstack_conn():
"""Get OpenStack Connection"""
return _openstack_connection_manager.get_connection()
def get_openstack_cloud_name() -> str:
return _connection_manager.get_cloud_name()
73 changes: 73 additions & 0 deletions src/openstack_mcp_server/tools/connection.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
import openstack

from fastmcp import FastMCP
from openstack import connection
from openstack.config.loader import OpenStackConfig

from openstack_mcp_server import config


class ConnectionManager:
_cloud_name = config.MCP_CLOUD_NAME

def __init__(self):
openstack.enable_logging(debug=config.MCP_DEBUG_MODE)

def register_tools(self, mcp: FastMCP):
mcp.tool(self.get_cloud_config)
mcp.tool(self.get_cloud_names)
mcp.tool(self.get_cloud_name)
mcp.tool(self.set_cloud_name)

def get_connection(self) -> connection.Connection:
return openstack.connect(cloud=self._cloud_name)

def get_cloud_names(self) -> list[str]:
"""List available cloud configurations.

:return: Names of OpenStack clouds from user's config file.
"""
config = OpenStackConfig()
return list(config.get_cloud_names())

def get_cloud_config(self) -> dict:
"""Provide cloud configuration with secrets masked of current user's config file.

:return: Cloud configuration dictionary with credentials masked.
"""
config = OpenStackConfig()
return ConnectionManager._mask_credential(
config.cloud_config, ["password"]
)

@staticmethod
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

staticmethod와 classmethod를 사용하신 이유가 궁금합니다.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

mask하는 경우 인스턴스 변수를 사용하지 않아 staticmethod로도 충분하다고 생각했습니다.

_cloud_name은 클래스 변수로 인스턴스 전역에서 공유하도록 의도하여 getter, setter도 classmethod로 정의하였습니다.

나머지 함수들은 register_tools에서 tool로 등록하기 위해 self 사용하는 인스턴스 메서드로 정의했습니다.

def _mask_credential(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

비밀번호를 없애는 것이 아닌 마스킹하신 이유가 궁금합니다.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

clouds.yaml에는 비밀번호 없이도 입력이 가능한데, 그것과 구분하기 위해 마스킹하도록 하였습니다.

config_dict: dict, credential_keys: list[str]
) -> dict:
masked = {}
for k, v in config_dict.items():
if k in credential_keys:
masked[k] = "****"
elif isinstance(v, dict):
masked[k] = ConnectionManager._mask_credential(
v, credential_keys
)
else:
masked[k] = v
return masked

@classmethod
def get_cloud_name(cls) -> str:
"""Return the currently selected cloud name.

:return: current OpenStack cloud name.
"""
return cls._cloud_name

@classmethod
def set_cloud_name(cls, cloud_name: str) -> None:
"""Set cloud name to use for later connections. Must set name from currently valid cloud config file.

:param cloud_name: Name of the OpenStack cloud profile to activate.
"""
cls._cloud_name = cloud_name