Skip to content

Commit 4835238

Browse files
committed
feat: connection manager related tool
1 parent 1b64f2b commit 4835238

File tree

4 files changed

+87
-19
lines changed

4 files changed

+87
-19
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: 8 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,16 @@
1-
import openstack
1+
from openstack_mcp_server.tools.connection import ConnectionManager
22

3-
from openstack import connection
43

5-
from openstack_mcp_server import config
4+
_connection_manager = ConnectionManager()
65

76

8-
class OpenStackConnectionManager:
9-
"""OpenStack Connection Manager"""
10-
11-
_connection: connection.Connection | None = None
12-
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
7+
def get_openstack_conn():
8+
return _connection_manager.get_connection()
209

2110

22-
_openstack_connection_manager = OpenStackConnectionManager()
11+
def set_openstack_cloud_name(cloud_name: str) -> None:
12+
_connection_manager.set_cloud_name(cloud_name)
2313

2414

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

0 commit comments

Comments
 (0)