lexmount 0.5.2

Python SDK for Lexmount browser automation service

Lexmount Python SDK

English | 中文

---

English

Python SDK for Lexmount browser automation service.

✨ Features

• ⚡ Async-First Session Create: sessions.create() defaults to async create and returns once session is active
• 🚀 Easy to Use: Simple and intuitive API design
• 🔒 Type-Safe: Complete type hints for better IDE support
• 📝 Well Documented: Comprehensive docstrings and examples
• 🛡️ Production Ready: Error handling, logging, and retry support
• 📊 Session Management: Create, list, and delete browser sessions
• 📥 Session Downloads: List, fetch, archive, and clear session downloads
• 🔍 Advanced Querying: Pagination support and status filtering
• 🧰 CDP Utilities: Send raw Chrome DevTools Protocol commands and handle integrated auth

📦 Installation

pip install lexmount

Requirements:

• Python 3.8+
• httpx >= 0.23.0
• pychrome >= 0.2.4
• python-dotenv >= 0.19.0
• playwright (optional, for browser automation)

🛠️ Development Setup (For Running Examples)

If you want to run the example code from the repository, follow these steps:

1. Create Virtual Environment

# Navigate to project directory
cd /path/to/lexmount-python-sdk

# Create virtual environment using Python's built-in venv
python3 -m venv venv

2. Activate Virtual Environment

# Linux/macOS
source venv/bin/activate

# Windows
venv\Scripts\activate

3. Install SDK in Development Mode

# Install SDK and all dependencies (including dev dependencies)
pip install -e ".[dev]"

This installs the SDK in editable mode, so code changes take effect immediately without reinstalling.

4. Configure Environment Variables

Create a .env file in the project root directory:

LEXMOUNT_API_KEY=your-api-key-here
LEXMOUNT_PROJECT_ID=your-project-id-here
# Optional: for test environments
LEXMOUNT_BASE_URL=https://api.lexmount.cn

5. Run Examples

# From project root directory
python3 examples/context_basic.py
python3 examples/context_fork.py
python3 examples/session_list.py

# Or from examples directory
cd examples
python3 ./context_basic.py
python3 ./context_fork.py

Context fork quick example:

source = client.contexts.create()
forked = client.contexts.fork(source.id)
print(forked.id)

Current context fork limitations:

• Only supports offline snapshot copy from an available context
• Does not support hot fork from a locked context
• Does not support copy-on-write
• Does not support parent/child lineage tracking

Note: Each time you open a new terminal, you need to activate the virtual environment again:

source venv/bin/activate

🚀 Quick Start

Basic Usage

from lexmount import Lexmount

# Initialize client (credentials from environment variables)
client = Lexmount(
    api_key="your-api-key",
    project_id="your-project-id"
)

# Create a browser session (defaults to async create under the hood)
session = client.sessions.create()
print(f"Session ID: {session.id}")
print(f"Connect URL: {session.connect_url}")

# Close session when done
session.close()

Using Context Managers (Recommended)

Context managers provide automatic resource cleanup:

from lexmount import Lexmount

# Client and session will be automatically closed
with Lexmount(api_key="key", project_id="proj") as client:
    with client.sessions.create() as session:
        print(f"Using session: {session.id}")
        # Use session...
        # Automatically closed on exit

Integration with Playwright

from lexmount import Lexmount
from playwright.sync_api import sync_playwright

# Create session
client = Lexmount(api_key="key", project_id="proj")
session = client.sessions.create()

# Connect with Playwright
with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(session.connect_url)
    page = browser.new_page()

    page.goto("https://example.com")
    print(page.title())

    browser.close()

session.close()

Raw Chrome DevTools Protocol

from lexmount import Lexmount, connect_over_cdp

client = Lexmount(api_key="key", project_id="proj")
session = client.sessions.create()

cdp = connect_over_cdp(session)
version = cdp.send("Browser.getVersion")
print(version["product"])

cdp.close()
session.close()
client.close()

Integrated Authentication Callback

from lexmount import Lexmount, register_integrated_auth_callback

