MXLite SDK

MXLite SDK 是 MXLite 的官方 Python SDK，为服务器部署和系统管理提供了全面的编程接口。该 SDK 封装了 MXD 服务的核心功能，使开发者能够通过简洁的 API 实现系统安装、网络配置、文件传输和远程命令执行等操作。

特性

• 全异步 API 设计：支持高效的并发操作，同时提供同步接口以兼容不同的使用场景
• 自动平台适配：自动检测并使用适合当前平台的二进制文件
• 多平台支持：兼容 Linux、macOS 和 Windows 等主流操作系统
• 内置 MXD 服务管理：可以自动启动、监控和关闭 MXD 服务
• 远程 MXA 部署：通过 SSH 将 MXA 客户端部署到远程服务器，支持多种复杂的网络结构和模式（甚至支持堡垒机后的服务器的部署）
• 全面的系统部署工具：提供从系统安装到网络配置的完整工作流
• 代码类型提示：完善的类型注解支持，提高开发效率

安装

从 PyPI 安装

# 基本安装
pip install mxlite-sdk

# 包含远程部署功能的安装
pip install "mxlite-sdk[deploy]"

快速入门

mxlite-sdk 支持连接已启动的 mxd 服务，也支持后台直接运行一个 mxd 服务，默认为使用 MXDRunner 运行一个 mxd 服务并作为子进程使用

基本使用示例（内建服务）

import asyncio
from mxlite import MXLite

async def main():
    # 使用异步上下文管理器自动管理资源
    async with MXLite() as mxc:
        # 启动 MXD 服务，注意，程序会自动获取一个可用的端口，并在此端口提供服务
        mxc.start_mxd()
        
        # 获取主机列表
        hosts, status = await mxc.get_host_list()
        print(f"找到 {len(hosts.sessions)} 个主机")
        
        if hosts.sessions:
            # 在远程主机上执行命令
            host_id = hosts.sessions[0]  # 使用第一个主机
            result, status = await mxc.command_exec(host_id, "ls -la")
            task_id = result.task_id
            
            # 等待任务完成
            task_result = await mxc.until_task_complete(host_id, task_id)
            print(f"命令输出：\n{task_result.payload.payload.stdout}")

if __name__ == "__main__":
    asyncio.run(main())

连接到外部 MXD 服务

如果您已经有一个运行中的 MXD 服务，可以直接连接到它而不是启动新实例：

import asyncio
from mxlite import MXLite

async def main():
    # 创建连接到外部服务的客户端
    client = MXLite(
        host="server-ip-address",  # 外部服务器地址
        http_port=8080,           # MXD HTTP 端口
        token="your-token-here"   # 认证令牌
    )
    
    try:
        # 确保连接成功
        connected = await client.connect_mxd()
        if not connected:
            print("连接到 MXD 服务失败")
            return
        
        # 执行操作...
        hosts, status = await client.get_host_list()
        print(f"成功连接到 MXD 服务，发现 {len(hosts.sessions)} 个主机")
        
    finally:
        # 关闭客户端
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

使用高级配置启动 MXD

在部分情况下，您可能需要使用 Https、设置指定端口或预置设置一些安全性配置，您可以传入 MXLiteConfig 实例

import asyncio
from mxlite import MXLite, MXLiteConfig

async def main():
    # 创建配置对象
    config = MXLiteConfig(
        root_dir="/path/to/certificates",  # 证书文件所在目录
        http_port=8080,                    # HTTP 端口
        https_port=8443,                   # HTTPS 端口
        token="your-token",                # 认证令牌
        verbose=True                       # 启用详细日志
    )

    async with MXLite(config) as mxc:
        # 启动 MXD 服务，注意，程序会自动获取一个可用的端口，并在此端口提供服务
        mxc.start_mxd()
        
        # 获取主机列表
        hosts, status = await mxc.get_host_list()
        print(f"找到 {len(hosts.sessions)} 个主机")
        
        if hosts.sessions:
            # 在远程主机上执行命令
            host_id = hosts.sessions[0]  # 使用第一个主机
            result, status = await mxc.command_exec(host_id, "ls -la")
            task_id = result.task_id
            
            # 等待任务完成
            task_result = await mxc.until_task_complete(host_id, task_id)
            print(f"命令输出：\n{task_result.payload.payload.stdout}")

if __name__ == "__main__":
    asyncio.run(main())

使用MXADeployer部署MXA到远程服务器

考虑到部分用户的网络环境较为复杂，或存在已经安装好的系统，我们提供了一个 MXADeployer 类来帮助用户将 MXA 部署到远程服务器。MXADeployer 类封装了 SSH 连接、MXA 部署和运行的逻辑，简化了部署过程。目前支持三种连接模式：

