模块 nonebot_plugin_marshoai.extensions.mcp_extension.config

class mcpConfig(BaseModel)

---

@model_validator(mode='after')

func validate_config(self) -> Self

源代码 或 在GitHub上查看

@model_validator(mode='after')
def validate_config(self) -> Self:
    srv_type = self.type
    command = self.command
    url = self.url
    if srv_type == 'stdio':
        if not command:
            raise ValueError("当 type 为 'stdio' 时，command 字段必须存在")
        elif not shutil.which(command):
            raise ValueError(f"命令 '{command}' 不存在或不可执行。")
    elif srv_type in ['sse', 'streamable_http'] and (not url):
        raise ValueError(f"当 type 为 '{srv_type}' 时，url 字段必须存在")
    return self

attr command: str = Field(default='')

attr args: list[str] = Field(default_factory=list)

attr env: dict[str, Any] = Field(default_factory=dict)

attr headers: dict[str, Any] = Field(default_factory=dict)

attr type: Literal['stdio', 'sse', 'streamable_http'] = Field(default='stdio')

attr url: str = Field(default='')

---

func get_mcp_server_config() -> dict[str, mcpConfig]

说明: 从 MCP 配置文件 config/mcp.json 中获取 MCP Server 配置

源代码 或 在GitHub上查看

def get_mcp_server_config() -> dict[str, mcpConfig]:
    if not mcp_config_file_path.exists():
        return {}
    try:
        with open(mcp_config_file_path, 'r', encoding='utf-8') as f:
            configs = json.load(f) or {}
    except (json.JSONDecodeError, IOError, OSError) as e:
        raise RuntimeError(f'读取 MCP 配置文件时发生错误: {e}')
    if not isinstance(configs, dict):
        raise TypeError('非预期的 MCP 配置文件格式')
    mcp_servers = configs.get('mcpServers', {})
    if not isinstance(mcp_servers, dict):
        raise TypeError('非预期的 MCP 配置文件格式')
    mcp_config: dict[str, mcpConfig] = {}
    for name, srv_config in mcp_servers.items():
        try:
            mcp_config[name] = mcpConfig(**srv_config)
        except (ValidationError, TypeError) as e:
            logger.warning(f"无效的MCP服务器配置 '{name}': {e}")
            continue
    return mcp_config