Skip to content

Sync

chatbox_modelscope_sync_mcp.sync

ModelScope MCP同步工具

该模块提供了将ModelScope平台上的MCP服务同步到Chatbox应用配置中的功能。 通过调用ModelScope API获取最新的MCP服务列表,并更新本地配置文件。

主要功能: 1. 自动检测Chatbox配置文件路径 2. 调用ModelScope API获取MCP服务列表 3. 智能过滤和更新MCP服务器配置 4. 自动备份配置文件 5. 支持自定义API密钥和配置路径

示例: >>> syncer = ModelScopeMCPSync(api_key="your_api_key") >>> syncer.sync() # 执行同步操作

ModelScopeMCPSync

ModelScope MCP同步工具类

用于同步ModelScope平台上的MCP服务到Chatbox应用的配置文件中。

Attributes:

Name Type Description
api_key str

ModelScope API密钥

config_path str

Chatbox配置文件路径

api_url str

ModelScope API的URL地址

Example

syncer = ModelScopeMCPSync(api_key="your_key") syncer.sync() # 执行同步

Source code in chatbox_modelscope_sync_mcp/sync.py
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
class ModelScopeMCPSync:
    """
    ModelScope MCP同步工具类

    用于同步ModelScope平台上的MCP服务到Chatbox应用的配置文件中。

    Attributes:
        api_key (str): ModelScope API密钥
        config_path (str): Chatbox配置文件路径
        api_url (str): ModelScope API的URL地址

    Example:
        >>> syncer = ModelScopeMCPSync(api_key="your_key")
        >>> syncer.sync()  # 执行同步
    """

    def __init__(self, api_key=None, config_path=None, api_url=None):
        """
        初始化ModelScope MCP同步工具

        Args:
            api_key (str, optional): ModelScope API密钥。如果未提供,将尝试从环境变量MODELSCOPE_API_KEY获取
            config_path (str, optional): Chatbox配置文件路径。如果未提供,将自动检测系统默认路径
            api_url (str, optional): ModelScope API的URL地址。默认为官方API地址

        配置路径自动检测规则:
        - Windows: ~/AppData/Roaming/xyz.chatboxapp.app/config.json
        - macOS: ~/Library/Application Support/xyz.chatboxapp.app/config.json
        - Linux: ~/.config/xyz.chatboxapp.app/config.json
        """
        self.api_key = api_key
        self.config_path = config_path
        self.api_url = api_url or "https://www.modelscope.cn/api/v1/mcp/services/operational"

    def _get_config_path(self):
        """
        获取Chatbox配置文件路径

        根据操作系统类型自动检测Chatbox应用的默认配置文件路径。

        Returns:
            str: 配置文件完整路径

        Raises:
            无异常抛出,如果无法确定路径,返回'config.json'
        """
        if self.config_path:
            return self.config_path

        # 自动检测Chatbox配置路径
        if os.name == 'nt':  # Windows
            return os.path.expanduser('~/AppData/Roaming/xyz.chatboxapp.app/config.json')
        elif os.name == 'posix':
            if sys.platform == 'darwin':  # macOS
                return os.path.expanduser('~/Library/Application Support/xyz.chatboxapp.app/config.json')
            else:  # Linux
                return os.path.expanduser('~/.config/xyz.chatboxapp.app/config.json')

        return 'config.json'

    def _get_api_key(self):
        """
        获取ModelScope API密钥

        优先级:
        1. 构造函数中提供的api_key参数
        2. 环境变量MODELSCOPE_API_KEY

        Returns:
            str: API密钥

        Raises:
            ValueError: 如果未找到有效的API密钥
        """
        return self.api_key or os.getenv('MODELSCOPE_API_KEY')

    def call_modelscope_api(self):
        """
        调用ModelScope API获取MCP服务列表

        使用提供的API密钥调用ModelScope平台的MCP服务API,
        获取当前可用的MCP服务信息。

        Returns:
            dict: API响应的JSON数据

        Raises:
            ValueError: 如果未提供API密钥
            RuntimeError: 如果API调用失败(网络错误、认证失败等)

        API端点:
            GET https://www.modelscope.cn/api/v1/mcp/services/operational

        响应格式:
            {
                "Data": {
                    "Result": [
                        {
                            "id": "服务唯一标识",
                            "name": "服务名称",
                            "chinese_name": "中文名称",
                            "locales": {"zh": {"name": "本地化名称"}},
                            "operational_urls": [{"url": "服务URL"}]
                        }
                    ]
                }
            }
        """
        api_key = self._get_api_key()
        if not api_key:
            raise ValueError("未提供API密钥,请使用--token参数或设置MODELSCOPE_API_KEY环境变量")

        headers = {
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json"
        }

        try:
            response = requests.get(self.api_url, headers=headers, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise RuntimeError(f"API调用失败: {e}")

    def load_config(self, config_path):
        """
        加载Chatbox配置文件

        从指定路径加载Chatbox应用的配置文件。
        如果文件不存在,会创建包含默认MCP服务器配置的新配置。

        Args:
            config_path (str): 配置文件路径

        Returns:
            dict: 配置数据的Python字典

        Raises:
            ValueError: 如果配置文件格式错误(JSON解析失败)
            RuntimeError: 如果文件读取失败

        默认配置结构:
            {
                "settings": {
                    "mcp": {
                        "servers": []
                    }
                }
            }
        """
        try:
            with open(config_path, 'r', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            # 如果文件不存在,创建默认配置
            return {
                "settings": {
                    "mcp": {
                        "servers": []
                    }
                }
            }
        except json.JSONDecodeError as e:
            raise ValueError(f"配置文件格式错误: {e}")
        except Exception as e:
            raise RuntimeError(f"加载配置文件失败: {e}")

    def backup_config(self, config_path):
        """
        备份配置文件

        在更新配置前创建配置文件的备份副本,
        备份文件命名为原文件名加.bak后缀。

        Args:
            config_path (str): 要备份的配置文件路径

        Returns:
            bool: 备份成功返回True,失败返回False

        备份文件命名:
            原文件: config.json
            备份文件: config.json.bak
        """
        backup_path = f"{config_path}.bak"
        try:
            shutil.copy2(config_path, backup_path)
            print(f"已备份配置文件到: {backup_path}")
            return True
        except Exception as e:
            print(f"备份配置文件失败: {e}")
            return False

    def save_config(self, config, config_path):
        """
        保存配置文件

        将更新后的配置数据保存到指定路径的配置文件中。
        如果目录不存在会自动创建。

        Args:
            config (dict): 要保存的配置数据
            config_path (str): 目标配置文件路径

        Returns:
            bool: 保存成功返回True

        Raises:
            RuntimeError: 如果保存失败

        文件格式:
            使用UTF-8编码,JSON格式,2空格缩进,保留非ASCII字符
        """
        try:
            os.makedirs(os.path.dirname(config_path), exist_ok=True)
            with open(config_path, 'w', encoding='utf-8') as f:
                json.dump(config, f, indent=2, ensure_ascii=False)
            print(f"已更新配置文件: {config_path}")
            return True
        except Exception as e:
            raise RuntimeError(f"保存配置文件失败: {e}")

    def get_server_name(self, service):
        """
        智能获取服务器名称

        从服务信息中提取最合适的服务器名称。
        优先级:中文名称 > 本地化名称 > 英文名称 > 原始名称 > ID转换

        Args:
            service (dict): 服务信息字典,包含名称相关字段

        Returns:
            str: 服务器名称,如果无法确定返回None

        名称提取规则:
        1. chinese_name字段(中文名称)
        2. locales.zh.name或locales.en.name(本地化名称)
        3. name字段(原始名称)
        4. id字段(转换@和/为-)
        """
        name = service.get('chinese_name', '').strip()
        if name:
            return name

        locales = service.get('locales', {})
        for lang in ['zh', 'en']:
            if lang in locales:
                name = locales[lang].get('name', '').strip()
                if name:
                    return name

        name = service.get('name', '').strip()
        if name:
            return name

        server_id = service.get('id', '').strip()
        if server_id:
            return server_id.replace('@', '').replace('/', '-')

        return None

    def filter_valid_servers(self, api_response):
        """
        过滤有效服务器

        从API响应中过滤出有效的MCP服务器。
        有效服务器的定义:
        - 有有效的服务器名称
        - 有operational_urls列表
        - 第一个URL有效

        Args:
            api_response (dict): ModelScope API的响应数据

        Returns:
            list: 有效服务器列表,每个元素包含name、url、id字段

        返回格式:
            [
                {
                    "name": "服务器名称",
                    "url": "服务URL",
                    "id": "服务唯一标识"
                }
            ]
        """
        if not api_response or 'Data' not in api_response or 'Result' not in api_response['Data']:
            return []

        valid_servers = []
        for service in api_response['Data']['Result']:
            name = self.get_server_name(service)
            if not name:
                continue

            urls = service.get('operational_urls', [])
            if not urls:
                continue

            url = urls[0].get('url')
            if not url:
                continue

            valid_servers.append({
                'name': name,
                'url': url,
                'id': service.get('id', 'unknown')
            })

        return valid_servers

    def sync(self, backup=True):
        """
        执行MCP服务同步

        完整的同步流程:
        1. 调用ModelScope API获取最新服务列表
        2. 加载当前Chatbox配置
        3. 过滤有效的新服务器
        4. 对比并更新现有配置
        5. 备份并保存更新后的配置

        Args:
            backup (bool, optional): 是否在更新前备份配置文件,默认为True

        Returns:
            bool: 同步成功返回True,无更新返回True,失败返回False

        控制台输出:
        - 正在调用ModelScope API...
        - 正在加载配置文件...
        - 更新: 旧名称 -> 新名称
        - 新增: 新服务器名称
        - 同步完成: 更新X个, 新增Y个

        同步规则:
        - 同名URL:仅更新名称
        - 新URL:添加为新服务器
        - 现有服务器保持不变
        - 每个新服务器分配唯一UUID
        """
        config_path = self._get_config_path()

        print("正在调用ModelScope API...")
        api_response = self.call_modelscope_api()

        print("正在加载配置文件...")
        config = self.load_config(config_path)

        valid_servers = self.filter_valid_servers(api_response)
        if not valid_servers:
            print("没有找到有效的MCP服务器")
            return False

        # 获取现有服务器
        servers = config['settings']['mcp']['servers']
        existing_urls = {s['transport']['url']: s for s in servers
                         if s.get('transport', {}).get('type') == 'http'}

        updated_count = 0
        added_count = 0

        for server_info in valid_servers:
            url = server_info['url']
            name = server_info['name']

            if url in existing_urls:
                old_server = existing_urls[url]
                if old_server['name'] != name:
                    old_name = old_server['name']
                    old_server['name'] = name
                    updated_count += 1
                    print(f"更新: {old_name} -> {name}")
            else:
                new_server = {
                    "id": str(uuid.uuid4()),
                    "name": name,
                    "enabled": True,
                    "transport": {
                        "type": "http",
                        "url": url
                    }
                }
                servers.append(new_server)
                added_count += 1
                print(f"新增: {name}")

        if updated_count == 0 and added_count == 0:
            print("配置已是最新,无需更新")
            return True

        # 备份并更新配置
        if backup and os.path.exists(config_path):
            self.backup_config(config_path)

        self.save_config(config, config_path)
        print(f"同步完成: 更新{updated_count}个, 新增{added_count}个")
        return True

    def export_mcp_json_to_file(self, output_path):
        """
        导出纯MCP JSON格式到指定文件

        将ModelScope的MCP服务转换为标准的mcp.json格式并保存到指定路径。

        Args:
            output_path (str): 输出文件路径

        Returns:
            bool: 导出成功返回True,失败返回False

        输出格式:
            {
                "mcpServers": {
                    "server-name": {
                        "type": "sse",
                        "url": "https://example.com/sse"
                    }
                }
            }
        """
        try:
            print("正在调用ModelScope API...")
            api_response = self.call_modelscope_api()

            valid_servers = self.filter_valid_servers(api_response)
            if not valid_servers:
                print("没有找到有效的MCP服务器")
                return False

            mcp_servers = {}
            for server_info in valid_servers:
                name = server_info['name']
                url = server_info['url']

                # 转换为安全的key名称(替换空格和特殊字符)
                safe_name = name.lower().replace(' ', '-').replace('_', '-')
                safe_name = ''.join(c for c in safe_name if c.isalnum() or c == '-')

                # 确保SSE URL格式
                if not url.endswith('/sse'):
                    if not url.endswith('/'):
                        url = url + '/'
                    url = url + 'sse'

                mcp_servers[safe_name] = {
                    "type": "sse",
                    "url": url
                }

            output_data = {"mcpServers": mcp_servers}

            # 创建输出目录(如果不存在)
            output_dir = os.path.dirname(output_path)
            if output_dir:
                os.makedirs(output_dir, exist_ok=True)

            # 写入JSON文件
            with open(output_path, 'w', encoding='utf-8') as f:
                json.dump(output_data, f, indent=2, ensure_ascii=False)

            print(f"MCP JSON已导出到: {output_path}")
            return True

        except Exception as e:
            print(f"导出失败: {e}")
            return False

__init__(api_key=None, config_path=None, api_url=None)

初始化ModelScope MCP同步工具

Parameters:

Name Type Description Default
api_key str

ModelScope API密钥。如果未提供,将尝试从环境变量MODELSCOPE_API_KEY获取

None
config_path str

Chatbox配置文件路径。如果未提供,将自动检测系统默认路径

None
api_url str

ModelScope API的URL地址。默认为官方API地址

None

配置路径自动检测规则: - Windows: ~/AppData/Roaming/xyz.chatboxapp.app/config.json - macOS: ~/Library/Application Support/xyz.chatboxapp.app/config.json - Linux: ~/.config/xyz.chatboxapp.app/config.json

Source code in chatbox_modelscope_sync_mcp/sync.py
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(self, api_key=None, config_path=None, api_url=None):
    """
    初始化ModelScope MCP同步工具

    Args:
        api_key (str, optional): ModelScope API密钥。如果未提供,将尝试从环境变量MODELSCOPE_API_KEY获取
        config_path (str, optional): Chatbox配置文件路径。如果未提供,将自动检测系统默认路径
        api_url (str, optional): ModelScope API的URL地址。默认为官方API地址

    配置路径自动检测规则:
    - Windows: ~/AppData/Roaming/xyz.chatboxapp.app/config.json
    - macOS: ~/Library/Application Support/xyz.chatboxapp.app/config.json
    - Linux: ~/.config/xyz.chatboxapp.app/config.json
    """
    self.api_key = api_key
    self.config_path = config_path
    self.api_url = api_url or "https://www.modelscope.cn/api/v1/mcp/services/operational"

backup_config(config_path)

备份配置文件

在更新配置前创建配置文件的备份副本, 备份文件命名为原文件名加.bak后缀。

Parameters:

Name Type Description Default
config_path str

要备份的配置文件路径

required

Returns:

Name Type Description
bool

备份成功返回True,失败返回False

备份文件命名: 原文件: config.json 备份文件: config.json.bak

Source code in chatbox_modelscope_sync_mcp/sync.py
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def backup_config(self, config_path):
    """
    备份配置文件

    在更新配置前创建配置文件的备份副本,
    备份文件命名为原文件名加.bak后缀。

    Args:
        config_path (str): 要备份的配置文件路径

    Returns:
        bool: 备份成功返回True,失败返回False

    备份文件命名:
        原文件: config.json
        备份文件: config.json.bak
    """
    backup_path = f"{config_path}.bak"
    try:
        shutil.copy2(config_path, backup_path)
        print(f"已备份配置文件到: {backup_path}")
        return True
    except Exception as e:
        print(f"备份配置文件失败: {e}")
        return False

call_modelscope_api()

调用ModelScope API获取MCP服务列表

使用提供的API密钥调用ModelScope平台的MCP服务API, 获取当前可用的MCP服务信息。

Returns:

Name Type Description
dict

API响应的JSON数据

Raises:

Type Description
ValueError

如果未提供API密钥

RuntimeError

如果API调用失败(网络错误、认证失败等)

API端点: GET https://www.modelscope.cn/api/v1/mcp/services/operational

响应格式: { "Data": { "Result": [ { "id": "服务唯一标识", "name": "服务名称", "chinese_name": "中文名称", "locales": {"zh": {"name": "本地化名称"}}, "operational_urls": [{"url": "服务URL"}] } ] } }

Source code in chatbox_modelscope_sync_mcp/sync.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def call_modelscope_api(self):
    """
    调用ModelScope API获取MCP服务列表

    使用提供的API密钥调用ModelScope平台的MCP服务API,
    获取当前可用的MCP服务信息。

    Returns:
        dict: API响应的JSON数据

    Raises:
        ValueError: 如果未提供API密钥
        RuntimeError: 如果API调用失败(网络错误、认证失败等)

    API端点:
        GET https://www.modelscope.cn/api/v1/mcp/services/operational

    响应格式:
        {
            "Data": {
                "Result": [
                    {
                        "id": "服务唯一标识",
                        "name": "服务名称",
                        "chinese_name": "中文名称",
                        "locales": {"zh": {"name": "本地化名称"}},
                        "operational_urls": [{"url": "服务URL"}]
                    }
                ]
            }
        }
    """
    api_key = self._get_api_key()
    if not api_key:
        raise ValueError("未提供API密钥,请使用--token参数或设置MODELSCOPE_API_KEY环境变量")

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Accept": "application/json"
    }

    try:
        response = requests.get(self.api_url, headers=headers, timeout=30)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"API调用失败: {e}")