1. 自动发现模式：MXA自动扫描网络寻找MXD
2. 直连模式(使用端口转发)：MXA通过WebSocket连接到通过端口转发的MXD
3. 直连模式(指定地址)：MXA通过WebSocket直接连接到指定URL

注意：需要安装带deploy扩展的包 pip install "mxlite-sdk[deploy]"

自动发现模式部署示例

在自动发现模式下，MXA会自动扫描网络查找MXD服务，适用于在同一子网内运行MXD（部署服务器）和MXA（被部署服务器）的情况。

import asyncio
import logging
from mxlite import MXADeployer, MXLite

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def deploy_discovery_mode(ssh_host, ssh_user, ssh_password):
    mxc = MXLite()
    mxc.start_mxd()
    mxd_port = mxc.config.http_port
    logger.info(f"MXD已启动, 监听端口: {mxd_port}")
    try:
        # 使用异步上下文管理器自动管理SSH连接
        async with MXADeployer(ssh_host, ssh_user, ssh_password) as deployer:
            # 部署MXA
            success = await deployer.deploy_mxa()
            if not success:
                print("MXA部署失败")
                return

            # 使用自动发现模式运行MXA (默认模式)
            success, _ = await deployer.systemd_run_mxa(discovery_mode=True)
            if success:
                print("MXA服务已启动，使用自动发现模式")
            else:
                print("MXA服务启动失败")
            
            # 等待连接建立
            host_list, _ = await mxc.get_host_list()
            max_attempts = 120  # 最多等待600秒
            attempts = 0
            
            while not host_list.sessions and attempts < max_attempts:
                attempts += 1
                logger.info(f"等待主机列表... (尝试 {attempts}/{max_attempts})")
                host_list, _ = await mxc.get_host_list()
                if host_list.sessions:
                    break
                await asyncio.sleep(5)
            
            if not host_list.sessions:
                logger.warning("等待超时, 未检测到连接的主机")
                return
                
            # 获取并显示主机信息
            host_id = host_list.sessions[0]
            logger.info(f"主机已连接, ID: {host_id}")
            host_info, _ = await mxc.get_host_info(host_id)
            logger.info(f"主机信息: {host_info}")
            
            # 在这里添加您的业务逻辑
            # 例如执行命令、传输文件等
            result, _ = await mxc.command_exec(host_id, "uname -a")
            task_id = result.task_id
            task_result = await mxc.until_task_complete(host_id, task_id)
            print(f"命令输出: {task_result.payload.payload.stdout}")
    except Exception as e:
        print(f"部署过程中出错: {e}")
    finally:
        await deployer.systemd_remove_mxa()
        logger.info("MXA服务已成功移除")

        # 关闭MXD服务
        await mxc.close()
        logger.info("MXD已停止")
        # 关闭SSH连接
        await deployer.close()
        logger.info("SSH连接已关闭")


if __name__ == "__main__":    
    asyncio.run(deploy_discovery_mode(
        ssh_host="192.168.1.100",  # 远程主机地址
        ssh_user="ubuntu",         # SSH用户名
        ssh_password="password"    # SSH密码
    ))

直连模式（使用端口转发）部署示例

在直连模式下，MXA会通过WebSocket直接连接到MXD。这种方式适用于MXD与MXA在不同网络，或者需要避免自动发现的情况。

import asyncio
import logging
from mxlite import MXLite, MXADeployer

# 设置日志级别
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def deploy_port_forward(ssh_host, ssh_user, ssh_password):
    # 创建本地MXD服务
    mxc = MXLite()
    mxc.start_mxd()
    mxd_port = mxc.config.http_port
    logger.info(f"MXD已启动, 监听端口: {mxd_port}")

    try:
        # 使用异步上下文管理器自动管理SSH连接
        async with MXADeployer(ssh_host, ssh_user, ssh_password) as deployer:
            # 部署MXA
            success = await deployer.deploy_mxa()
            if not success:
                logger.error("MXA部署失败")
                return

            # 使用直连模式运行MXA, 并建立端口转发
            success, forwarded_port = await deployer.systemd_run_mxa(
                discovery_mode=False,
                local_mxd_port=mxd_port
            )
            
            if not success or forwarded_port is None:
                logger.error("MXA服务启动失败")
                return
                
            logger.info(f"MXA服务已启动, 远程端口 {forwarded_port} 已转发到本地 {mxd_port}")
            
            # 等待连接建立
            host_list, _ = await mxc.get_host_list()
            max_attempts = 120  # 最多等待600秒
            attempts = 0
            
            while not host_list.sessions and attempts < max_attempts:
                attempts += 1
                logger.info(f"等待主机列表... (尝试 {attempts}/{max_attempts})")
                host_list, _ = await mxc.get_host_list()
                if host_list.sessions:
                    break
                await asyncio.sleep(5)
            
            if not host_list.sessions:
                logger.warning("等待超时, 未检测到连接的主机")
                return
                
            # 获取并显示主机信息
            host_id = host_list.sessions[0]
            logger.info(f"主机已连接, ID: {host_id}")
            host_info, _ = await mxc.get_host_info(host_id)
            logger.info(f"主机信息: {host_info}")
            
            # 在这里添加您的业务逻辑
            # 例如执行命令、传输文件等
            result, _ = await mxc.command_exec(host_id, "uname -a")
            task_id = result.task_id
            task_result = await mxc.until_task_complete(host_id, task_id)
            print(f"命令输出: {task_result.payload.payload.stdout}")
            
    finally:
        await deployer.systemd_remove_mxa()
        logger.info("MXA服务已成功移除")

        # 关闭MXD服务
        await mxc.close()
        logger.info("MXD已停止")
        
        # 关闭SSH连接
        await deployer.close()
        logger.info("SSH连接已关闭")


