1
- import json
2
1
import logging
3
2
import os
4
3
import re
5
- from typing import Any , Dict , List , Optional , Union
4
+ from typing import Any , Dict , Optional
6
5
7
6
from pydantic import TypeAdapter
8
7
@@ -18,8 +17,6 @@ class FiveireManager(JSONClientManager):
18
17
display_name = "5ire"
19
18
download_url = "https://5ire.app/"
20
19
21
- configure_key_name = "servers"
22
-
23
20
def __init__ (self , config_path = None ):
24
21
"""Initialize the 5ire client manager
25
22
@@ -40,124 +37,35 @@ def __init__(self, config_path=None):
40
37
# Linux
41
38
self .config_path = os .path .expanduser ("~/.config/5ire/mcp.json" )
42
39
40
+ self .server_name_key = {}
41
+
43
42
def _get_empty_config (self ) -> Dict [str , Any ]:
44
43
"""Get an empty configuration structure for this client
45
44
46
45
Returns:
47
46
Dict containing the client configuration with at least {"servers": []}
48
47
"""
49
- return {self .configure_key_name : []}
50
-
51
- def _load_config (self ) -> Dict [str , Any ]:
52
- """Load client configuration file
53
-
54
- Returns:
55
- Dict containing the client configuration with at least {"servers": []}
56
- """
57
- # Create empty config with the correct structure
58
- empty_config = self ._get_empty_config ()
59
-
60
- if not os .path .exists (self .config_path ):
61
- logger .warning (f"Client config file not found at: { self .config_path } " )
62
- return empty_config
63
-
64
- try :
65
- with open (self .config_path , "r" , encoding = "utf-8" ) as f :
66
- config = json .load (f )
67
- # Ensure servers section exists
68
- if self .configure_key_name not in config :
69
- config [self .configure_key_name ] = []
70
- return config
71
- except json .JSONDecodeError :
72
- logger .error (f"Error parsing client config file: { self .config_path } " )
73
-
74
- # Backup the corrupt file
75
- if os .path .exists (self .config_path ):
76
- backup_path = f"{ self .config_path } .bak"
77
- try :
78
- os .rename (self .config_path , backup_path )
79
- logger .info (f"Backed up corrupt config file to: { backup_path } " )
80
- except Exception as e :
81
- logger .error (f"Failed to backup corrupt file: { str (e )} " )
82
-
83
- # Return empty config
84
- return empty_config
85
-
86
- def get_servers (self ) -> Dict [str , Any ]:
87
- """Get all MCP servers configured for this client
88
-
89
- Returns:
90
- Dict of server configurations by name
91
- """
92
- config = self ._load_config ()
93
- servers = {}
94
-
95
- # Convert list of servers to dictionary by name
96
- for server in config .get (self .configure_key_name , []):
97
- if "name" in server :
98
- servers [server ["name" ]] = server
99
-
100
- return servers
101
-
102
- def get_server (self , server_name : str ) -> Optional [ServerConfig ]:
103
- """Get a server configuration
48
+ return {"mcpServers" : {}}
104
49
105
- Args:
106
- server_name: Name of the server
107
-
108
- Returns:
109
- ServerConfig object if found, None otherwise
110
- """
50
+ def _update_server_name_key (self ):
51
+ self .server_name_key = {}
111
52
servers = self .get_servers ()
53
+ for key , server_config in servers .items ():
54
+ self .server_name_key [server_config .get ("name" , key )] = key
112
55
113
- # Check if the server exists
114
- if server_name not in servers :
115
- logger .debug (f"Server { server_name } not found in { self .display_name } config" )
116
- return None
117
-
118
- # Get the server config and convert to ServerConfig
119
- return servers [server_name ]
120
-
121
- def add_server (self , server_config : Union [ServerConfig , Dict [str , Any ]], name : Optional [str ] = None ) -> bool :
122
- """Add or update a server in the client config
123
-
124
- Args:
125
- server_config: ServerConfig object or dictionary in client format
126
- name: Required server name when using dictionary format
127
-
128
- Returns:
129
- bool: Success or failure
130
- """
131
- # Handle direct dictionary input
132
- if isinstance (server_config , dict ):
133
- if name is None :
134
- raise ValueError ("Name must be provided when using dictionary format" )
135
- server_name = name
136
- client_config = server_config # Already in client format
137
- # Handle ServerConfig objects
138
- else :
139
- server_name = server_config .name
140
- client_config = self .to_client_format (server_config )
141
- client_config ["name" ] = server_name # Ensure name is in the config
142
-
143
- # Update config
144
- config = self ._load_config ()
145
-
146
- # Check if server already exists and update it
147
- server_exists = False
148
- for i , server in enumerate (config .get (self .configure_key_name , [])):
149
- if server .get ("name" ) == server_name :
150
- config [self .configure_key_name ][i ] = client_config
151
- server_exists = True
152
- break
153
-
154
- # If server doesn't exist, add it
155
- if not server_exists :
156
- if self .configure_key_name not in config :
157
- config [self .configure_key_name ] = []
158
- config [self .configure_key_name ].append (client_config )
56
+ def get_server (self , server_name : str ) -> Optional [ServerConfig ]:
57
+ self ._update_server_name_key ()
58
+ key = self .server_name_key .get (server_name )
59
+ if key :
60
+ return super ().get_server (key )
61
+ return None
159
62
160
- return self ._save_config (config )
63
+ def remove_server (self , server_name : str ) -> bool :
64
+ self ._update_server_name_key ()
65
+ key = self .server_name_key .get (server_name )
66
+ if key :
67
+ return super ().remove_server (key )
68
+ return False
161
69
162
70
def to_client_format (self , server_config : ServerConfig ) -> Dict [str , Any ]:
163
71
"""Convert ServerConfig to client-specific format
@@ -179,8 +87,10 @@ def to_client_format(self, server_config: ServerConfig) -> Dict[str, Any]:
179
87
non_empty_env = server_config .get_filtered_env_vars (os .environ )
180
88
if non_empty_env :
181
89
result ["env" ] = non_empty_env
90
+ result ["type" ] = "local"
182
91
else :
183
92
result = server_config .to_dict ()
93
+ result ["type" ] = "remote"
184
94
185
95
# Base result containing essential information
186
96
key_slug = re .sub (r"[^a-zA-Z0-9]" , "" , server_config .name )
@@ -214,39 +124,6 @@ def from_client_format(cls, server_name: str, client_config: Dict[str, Any]) ->
214
124
server_data .update (client_config )
215
125
return TypeAdapter (ServerConfig ).validate_python (server_data )
216
126
217
- def list_servers (self ) -> List [str ]:
218
- """List all MCP servers in client config
219
-
220
- Returns:
221
- List of server names
222
- """
223
- return list (self .get_servers ().keys ())
224
-
225
- def remove_server (self , server_name : str ) -> bool :
226
- """Remove an MCP server from client config
227
-
228
- Args:
229
- server_name: Name of the server to remove
230
-
231
- Returns:
232
- bool: Success or failure
233
- """
234
- config = self ._load_config ()
235
-
236
- # Find and remove the server
237
- server_found = False
238
- for i , server in enumerate (config .get (self .configure_key_name , [])):
239
- if server .get ("name" ) == server_name :
240
- config [self .configure_key_name ].pop (i )
241
- server_found = True
242
- break
243
-
244
- if not server_found :
245
- logger .warning (f"Server { server_name } not found in { self .display_name } config" )
246
- return False
247
-
248
- return self ._save_config (config )
249
-
250
127
def disable_server (self , server_name : str ) -> bool :
251
128
"""Temporarily disable a server by setting isActive to False
252
129
@@ -258,18 +135,12 @@ def disable_server(self, server_name: str) -> bool:
258
135
"""
259
136
config = self ._load_config ()
260
137
261
- # Find and disable the server
262
- server_found = False
263
- for i , server in enumerate (config .get (self .configure_key_name , [])):
264
- if server .get ("name" ) == server_name :
265
- config [self .configure_key_name ][i ]["isActive" ] = False
266
- server_found = True
267
- break
268
-
269
- if not server_found :
270
- logger .warning (f"Server { server_name } not found in { self .display_name } config" )
138
+ if "mcpServers" not in config or server_name not in config ["mcpServers" ]:
139
+ logger .warning (f"Server '{ server_name } ' not found in active servers" )
271
140
return False
272
141
142
+ config ["mcpServers" ][server_name ]["isActive" ] = False
143
+
273
144
return self ._save_config (config )
274
145
275
146
def enable_server (self , server_name : str ) -> bool :
@@ -283,18 +154,12 @@ def enable_server(self, server_name: str) -> bool:
283
154
"""
284
155
config = self ._load_config ()
285
156
286
- # Find and enable the server
287
- server_found = False
288
- for i , server in enumerate (config .get (self .configure_key_name , [])):
289
- if server .get ("name" ) == server_name :
290
- config [self .configure_key_name ][i ]["isActive" ] = True
291
- server_found = True
292
- break
293
-
294
- if not server_found :
295
- logger .warning (f"Server { server_name } not found in { self .display_name } config" )
157
+ if "mcpServers" not in config or server_name not in config ["mcpServers" ]:
158
+ logger .warning (f"Server '{ server_name } ' not found in active servers" )
296
159
return False
297
160
161
+ config ["mcpServers" ][server_name ]["isActive" ] = True
162
+
298
163
return self ._save_config (config )
299
164
300
165
def is_server_disabled (self , server_name : str ) -> bool :
0 commit comments