export_mcp_json_to_file(output_path)

导出纯MCP JSON格式到指定文件

将ModelScope的MCP服务转换为标准的mcp.json格式并保存到指定路径。

Parameters:

Name Type Description Default
output_path str

输出文件路径

required

Returns:

Name Type Description
bool

导出成功返回True,失败返回False

输出格式: { "mcpServers": { "server-name": { "type": "sse", "url": "https://example.com/sse" } } }

Source code in chatbox_modelscope_sync_mcp/sync.py
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
def export_mcp_json_to_file(self, output_path):
    """
    导出纯MCP JSON格式到指定文件

    将ModelScope的MCP服务转换为标准的mcp.json格式并保存到指定路径。

    Args:
        output_path (str): 输出文件路径

    Returns:
        bool: 导出成功返回True,失败返回False

    输出格式:
        {
            "mcpServers": {
                "server-name": {
                    "type": "sse",
                    "url": "https://example.com/sse"
                }
            }
        }
    """
    try:
        print("正在调用ModelScope API...")
        api_response = self.call_modelscope_api()

        valid_servers = self.filter_valid_servers(api_response)
        if not valid_servers:
            print("没有找到有效的MCP服务器")
            return False

        mcp_servers = {}
        for server_info in valid_servers:
            name = server_info['name']
            url = server_info['url']

            # 转换为安全的key名称(替换空格和特殊字符)
            safe_name = name.lower().replace(' ', '-').replace('_', '-')
            safe_name = ''.join(c for c in safe_name if c.isalnum() or c == '-')

            # 确保SSE URL格式
            if not url.endswith('/sse'):
                if not url.endswith('/'):
                    url = url + '/'
                url = url + 'sse'

            mcp_servers[safe_name] = {
                "type": "sse",
                "url": url
            }

        output_data = {"mcpServers": mcp_servers}

        # 创建输出目录(如果不存在)
        output_dir = os.path.dirname(output_path)
        if output_dir:
            os.makedirs(output_dir, exist_ok=True)

        # 写入JSON文件
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(output_data, f, indent=2, ensure_ascii=False)

        print(f"MCP JSON已导出到: {output_path}")
        return True

    except Exception as e:
        print(f"导出失败: {e}")
        return False