if __name__ == "__main__":
    asyncio.run(deploy_port_forward(
        ssh_host="192.168.1.100",  # 远程主机地址
        ssh_user="ubuntu",         # SSH用户名
        ssh_password="password"    # SSH密码
    ))

直连模式（指定WebSocket URL）部署示例

当MXD服务器已经存在并可以从远程访问时，可以直接指定WebSocket URL。

import asyncio
import logging
from mxlite import MXADeployer, MXLite

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def deploy_with_direct_url(ssh_host, ssh_user, ssh_password):
    ws_url = "ws://mxd-server.example.com:8080/ws"  # 指定的WebSocket URL
    
    mxc = MXLite()
    mxc.start_mxd()
    mxd_port = mxc.config.http_port
    logger.info(f"MXD已启动, 监听端口: {mxd_port}")

    try:
        # 使用异步上下文管理器自动管理SSH连接
        async with MXADeployer(ssh_host, ssh_user, ssh_password) as deployer:
            # 检测远程主机架构
            arch = await deployer.detect_arch()
            print(f"检测到架构: {arch}")
            
            # 部署MXA
            success = await deployer.deploy_mxa()
            if not success:
                print("MXA部署失败")
                return
                
            # 使用指定WebSocket URL运行MXA
            success, _ = await deployer.systemd_run_mxa(ws_url=ws_url)
            if success:
                print(f"MXA服务已启动，直接连接到: {ws_url}")
            else:
                print("MXA服务启动失败")
            
            # 等待连接建立
            host_list, _ = await mxc.get_host_list()
            max_attempts = 120  # 最多等待600秒
            attempts = 0
            
            while not host_list.sessions and attempts < max_attempts:
                attempts += 1
                logger.info(f"等待主机列表... (尝试 {attempts}/{max_attempts})")
                host_list, _ = await mxc.get_host_list()
                if host_list.sessions:
                    break
                await asyncio.sleep(5)
            
            if not host_list.sessions:
                logger.warning("等待超时, 未检测到连接的主机")
                return
                
            # 获取并显示主机信息
            host_id = host_list.sessions[0]
            logger.info(f"主机已连接, ID: {host_id}")
            host_info, _ = await mxc.get_host_info(host_id)
            logger.info(f"主机信息: {host_info}")
            
            # 在这里添加您的业务逻辑
            # 例如执行命令、传输文件等
            result, _ = await mxc.command_exec(host_id, "uname -a")
            task_id = result.task_id
            task_result = await mxc.until_task_complete(host_id, task_id)
            print(f"命令输出: {task_result.payload.payload.stdout}")

    except Exception as e:
        print(f"部署过程中出错: {e}")
    finally:
        await deployer.systemd_remove_mxa()
        logger.info("MXA服务已成功移除")

        # 关闭MXD服务
        await mxc.close()
        logger.info("MXD已停止")
        
        # 关闭SSH连接
        await deployer.close()
        logger.info("SSH连接已关闭")


if __name__ == "__main__":
    asyncio.run(deploy_with_direct_url(
        ssh_host="192.168.1.100",  # 远程主机地址
        ssh_user="ubuntu",         # SSH用户名
        ssh_password="password"    # SSH密码
    ))

API 文档

MXLiteConfig

配置类，用于设置 MXLite 客户端参数。

config = MXLiteConfig(
    root_dir=None,         # 根目录，一般为证书文件所在的父级目录，默认为当前目录
    http_port=None,        # HTTP 端口，None 则随机选择
    https_port=None,       # HTTPS 端口，None 则随机选择
    token=None,            # 认证令牌
    certificates_dir=None, # 证书目录
    verbose=False,         # 是否输出详细日志
    host="127.0.0.1"       # 主机地址
)