client = Lexmount(api_key="key", project_id="proj")
session = client.sessions.create()

def on_auth_required(params, sender):
    sender({
        "cancel": False,
        "scheme": "Negotiate",
        "token": "base64-token",
    })

register_integrated_auth_callback(session, on_auth_required)

📚 Core Features

Session Management

Create Session

# Basic creation
session = client.sessions.create()

# With specific browser mode
session = client.sessions.create(browser_mode="normal")

# With uploaded extensions
session = client.sessions.create(
    browser_mode="normal",
    extension_ids=["ext_11ff6ce20fb0"]
)

# With authenticated upstream proxy
session = client.sessions.create(
    proxy={
        "type": "external",
        "server": "http://10.3.6.54:28080",
        "username": "user",
        "password": "pass",
    }
)

# Available browser modes:
# - "normal": Chrome in Docker container
# - "light": Lightweight Chrome in Docker

Extension Management

# Upload extension archive
extension = client.extensions.upload("./test_extension.zip", name="demo-extension")

# List extensions
extensions = client.extensions.list(limit=10)
for item in extensions:
    print(item.id, item.name)

# Get extension details
extension = client.extensions.get(extension.id)

# Use extension when creating session
session = client.sessions.create(extension_ids=[extension.id])

List Sessions

# List all sessions with pagination
result = client.sessions.list()
print(f"Total: {result.pagination.total_count}")
print(f"Active: {result.pagination.active_count}")

for session in result.sessions:
    print(f"{session.id}: {session.status}")

# Filter by status (server-side filtering)
active_sessions = client.sessions.list(status="active")
print(f"Found {len(active_sessions)} active sessions")

Delete Session

# Delete by session ID
client.sessions.delete(session_id="session-id")

# Or use session object
session.close()

Session Downloads

# List downloads for a session
downloads = client.sessions.downloads.list(session.id)
print(downloads.summary["count"])

# Download a single file
file_bytes = client.sessions.downloads.get(session.id, "download-id")

# Download all files as zip
archive_bytes = client.sessions.downloads.archive(session.id)

# Clear session downloads
client.sessions.downloads.delete(session.id)

Error Handling

from lexmount import (
    Lexmount,
    AuthenticationError,
    SessionNotFoundError,
    TimeoutError,
    NetworkError,
    APIError
)

try:
    client = Lexmount(api_key="key", project_id="proj")
    session = client.sessions.create()
except AuthenticationError:
    print("Invalid credentials")
except TimeoutError:
    print("Request timed out")
except NetworkError:
    print("Network connection failed")
except APIError as e:
    print(f"API error: {e.status_code}")

Logging

from lexmount import set_log_level

# Enable debug logging
set_log_level("DEBUG")

# Available levels: DEBUG, INFO, WARNING, ERROR, CRITICAL
client = Lexmount(api_key="key", project_id="proj")
session = client.sessions.create()  # Will log detailed info

📖 API Reference

Lexmount Client

Lexmount(
    api_key: str = None,           # API key (or LEXMOUNT_API_KEY env var)
    project_id: str = None,        # Project ID (or LEXMOUNT_PROJECT_ID env var)
    base_url: str = None,          # API URL (or LEXMOUNT_BASE_URL env var)
    region: str = None,            # Optional region ID
    timeout: float = 60.0          # Request timeout in seconds
)

Region selection uses /v1/regions/catalog and /v1/region/probe when available. If catalog/probe is missing, unavailable, or invalid, the SDK falls back to the original base_url behavior. If base_url is provided, the SDK requests the catalog from that base_url and can still select a region.

Inspect the current region state:

info = client.region_info()
print(info["requested_region"], info["selected_region"], info["endpoint_base_url"])

Inspect the public endpoint catalog directly:

catalog = client.catalog_info()
print(catalog["available"], catalog["regions"])

Sessions Resource

sessions.create() - Create a new browser session (async by default)