filter_valid_servers(api_response)

过滤有效服务器

从API响应中过滤出有效的MCP服务器。 有效服务器的定义: - 有有效的服务器名称 - 有operational_urls列表 - 第一个URL有效

Parameters:

Name Type Description Default
api_response dict

ModelScope API的响应数据

required

Returns:

Name Type Description
list

有效服务器列表,每个元素包含name、url、id字段

返回格式: [ { "name": "服务器名称", "url": "服务URL", "id": "服务唯一标识" } ]

Source code in chatbox_modelscope_sync_mcp/sync.py
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def filter_valid_servers(self, api_response):
    """
    过滤有效服务器

    从API响应中过滤出有效的MCP服务器。
    有效服务器的定义:
    - 有有效的服务器名称
    - 有operational_urls列表
    - 第一个URL有效

    Args:
        api_response (dict): ModelScope API的响应数据

    Returns:
        list: 有效服务器列表,每个元素包含name、url、id字段

    返回格式:
        [
            {
                "name": "服务器名称",
                "url": "服务URL",
                "id": "服务唯一标识"
            }
        ]
    """
    if not api_response or 'Data' not in api_response or 'Result' not in api_response['Data']:
        return []

    valid_servers = []
    for service in api_response['Data']['Result']:
        name = self.get_server_name(service)
        if not name:
            continue

        urls = service.get('operational_urls', [])
        if not urls:
            continue

        url = urls[0].get('url')
        if not url:
            continue

        valid_servers.append({
            'name': name,
            'url': url,
            'id': service.get('id', 'unknown')
        })

    return valid_servers