MXLite

MXLite 客户端类，提供 MXD 服务管理和 API 操作。

# 创建客户端
mxlite = MXLite(
    config=None,             # MXLiteConfig 配置对象
    host=None,               # 外部 MXD 服务主机地址
    http_port=None,          # 外部 MXD 服务 HTTP 端口
    token=None,              # 外部 MXD 服务认证令牌
    auto_connect=True        # 是否自动连接到外部 MXD 服务
)

# 启动和关闭 MXD 服务
mxlite.start_mxd()
mxlite.kill_mxd()

# 连接与断开外部 MXD 服务
mxlite.connect_mxd()
mxlite.disconnect_mxd()

# 主机管理
hosts, status = await mxlite.get_host_list()  # 获取主机列表
host_info, status = await mxlite.get_host_info(host_id)  # 获取主机信息
host_list_info, status = await mxlite.get_host_list_info()  # 获取主机列表详细信息

# 任务管理
task_result, status = await mxlite.get_task_result(host_id, task_id)  # 获取任务结果
result = await mxlite.until_task_complete(host_id, task_id, interval=1)  # 等待任务完成

# 命令执行
result, status = await mxlite.command_exec(host_id, command)  # 在远程主机执行命令

# 文件操作
await mxlite.upload_file(host_id, src_path, target_url)  # 上传文件
await mxlite.download_file(host_id, src_url, target_path)  # 下载文件
await mxlite.add_file_map(file, publish_name)  # 添加文件映射
await mxlite.add_dir_map(dirname, publish_name)  # 添加目录映射
await mxlite.remove_file_map(file)  # 移除文件映射
maps, _ = await mxlite.get_file_map()  # 获取文件映射列表

# 文件系统操作
files, status = await mxlite.lsdir(path)  # 列出目录内容
content, status = await mxlite.read_file(path, max_size)  # 读取文件内容
hash_value = await mxlite.get_file_hash(file, algorithm)  # 获取文件哈希值

# 资源释放
await mxlite.close()  # 异步方式关闭
mxlite.close_sync()  # 同步方式关闭

MXADeployer

MXA部署类，提供通过SSH将MXA部署到远程服务器的能力。

# 创建MXA部署器
deployer = MXADeployer(
    ssh_host,      # SSH主机地址
    ssh_user,      # SSH用户名
    ssh_password,  # SSH密码 (与ssh_key二选一)
    ssh_key=None   # SSH密钥文件路径 (与ssh_password二选一)
)

# 检测远程主机架构
arch = await deployer.detect_arch()

# 检测是否有sudo权限
has_sudo = await deployer.detect_sudo()

# 部署MXA到远程主机
success = await deployer.deploy_mxa()

# 使用systemd运行MXA - 支持三种模式
# 1. 自动发现模式
success, _ = await deployer.systemd_run_mxa(discovery_mode=True)

# 2. 直连模式(使用端口转发)
success, forwarded_port = await deployer.systemd_run_mxa(
    discovery_mode=False,  
    local_mxd_port=8080    # 本地MXD的HTTP端口
)

# 3. 直连模式(指定WebSocket URL)
success, _ = await deployer.systemd_run_mxa(
    ws_url="ws://mxd-server.example.com:8080/ws"  # 指定的WebSocket URL
)

# 停止并删除MXA服务
success = await deployer.systemd_remove_mxa()

# 管理端口转发
listener = await deployer.create_port_forward(local_port, remote_host, remote_port)
success = await deployer.stop_port_forward(local_port, remote_host, remote_port)
await deployer.stop_all_forwards()

# 关闭连接
await deployer.close()

在本地开发

构建平台特定的 wheel 包

# 准备二进制文件
# 在 mxlite/bin 目录中放置适当的可执行文件，例如：
# - mxd-linux-x86_64, mxa-linux-x86_64         # Linux x86_64
# - mxd-macos-arm64, mxa-macos-arm64           # macOS ARM64
# - mxd-windows-x86_64.exe, mxa-windows-x86_64.exe  # Windows x86_64

# 构建 wheel
python setup.py bdist_wheel

系统要求

• Python 3.10+
• 基本依赖包：
  • aiohttp >= 3.11.18
  • pydantic >= 2.10.0
• 部署功能依赖（仅当使用 [deploy] 扩展时必需）：
  • asyncssh >= 2.14.0

许可证

本项目采用 Apache 2.0 许可证。详情请参见 LICENSE 文件。

相关链接

• MXLite 项目 - MXLite 的主要仓库
• 问题反馈 - 提交问题和反馈