Skip to content

Commit e9fe44b

Browse files
choieastseahalucinor
authored andcommitted
feat: connection manager related tool (#90)
* feat: connection manager related tool * fix: logging은 base.py에서 수행하도록 수정
1 parent be9420e commit e9fe44b

File tree

4 files changed

+86
-16
lines changed

4 files changed

+86
-16
lines changed

.gitignore

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -194,3 +194,6 @@ cython_debug/
194194
# refer to https://docs.cursor.com/context/ignore-files
195195
.cursorignore
196196
.cursorindexingignore
197+
198+
# Openstack Config Files
199+
clouds.yaml

src/openstack_mcp_server/tools/__init__.py

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
from fastmcp import FastMCP
22

3+
from openstack_mcp_server.tools.connection import ConnectionManager
4+
35

46
def register_tool(mcp: FastMCP):
57
"""
@@ -17,3 +19,4 @@ def register_tool(mcp: FastMCP):
1719
IdentityTools().register_tools(mcp)
1820
NetworkTools().register_tools(mcp)
1921
BlockStorageTools().register_tools(mcp)
22+
ConnectionManager().register_tools(mcp)
Lines changed: 10 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,21 @@
11
import openstack
22

3-
from openstack import connection
4-
53
from openstack_mcp_server import config
4+
from openstack_mcp_server.tools.connection import ConnectionManager
5+
66

7+
_connection_manager = ConnectionManager()
78

8-
class OpenStackConnectionManager:
9-
"""OpenStack Connection Manager"""
9+
openstack.enable_logging(debug=config.MCP_DEBUG_MODE)
1010

11-
_connection: connection.Connection | None = None
1211

13-
@classmethod
14-
def get_connection(cls) -> connection.Connection:
15-
"""OpenStack Connection"""
16-
if cls._connection is None:
17-
openstack.enable_logging(debug=config.MCP_DEBUG_MODE)
18-
cls._connection = openstack.connect(cloud=config.MCP_CLOUD_NAME)
19-
return cls._connection
12+
def get_openstack_conn():
13+
return _connection_manager.get_connection()
2014

2115

22-
_openstack_connection_manager = OpenStackConnectionManager()
16+
def set_openstack_cloud_name(cloud_name: str) -> None:
17+
_connection_manager.set_cloud_name(cloud_name)
2318

2419

25-
def get_openstack_conn():
26-
"""Get OpenStack Connection"""
27-
return _openstack_connection_manager.get_connection()
20+
def get_openstack_cloud_name() -> str:
21+
return _connection_manager.get_cloud_name()
Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
import openstack
2+
3+
from fastmcp import FastMCP
4+
from openstack import connection
5+
from openstack.config.loader import OpenStackConfig
6+
7+
from openstack_mcp_server import config
8+
9+
10+
class ConnectionManager:
11+
_cloud_name = config.MCP_CLOUD_NAME
12+
13+
def register_tools(self, mcp: FastMCP):
14+
mcp.tool(self.get_cloud_config)
15+
mcp.tool(self.get_cloud_names)
16+
mcp.tool(self.get_cloud_name)
17+
mcp.tool(self.set_cloud_name)
18+
19+
def get_connection(self) -> connection.Connection:
20+
return openstack.connect(cloud=self._cloud_name)
21+
22+
def get_cloud_names(self) -> list[str]:
23+
"""List available cloud configurations.
24+
25+
:return: Names of OpenStack clouds from user's config file.
26+
"""
27+
config = OpenStackConfig()
28+
return list(config.get_cloud_names())
29+
30+
def get_cloud_config(self) -> dict:
31+
"""Provide cloud configuration with secrets masked of current user's config file.
32+
33+
:return: Cloud configuration dictionary with credentials masked.
34+
"""
35+
config = OpenStackConfig()
36+
return ConnectionManager._mask_credential(
37+
config.cloud_config, ["password"]
38+
)
39+
40+
@staticmethod
41+
def _mask_credential(
42+
config_dict: dict, credential_keys: list[str]
43+
) -> dict:
44+
masked = {}
45+
for k, v in config_dict.items():
46+
if k in credential_keys:
47+
masked[k] = "****"
48+
elif isinstance(v, dict):
49+
masked[k] = ConnectionManager._mask_credential(
50+
v, credential_keys
51+
)
52+
else:
53+
masked[k] = v
54+
return masked
55+
56+
@classmethod
57+
def get_cloud_name(cls) -> str:
58+
"""Return the currently selected cloud name.
59+
60+
:return: current OpenStack cloud name.
61+
"""
62+
return cls._cloud_name
63+
64+
@classmethod
65+
def set_cloud_name(cls, cloud_name: str) -> None:
66+
"""Set cloud name to use for later connections. Must set name from currently valid cloud config file.
67+
68+
:param cloud_name: Name of the OpenStack cloud profile to activate.
69+
"""
70+
cls._cloud_name = cloud_name

0 commit comments

Comments
 (0)