get_server_name(service)

智能获取服务器名称

从服务信息中提取最合适的服务器名称。 优先级:中文名称 > 本地化名称 > 英文名称 > 原始名称 > ID转换

Parameters:

Name Type Description Default
service dict

服务信息字典,包含名称相关字段

required

Returns:

Name Type Description
str

服务器名称,如果无法确定返回None

名称提取规则: 1. chinese_name字段(中文名称) 2. locales.zh.name或locales.en.name(本地化名称) 3. name字段(原始名称) 4. id字段(转换@和/为-)

Source code in chatbox_modelscope_sync_mcp/sync.py
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def get_server_name(self, service):
    """
    智能获取服务器名称

    从服务信息中提取最合适的服务器名称。
    优先级:中文名称 > 本地化名称 > 英文名称 > 原始名称 > ID转换

    Args:
        service (dict): 服务信息字典,包含名称相关字段

    Returns:
        str: 服务器名称,如果无法确定返回None

    名称提取规则:
    1. chinese_name字段(中文名称)
    2. locales.zh.name或locales.en.name(本地化名称)
    3. name字段(原始名称)
    4. id字段(转换@和/为-)
    """
    name = service.get('chinese_name', '').strip()
    if name:
        return name

    locales = service.get('locales', {})
    for lang in ['zh', 'en']:
        if lang in locales:
            name = locales[lang].get('name', '').strip()
            if name:
                return name

    name = service.get('name', '').strip()
    if name:
        return name

    server_id = service.get('id', '').strip()
    if server_id:
        return server_id.replace('@', '').replace('/', '-')

    return None