• Default behavior: uses POST /instance/v2 and polls POST /instance/session until active
• Parameters:
  • project_id (str, optional): Override default project ID
  • browser_mode (str, optional): Browser mode (default: "normal")
  • context (dict, optional): Persistent context configuration
  • extension_ids (list[str], optional): Uploaded extension IDs to mount
  • weak_lock (bool, optional): Pass through weak lock flag when using context
  • async_create (bool, optional): Default True; set False to use legacy sync /instance create
  • poll_interval_sec (float, optional): Poll interval (seconds) when async create is enabled
  • poll_timeout_sec (float, optional): Max wait before raising TimeoutError
• Returns: SessionInfo object
• Raises: AuthenticationError, ValidationError, APIError, NetworkError, TimeoutError

Extensions Resource

extensions.upload() - Upload a zip/crx browser extension

• Parameters:
  • file_path (str, required): Local zip/crx path
  • name (str, optional): Display name
  • project_id (str, optional): Override default project ID
• Returns: ExtensionInfo

extensions.list() - List uploaded extensions

• Parameters:
  • project_id (str, optional): Override default project ID
  • limit (int, optional): Page size
  • offset (int, optional): Page offset
• Returns: list[ExtensionInfo]

extensions.get() - Get one extension

• Parameters:
  • extension_id (str, required): Extension ID
  • project_id (str, optional): Override default project ID
• Returns: ExtensionInfo

extensions.delete() - Delete one extension

• Parameters:
  • extension_id (str, required): Extension ID
• Returns: None

sessions.list() - List browser sessions with pagination

• Parameters:
  • project_id (str, optional): Override default project ID
  • status (str, optional): Filter by status ("active", "closed", or None for all)
• Returns: SessionListResponse with sessions and pagination info
• Raises: AuthenticationError, APIError, NetworkError, TimeoutError

sessions.delete() - Delete a browser session

• Parameters:
  • session_id (str, required): Session ID to delete
  • project_id (str, optional): Override default project ID
• Returns: None
• Raises: AuthenticationError, SessionNotFoundError, APIError

SessionInfo Object

Attributes:

• id (str): Session identifier
• session_id (str): Alias for id
• status (str): Session status ("active", "closed")
• browser_type (str): Browser mode used
• created_at (str): ISO 8601 timestamp
• connect_url (str): WebSocket URL for Playwright
• ws (str): Alias for connect_url
• container_id (str): Docker container ID (if applicable)
• inspect_url (str): Chrome DevTools inspection URL

Methods:

• close(): Close the session and release resources
• __enter__(), __exit__(): Context manager support

🔧 Configuration

Environment Variables

export LEXMOUNT_API_KEY="your-api-key"
export LEXMOUNT_PROJECT_ID="your-project-id"
export LEXMOUNT_BASE_URL="https://api.lexmount.com"  # Optional

Or use a .env file:

LEXMOUNT_API_KEY=your-api-key
LEXMOUNT_PROJECT_ID=your-project-id

📄 License

MIT License

---

中文

Lexmount 浏览器自动化服务的 Python SDK。

✨ 特性

• ⚡ 异步优先创建：sessions.create() 默认异步创建，直到会话 active 才返回
• 🚀 简单易用：直观的 API 设计
• 🔒 类型安全：完整的类型提示，IDE 支持友好
• 📝 文档完善：详细的文档和示例
• 🛡️ 生产就绪：错误处理、日志记录、重试支持
• 📊 会话管理：创建、列表、删除浏览器会话
• 🔍 高级查询：支持分页和状态过滤

📦 安装

pip install lexmount

依赖要求：

• Python 3.8+
• httpx >= 0.23.0
• python-dotenv >= 0.19.0
• playwright（可选，用于浏览器自动化）

🛠️ 开发环境设置（运行示例代码）

如果你想运行仓库中的示例代码，请按以下步骤操作：

1. 创建虚拟环境

# 进入项目目录
cd /path/to/lexmount-python-sdk

# 使用 Python 内置的 venv 创建虚拟环境
python3 -m venv venv

2. 激活虚拟环境

# Linux/macOS
source venv/bin/activate

