Python SDK
Python 核心能力
当前 client、session、context、extension、download、CDP、错误和日志 API。
Client 与区域
from lexmount import Lexmount
client = Lexmount(
region="<region-id>",
timeout=60.0,
)
print(client.region_info())
print(client.catalog_info())Client 从构造函数参数或环境变量读取凭证。区域目录发现是 best effort:目录或探测不可用时,会回退到配置的 base URL。
Sessions
创建参数
session = client.sessions.create(
browser_mode="normal",
window_size="1920,1080",
downloads={"enabled": True},
recording={"persistent": True},
)| 参数 | 用途 |
|---|---|
browser_mode | normal 表示 Chromium,light 表示 Moli |
context / context_description | 复用或创建持久化浏览器数据 |
extension_ids | 挂载已上传扩展 |
proxy / official_proxy | 使用自定义上游代理或 Lexmount 官方代理池 |
downloads | 开启或关闭 session 下载文件存储 |
recording | 开启或关闭持久化回放;Moli 会强制关闭 |
custom_image_id | 启动允许使用的 custom browser image |
window_size | 使用 width,height 设置 ACE 浏览器尺寸 |
enable_lightmount_layout | LightMount layout 集成,默认 True |
enable_lightmount_resource | 覆盖本次 session 的 layout resource 集成 |
async_create | 异步创建并等待 active,默认 True |
poll_interval_sec / poll_timeout_sec | 控制 active 状态轮询 |
official_proxy=True 不能和 proxy={...} 同时使用。Context、扩展、代理和 custom image 是否可用取决于项目方案。
详情、列表、targets 与关闭
current = client.sessions.get(session.id)
sessions = client.sessions.list(status="active")
print(sessions.pagination.total_count)
for item in sessions:
print(item.id, item.status, item.region_id, item.browser_type)
for target in client.sessions.list_targets(session.id):
print(target.title, target.url, target.inspect_url)
session.close()
# 只有 ID 时使用:
client.sessions.delete(session_id=session.id)下载文件
result = client.sessions.downloads.list(session.id)
print(result.summary["count"], result.summary["total_size"])
file_bytes = client.sessions.downloads.get(session.id, "<download-id>")
archive_bytes = client.sessions.downloads.archive(session.id)
deleted = client.sessions.downloads.delete(session.id)
print(deleted.deleted_count)存储下载文件当前默认开启。应按 quickstart 仓库示例配置远程浏览器下载目录,确保文件写入持久化下载路径。
Contexts
context = client.contexts.create(
description="Authenticated storefront",
metadata={"owner": "agent-a"},
)
session = client.sessions.create(
context={"id": context.id, "mode": "read_write"},
)
contexts = client.contexts.list(status="available", limit=20)
details = client.contexts.get(context.id)
forked = client.contexts.fork(context.id)
client.contexts.force_release(context.id)
client.contexts.delete(forked.id)Active read-write session 使用的 context 会被锁定。当前 fork 只复制离线的 available context;暂不支持 hot fork、copy-on-write 和 lineage tracking。
Extensions
extension = client.extensions.upload(
"/absolute/path/to/extension.zip",
name="demo-extension",
)
extensions = client.extensions.list(limit=20, offset=0)
details = client.extensions.get(extension.id)
session = client.sessions.create(extension_ids=[extension.id])
session.close()
client.extensions.delete(extension.id)服务端接受 .zip 和 .crx 包。Active session 正在使用的扩展不能删除。
CDP 辅助方法与集成认证
from lexmount import Lexmount, connect_over_cdp
client = Lexmount()
session = client.sessions.create()
cdp = connect_over_cdp(session)
print(cdp.send("Browser.getVersion"))
cdp.close()
session.close()
client.close()SDK 还导出 register_integrated_auth_callback。可选包 lexmount-extra 提供 NTLM 与 Kerberos callback factory。
错误与日志
from lexmount import (
APIError,
AuthenticationError,
ContextLockedError,
NetworkError,
TimeoutError,
set_log_level,
)
set_log_level("INFO")
try:
session = client.sessions.create()
except AuthenticationError:
print("Check API Key and Project ID")
except ContextLockedError as error:
print(error.active_session_id, error.retry_after)
except (NetworkError, TimeoutError):
print("Retry only after checking whether a session was reserved")
except APIError as error:
print(error.status_code)可选质量上报
质量上报默认关闭,也不会改变用户侧异常。只有项目已批准采集时才开启:
LEXMOUNT_QUALITY_REPORTING_ENABLED=true
LEXMOUNT_QUALITY_CDP_SLOW_THRESHOLD_MS=3000SDK 可以上报 session 创建失败和汇总。只有通过 SDK connect_over_cdp() 辅助方法发送命令时,才能采集单条 CDP 延迟事件。
Lexmount Browser