re：以直接代码而不是submodule形式添加内置插件

plugins/MaiBot_MCPBridgePlugin/config_converter.py

@@ -0,0 +1,448 @@
+"""
+MCP 配置格式转换模块 v1.0.0
+
+支持的格式:
+- Claude Desktop (claude_desktop_config.json)
+- Kiro MCP (mcp.json)
+- MaiBot MCP Bridge Plugin (本插件格式)
+
+转换规则:
+- stdio: command + args + env
+- sse/http/streamable_http: url + headers
+"""
+
+import json
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Tuple
+
+
+@dataclass
+class ConversionResult:
+    """转换结果"""
+    success: bool
+    servers: List[Dict[str, Any]] = field(default_factory=list)
+    errors: List[str] = field(default_factory=list)
+    warnings: List[str] = field(default_factory=list)
+    skipped: List[str] = field(default_factory=list)
+
+
+class ConfigConverter:
+    """MCP 配置格式转换器"""
+
+    # transport 类型映射 (外部格式 -> 内部格式)
+    TRANSPORT_MAP_IN = {
+        "sse": "sse",
+        "http": "http",
+        "streamable-http": "streamable_http",
+        "streamable_http": "streamable_http",
+        "streamable-http": "streamable_http",
+        "stdio": "stdio",
+    }
+
+    # 支持的 transport 字段名（有些格式用 type 而不是 transport）
+    TRANSPORT_FIELD_NAMES = ["transport", "type"]
+
+    # transport 类型映射 (内部格式 -> Claude 格式)
+    TRANSPORT_MAP_OUT = {
+        "sse": "sse",
+        "http": "http",
+        "streamable_http": "streamable-http",
+        "stdio": "stdio",
+    }
+
+    @classmethod
+    def detect_format(cls, config: Dict[str, Any]) -> Optional[str]:
+        """检测配置格式类型
+        
+        Returns:
+            "claude": Claude Desktop 格式 (mcpServers 对象)
+            "kiro": Kiro MCP 格式 (mcpServers 对象，与 Claude 相同)
+            "maibot": MaiBot 插件格式 (数组)
+            None: 无法识别
+        """
+        if isinstance(config, list):
+            # 数组格式，检查是否是 MaiBot 格式
+            if len(config) == 0:
+                return "maibot"
+            if isinstance(config[0], dict) and "name" in config[0]:
+                return "maibot"
+            return None
+
+        if isinstance(config, dict):
+            # 对象格式
+            if "mcpServers" in config:
+                return "claude"  # Claude 和 Kiro 格式相同
+            # 可能是单个服务器配置
+            if "name" in config:
+                return "maibot_single"
+            return None
+
+        return None
+
+    @classmethod
+    def parse_json_safe(cls, json_str: str) -> Tuple[Optional[Any], Optional[str]]:
+        """安全解析 JSON 字符串
+        
+        Returns:
+            (解析结果, 错误信息)
+        """
+        if not json_str or not json_str.strip():
+            return None, "输入为空"
+
+        json_str = json_str.strip()
+
+        try:
+            return json.loads(json_str), None
+        except json.JSONDecodeError as e:
+            # 尝试提供更友好的错误信息
+            line = e.lineno
+            col = e.colno
+            return None, f"JSON 解析失败 (行 {line}, 列 {col}): {e.msg}"
+
+    @classmethod
+    def validate_server_config(cls, name: str, config: Dict[str, Any]) -> Tuple[bool, Optional[str], List[str]]:
+        """验证单个服务器配置
+        
+        Args:
+            name: 服务器名称
+            config: 服务器配置字典
+            
+        Returns:
+            (是否有效, 错误信息, 警告列表)
+        """
+        warnings = []
+
+        if not isinstance(config, dict):
+            return False, f"服务器 '{name}' 配置必须是对象", []
+
+        has_command = "command" in config
+        has_url = "url" in config
+
+        # 必须有 command 或 url 之一
+        if not has_command and not has_url:
+            return False, f"服务器 '{name}' 缺少 'command' 或 'url' 字段", []
+
+        # 同时有 command 和 url 时给出警告
+        if has_command and has_url:
+            warnings.append(f"'{name}': 同时存在 command 和 url，将优先使用 stdio 模式")
+
+        # 验证 url 格式
+        if has_url and not has_command:
+            url = config.get("url", "")
+            if not isinstance(url, str):
+                return False, f"服务器 '{name}' 的 url 必须是字符串", []
+            if not url.startswith(("http://", "https://")):
+                warnings.append(f"'{name}': url 不是标准 HTTP(S) 地址")
+
+        # 验证 command 格式
+        if has_command:
+            command = config.get("command", "")
+            if not isinstance(command, str):
+                return False, f"服务器 '{name}' 的 command 必须是字符串", []
+            if not command.strip():
+                return False, f"服务器 '{name}' 的 command 不能为空", []
+
+        # 验证 args 格式
+        if "args" in config:
+            args = config.get("args")
+            if not isinstance(args, list):
+                return False, f"服务器 '{name}' 的 args 必须是数组", []
+            for i, arg in enumerate(args):
+                if not isinstance(arg, str):
+                    warnings.append(f"'{name}': args[{i}] 不是字符串，将自动转换")
+
+        # 验证 env 格式
+        if "env" in config:
+            env = config.get("env")
+            if not isinstance(env, dict):
+                return False, f"服务器 '{name}' 的 env 必须是对象", []
+
+        # 验证 headers 格式
+        if "headers" in config:
+            headers = config.get("headers")
+            if not isinstance(headers, dict):
+                return False, f"服务器 '{name}' 的 headers 必须是对象", []
+
+        # 验证 transport/type 格式
+        transport_value = None
+        for field_name in cls.TRANSPORT_FIELD_NAMES:
+            if field_name in config:
+                transport_value = config.get(field_name, "").lower()
+                break
+        if transport_value and transport_value not in cls.TRANSPORT_MAP_IN:
+            warnings.append(f"'{name}': 未知的 transport 类型 '{transport_value}'，将自动推断")
+
+        return True, None, warnings
+
+    @classmethod
+    def convert_claude_server(cls, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
+        """将单个 Claude 格式服务器配置转换为 MaiBot 格式
+        
+        Args:
+            name: 服务器名称
+            config: Claude 格式的服务器配置
+            
+        Returns:
+            MaiBot 格式的服务器配置
+        """
+        result = {
+            "name": name,
+            "enabled": True,
+        }
+
+        has_command = "command" in config
+
+        if has_command:
+            # stdio 模式
+            result["transport"] = "stdio"
+            result["command"] = config.get("command", "")
+
+            # 处理 args
+            args = config.get("args", [])
+            if args:
+                # 确保所有 args 都是字符串
+                result["args"] = [str(arg) for arg in args]
+
+            # 处理 env
+            env = config.get("env", {})
+            if env and isinstance(env, dict):
+                result["env"] = env
+
+        else:
+            # 远程模式 (sse/http/streamable_http)
+            # 支持 transport 或 type 字段
+            transport_raw = None
+            for field_name in cls.TRANSPORT_FIELD_NAMES:
+                if field_name in config:
+                    transport_raw = config.get(field_name, "").lower()
+                    break
+            if not transport_raw:
+                transport_raw = "sse"
+            result["transport"] = cls.TRANSPORT_MAP_IN.get(transport_raw, "sse")
+            result["url"] = config.get("url", "")
+
+            # 处理 headers
+            headers = config.get("headers", {})
+            if headers and isinstance(headers, dict):
+                result["headers"] = headers
+
+        return result
+
+    @classmethod
+    def convert_maibot_server(cls, config: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
+        """将单个 MaiBot 格式服务器配置转换为 Claude 格式
+        
+        Args:
+            config: MaiBot 格式的服务器配置
+            
+        Returns:
+            (服务器名称, Claude 格式的服务器配置)
+        """
+        name = config.get("name", "unnamed")
+        result = {}
+
+        transport = config.get("transport", "stdio").lower()
+
+        if transport == "stdio":
+            # stdio 模式
+            result["command"] = config.get("command", "")
+
+            args = config.get("args", [])
+            if args:
+                result["args"] = args
+
+            env = config.get("env", {})
+            if env:
+                result["env"] = env
+
+        else:
+            # 远程模式
+            result["url"] = config.get("url", "")
+
+            # 转换 transport 名称
+            claude_transport = cls.TRANSPORT_MAP_OUT.get(transport, "sse")
+            if claude_transport != "sse":  # sse 是默认值，可以省略
+                result["transport"] = claude_transport
+
+            headers = config.get("headers", {})
+            if headers:
+                result["headers"] = headers
+
+        return name, result
+
+    @classmethod
+    def from_claude_format(
+        cls,
+        config: Dict[str, Any],
+        existing_names: Optional[set] = None
+    ) -> ConversionResult:
+        """从 Claude Desktop 格式转换为 MaiBot 格式
+        
+        Args:
+            config: Claude Desktop 配置 (包含 mcpServers 字段)
+            existing_names: 已存在的服务器名称集合，用于跳过重复
+            
+        Returns:
+            ConversionResult
+        """
+        result = ConversionResult(success=True)
+        existing_names = existing_names or set()
+
+        # 检查格式
+        if not isinstance(config, dict):
+            result.success = False
+            result.errors.append("配置必须是 JSON 对象")
+            return result
+
+        mcp_servers = config.get("mcpServers", {})
+
+        if not isinstance(mcp_servers, dict):
+            result.success = False
+            result.errors.append("mcpServers 必须是对象")
+            return result
+
+        if not mcp_servers:
+            result.warnings.append("mcpServers 为空，没有服务器可导入")
+            return result
+
+        # 转换每个服务器
+        for name, srv_config in mcp_servers.items():
+            # 检查名称是否已存在
+            if name in existing_names:
+                result.skipped.append(f"'{name}' (已存在)")
+                continue
+
+            # 验证配置
+            valid, error, warnings = cls.validate_server_config(name, srv_config)
+            result.warnings.extend(warnings)
+
+            if not valid:
+                result.errors.append(error)
+                continue
+
+            # 转换配置
+            try:
+                converted = cls.convert_claude_server(name, srv_config)
+                result.servers.append(converted)
+            except Exception as e:
+                result.errors.append(f"转换服务器 '{name}' 失败: {str(e)}")
+
+        # 如果有错误但也有成功的，仍然标记为成功（部分成功）
+        if result.errors and not result.servers:
+            result.success = False
+
+        return result
+
+    @classmethod
+    def to_claude_format(cls, servers: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """将 MaiBot 格式转换为 Claude Desktop 格式
+        
+        Args:
+            servers: MaiBot 格式的服务器列表
+            
+        Returns:
+            Claude Desktop 格式的配置
+        """
+        mcp_servers = {}
+
+        for srv in servers:
+            if not isinstance(srv, dict):
+                continue
+
+            name, config = cls.convert_maibot_server(srv)
+            mcp_servers[name] = config
+
+        return {"mcpServers": mcp_servers}
+
+    @classmethod
+    def import_from_string(
+        cls,
+        json_str: str,
+        existing_names: Optional[set] = None
+    ) -> ConversionResult:
+        """从 JSON 字符串导入配置
+        
+        自动检测格式并转换为 MaiBot 格式
+        
+        Args:
+            json_str: JSON 字符串
+            existing_names: 已存在的服务器名称集合
+            
+        Returns:
+            ConversionResult
+        """
+        result = ConversionResult(success=True)
+        existing_names = existing_names or set()
+
+        # 解析 JSON
+        parsed, error = cls.parse_json_safe(json_str)
+        if error:
+            result.success = False
+            result.errors.append(error)
+            return result
+
+        # 检测格式
+        fmt = cls.detect_format(parsed)
+
+        if fmt is None:
+            result.success = False
+            result.errors.append("无法识别的配置格式")
+            return result
+
+        if fmt == "maibot":
+            # 已经是 MaiBot 格式，直接验证并返回
+            for srv in parsed:
+                if not isinstance(srv, dict):
+                    result.warnings.append("跳过非对象元素")
+                    continue
+
+                name = srv.get("name", "")
+                if not name:
+                    result.warnings.append("跳过缺少 name 的服务器")
+                    continue
+
+                if name in existing_names:
+                    result.skipped.append(f"'{name}' (已存在)")
+                    continue
+
+                result.servers.append(srv)
+
+        elif fmt == "maibot_single":
+            # 单个 MaiBot 格式服务器
+            name = parsed.get("name", "")
+            if name in existing_names:
+                result.skipped.append(f"'{name}' (已存在)")
+            else:
+                result.servers.append(parsed)
+
+        elif fmt in ("claude", "kiro"):
+            # Claude/Kiro 格式
+            return cls.from_claude_format(parsed, existing_names)
+
+        return result
+
+    @classmethod
+    def export_to_string(
+        cls,
+        servers: List[Dict[str, Any]],
+        format_type: str = "claude",
+        pretty: bool = True
+    ) -> str:
+        """导出配置为 JSON 字符串
+        
+        Args:
+            servers: MaiBot 格式的服务器列表
+            format_type: 导出格式 ("claude", "kiro", "maibot")
+            pretty: 是否格式化输出
+            
+        Returns:
+            JSON 字符串
+        """
+        indent = 2 if pretty else None
+
+        if format_type in ("claude", "kiro"):
+            config = cls.to_claude_format(servers)
+        else:
+            config = servers
+
+        return json.dumps(config, ensure_ascii=False, indent=indent)