# Windows
venv\Scripts\activate

3. 以开发模式安装 SDK

# 安装 SDK 和所有依赖（包括开发依赖）
pip install -e ".[dev]"

这会以可编辑模式安装 SDK，代码修改会立即生效，无需重新安装。

4. 配置环境变量

在项目根目录创建 .env 文件：

LEXMOUNT_API_KEY=your-api-key-here
LEXMOUNT_PROJECT_ID=your-project-id-here
# 可选：测试环境
LEXMOUNT_BASE_URL=https://api.lexmount.cn

5. 运行示例

# 从项目根目录运行
python3 examples/context_basic.py
python3 examples/context_fork.py
python3 examples/session_list.py

# 或从 examples 目录运行
cd examples
python3 ./context_basic.py
python3 ./context_fork.py

context fork 最小示例：

source = client.contexts.create()
forked = client.contexts.fork(source.id)
print(forked.id)

当前 context fork 第一版限制：

• 只支持对 available context 做离线快照复制
• 不支持 locked context 的热 fork
• 不支持 copy-on-write
• 不支持父子链追踪

注意： 每次新开终端时，需要重新激活虚拟环境：

source venv/bin/activate

🚀 快速开始

基础用法

from lexmount import Lexmount

# 初始化客户端（从环境变量读取凭据）
client = Lexmount(
    api_key="your-api-key",
    project_id="your-project-id"
)

# 创建浏览器会话（默认走异步创建）
session = client.sessions.create()
print(f"会话 ID: {session.id}")
print(f"连接 URL: {session.connect_url}")

# 使用完毕后关闭会话
session.close()

使用上下文管理器（推荐）

上下文管理器提供自动资源清理：

from lexmount import Lexmount

# 客户端和会话会自动关闭
with Lexmount(api_key="key", project_id="proj") as client:
    with client.sessions.create() as session:
        print(f"使用会话: {session.id}")
        # 使用会话...
        # 退出时自动关闭

与 Playwright 集成

from lexmount import Lexmount
from playwright.sync_api import sync_playwright

# 创建会话
client = Lexmount(api_key="key", project_id="proj")
session = client.sessions.create()

# 使用 Playwright 连接
with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(session.connect_url)
    page = browser.new_page()

    page.goto("https://example.com")
    print(page.title())

    browser.close()

session.close()

📚 核心功能

会话管理

创建会话

# 基础创建
session = client.sessions.create()

# 指定浏览器模式
session = client.sessions.create(browser_mode="normal")

# 挂载已上传的插件
session = client.sessions.create(
    browser_mode="normal",
    extension_ids=["ext_11ff6ce20fb0"]
)

# 可用的浏览器模式：
# - "normal": Docker 中的 Chrome
# - "light": Docker 中的轻量级 Chrome

插件管理

# 上传插件压缩包
extension = client.extensions.upload("./test_extension.zip", name="demo-extension")

# 查看插件列表
extensions = client.extensions.list(limit=10)
for item in extensions:
    print(item.id, item.name)

# 获取插件详情
extension = client.extensions.get(extension.id)

# 创建带插件的会话
session = client.sessions.create(extension_ids=[extension.id])

列出会话

# 列出所有会话（带分页信息）
result = client.sessions.list()
print(f"总计: {result.pagination.total_count}")
print(f"活跃: {result.pagination.active_count}")

for session in result.sessions:
    print(f"{session.id}: {session.status}")

# 按状态过滤（服务端过滤）
active_sessions = client.sessions.list(status="active")
print(f"找到 {len(active_sessions)} 个活跃会话")

删除会话

# 通过会话 ID 删除
client.sessions.delete(session_id="session-id")

# 或使用会话对象
session.close()

错误处理

from lexmount import (
    Lexmount,
    AuthenticationError,
    SessionNotFoundError,
    TimeoutError,
    NetworkError,
    APIError
)

try:
    client = Lexmount(api_key="key", project_id="proj")
    session = client.sessions.create()
except AuthenticationError:
    print("认证失败")
except TimeoutError:
    print("请求超时")