load_config(config_path)

加载Chatbox配置文件

从指定路径加载Chatbox应用的配置文件。 如果文件不存在,会创建包含默认MCP服务器配置的新配置。

Parameters:

Name Type Description Default
config_path str

配置文件路径

required

Returns:

Name Type Description
dict

配置数据的Python字典

Raises:

Type Description
ValueError

如果配置文件格式错误(JSON解析失败)

RuntimeError

如果文件读取失败

默认配置结构: { "settings": { "mcp": { "servers": [] } } }

Source code in chatbox_modelscope_sync_mcp/sync.py
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def load_config(self, config_path):
    """
    加载Chatbox配置文件

    从指定路径加载Chatbox应用的配置文件。
    如果文件不存在,会创建包含默认MCP服务器配置的新配置。

    Args:
        config_path (str): 配置文件路径

    Returns:
        dict: 配置数据的Python字典

    Raises:
        ValueError: 如果配置文件格式错误(JSON解析失败)
        RuntimeError: 如果文件读取失败

    默认配置结构:
        {
            "settings": {
                "mcp": {
                    "servers": []
                }
            }
        }
    """
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except FileNotFoundError:
        # 如果文件不存在,创建默认配置
        return {
            "settings": {
                "mcp": {
                    "servers": []
                }
            }
        }
    except json.JSONDecodeError as e:
        raise ValueError(f"配置文件格式错误: {e}")
    except Exception as e:
        raise RuntimeError(f"加载配置文件失败: {e}")