except NetworkError:
    print("网络连接失败")
except APIError as e:
    print(f"API 错误: {e.status_code}")

日志配置

from lexmount import set_log_level

# 启用调试日志
set_log_level("DEBUG")

# 可用级别：DEBUG, INFO, WARNING, ERROR, CRITICAL
client = Lexmount(api_key="key", project_id="proj")
session = client.sessions.create()  # 会输出详细日志

📖 API 参考

Lexmount 客户端

Lexmount(
    api_key: str = None,           # API 密钥（或 LEXMOUNT_API_KEY 环境变量）
    project_id: str = None,        # 项目 ID（或 LEXMOUNT_PROJECT_ID 环境变量）
    base_url: str = None,          # API 地址（或 LEXMOUNT_BASE_URL 环境变量）
    region: str = None,            # 可选 Region ID
    timeout: float = 60.0          # 请求超时时间（秒）
)

Region 选择会在可用时使用 /v1/regions/catalog 和 /v1/region/probe。 如果服务端还没有 catalog/probe、接口不可用或返回数据不合法，SDK 会退化为 原来的 base_url 逻辑。如果传入 base_url，SDK 会从这个 base_url 读取 catalog，并且仍然可以选择 region。

查看当前 region 状态：

info = client.region_info()
print(info["requested_region"], info["selected_region"], info["endpoint_base_url"])

直接查看 public endpoint catalog：

catalog = client.catalog_info()
print(catalog["available"], catalog["regions"])

Sessions 资源

sessions.create() - 创建新的浏览器会话（默认异步）

• 默认行为：使用 POST /instance/v2 创建并轮询 POST /instance/session 直到 active
• 参数：
  • project_id (str, 可选): 覆盖默认项目 ID
  • browser_mode (str, 可选): 浏览器模式（默认: "normal"）
  • context (dict, 可选): 持久化上下文配置
  • extension_ids (list[str], 可选): 要挂载的已上传插件 ID
  • weak_lock (bool, 可选): context 场景下透传 weak lock 标志
  • async_create (bool, 可选): 默认 True；设为 False 使用旧同步 /instance
  • poll_interval_sec (float, 可选): 异步创建轮询间隔（秒）
  • poll_timeout_sec (float, 可选): 最大等待时间，超时抛 TimeoutError
• 返回： SessionInfo 对象
• 异常： AuthenticationError, ValidationError, APIError, NetworkError, TimeoutError

sessions.list() - 列出浏览器会话（带分页）

• 参数：
  • project_id (str, 可选): 覆盖默认项目 ID
  • status (str, 可选): 按状态过滤（"active"、"closed" 或 None 表示全部）
• 返回： SessionListResponse，包含会话列表和分页信息
• 异常： AuthenticationError, APIError, NetworkError, TimeoutError

sessions.delete() - 删除浏览器会话

• 参数：
  • session_id (str, 必需): 要删除的会话 ID
  • project_id (str, 可选): 覆盖默认项目 ID
• 返回： None
• 异常： AuthenticationError, SessionNotFoundError, APIError

SessionInfo 对象

属性：

• id (str): 会话标识符
• session_id (str): id 的别名
• status (str): 会话状态（"active", "closed"）
• browser_type (str): 使用的浏览器模式
• created_at (str): ISO 8601 时间戳
• connect_url (str): Playwright 的 WebSocket URL
• ws (str): connect_url 的别名
• container_id (str): Docker 容器 ID（如适用）
• inspect_url (str): Chrome DevTools 检查 URL

方法：

• close(): 关闭会话并释放资源
• __enter__(), __exit__(): 上下文管理器支持

🔧 配置

环境变量

export LEXMOUNT_API_KEY="your-api-key"
export LEXMOUNT_PROJECT_ID="your-project-id"
export LEXMOUNT_BASE_URL="https://api.lexmount.com"  # 可选

或使用 .env 文件：

LEXMOUNT_API_KEY=your-api-key
LEXMOUNT_PROJECT_ID=your-project-id

📄 许可证

MIT License