save_config(config, config_path)

保存配置文件

将更新后的配置数据保存到指定路径的配置文件中。 如果目录不存在会自动创建。

Parameters:

Name Type Description Default
config dict

要保存的配置数据

required
config_path str

目标配置文件路径

required

Returns:

Name Type Description
bool

保存成功返回True

Raises:

Type Description
RuntimeError

如果保存失败

文件格式: 使用UTF-8编码,JSON格式,2空格缩进,保留非ASCII字符

Source code in chatbox_modelscope_sync_mcp/sync.py
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def save_config(self, config, config_path):
    """
    保存配置文件

    将更新后的配置数据保存到指定路径的配置文件中。
    如果目录不存在会自动创建。

    Args:
        config (dict): 要保存的配置数据
        config_path (str): 目标配置文件路径

    Returns:
        bool: 保存成功返回True

    Raises:
        RuntimeError: 如果保存失败

    文件格式:
        使用UTF-8编码,JSON格式,2空格缩进,保留非ASCII字符
    """
    try:
        os.makedirs(os.path.dirname(config_path), exist_ok=True)
        with open(config_path, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
        print(f"已更新配置文件: {config_path}")
        return True
    except Exception as e:
        raise RuntimeError(f"保存配置文件失败: {e}")

sync(backup=True)

执行MCP服务同步

完整的同步流程: 1. 调用ModelScope API获取最新服务列表 2. 加载当前Chatbox配置 3. 过滤有效的新服务器 4. 对比并更新现有配置 5. 备份并保存更新后的配置

Parameters:

Name Type Description Default
backup bool

是否在更新前备份配置文件,默认为True

True

Returns:

Name Type Description
bool

同步成功返回True,无更新返回True,失败返回False

控制台输出: - 正在调用ModelScope API... - 正在加载配置文件... - 更新: 旧名称 -> 新名称 - 新增: 新服务器名称 - 同步完成: 更新X个, 新增Y个

同步规则: - 同名URL:仅更新名称 - 新URL:添加为新服务器 - 现有服务器保持不变 - 每个新服务器分配唯一UUID

Source code in chatbox_modelscope_sync_mcp/sync.py
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def sync(self, backup=True):
    """
    执行MCP服务同步

    完整的同步流程:
    1. 调用ModelScope API获取最新服务列表
    2. 加载当前Chatbox配置
    3. 过滤有效的新服务器
    4. 对比并更新现有配置
    5. 备份并保存更新后的配置

    Args:
        backup (bool, optional): 是否在更新前备份配置文件,默认为True

    Returns:
        bool: 同步成功返回True,无更新返回True,失败返回False

    控制台输出:
    - 正在调用ModelScope API...
    - 正在加载配置文件...
    - 更新: 旧名称 -> 新名称
    - 新增: 新服务器名称
    - 同步完成: 更新X个, 新增Y个

    同步规则:
    - 同名URL:仅更新名称
    - 新URL:添加为新服务器
    - 现有服务器保持不变
    - 每个新服务器分配唯一UUID
    """
    config_path = self._get_config_path()

    print("正在调用ModelScope API...")
    api_response = self.call_modelscope_api()

    print("正在加载配置文件...")
    config = self.load_config(config_path)

    valid_servers = self.filter_valid_servers(api_response)
    if not valid_servers:
        print("没有找到有效的MCP服务器")
        return False

    # 获取现有服务器
    servers = config['settings']['mcp']['servers']
    existing_urls = {s['transport']['url']: s for s in servers
                     if s.get('transport', {}).get('type') == 'http'}

    updated_count = 0
    added_count = 0

    for server_info in valid_servers:
        url = server_info['url']
        name = server_info['name']

        if url in existing_urls:
            old_server = existing_urls[url]
            if old_server['name'] != name:
                old_name = old_server['name']
                old_server['name'] = name
                updated_count += 1
                print(f"更新: {old_name} -> {name}")
        else:
            new_server = {
                "id": str(uuid.uuid4()),
                "name": name,
                "enabled": True,
                "transport": {
                    "type": "http",
                    "url": url
                }
            }
            servers.append(new_server)
            added_count += 1
            print(f"新增: {name}")

    if updated_count == 0 and added_count == 0:
        print("配置已是最新,无需更新")
        return True

    # 备份并更新配置
    if backup and os.path.exists(config_path):
        self.backup_config(config_path)

    self.save_config(config, config_path)
    print(f"同步完成: 更新{updated_count}个, 新增{added_count}个")
    return True