Python 核心能力

当前 client、session、context、extension、download、CDP、Browser Agent、错误和日志 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`	选择本地代理、外部代理或官方代理池
`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 集成
`enable_captcha`	为当前会话开启自动验证码处理，默认关闭
`async_create`	异步创建并等待 `active`，默认 `True`
`poll_interval_sec` / `poll_timeout_sec`	控制 active 状态轮询

official_proxy=True 不能和 proxy={...} 同时使用。Context、扩展、代理和 custom image 是否可用取决于项目方案。

自动处理验证码

使用 SDK 0.6.1 或更新版本，在创建会话时显式传入 enable_captcha：

with client.sessions.create(
    browser_mode="normal",
    enable_captcha=True,
) as session:
    print(session.id)

默认值是 False。此设置仅作用于新建的当前会话，同一项目的其他会话可以保持关闭。SDK 在同步和异步创建请求中都会发送顶层 enable_captcha 布尔字段。

服务端会为已开启且符合项目权益的会话自动检测、处理验证码。可用性和使用量受项目套餐及剩余验证码额度约束；客户端继续使用原有 SDK 或 CDP 控制浏览器。

会话生命周期介绍控制台中的对应开关。

详情、列表、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 仓库示例配置远程浏览器下载目录，确保文件写入持久化下载路径。

本地代理隧道

SDK 0.6.0 起新增 client.tunnels.open()；建议使用 0.6.1 或更高版本。隧道就绪后，以 proxy={"type": "local", "tunnel_id": tunnel.id} 创建 normal 会话，让云浏览器使用 SDK 电脑的网络与 DNS。需要 custom_proxy 权限，同项目/Key/区域，且与官方或外部代理互斥。先关会话，再关隧道；完整示例及选项见本地代理指南。

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=3000

SDK 可以上报 session 创建失败和汇总。只有通过 SDK connect_over_cdp() 辅助方法发送命令时，才能采集单条 CDP 延迟事件。

Browser Agent

Browser Agent 根据自然语言指令执行浏览器任务。Run 是一次执行；保存的 Agent 包含可复用指令及可选输出结构。SDK 0.7.0 通过 client.browser_agent 提供这些能力。

配置

pip install lexmount==0.7.0
export LEXMOUNT_API_KEY='<your-api-key>'
export LEXMOUNT_PROJECT_ID='<your-project-id>'
# 可选：SDK 0.7.0 默认使用全球 COM 地址。
export LEXMOUNT_BASE_URL=https://api.lexmount.com

API Key 和 Project ID 必须属于同一项目，并在所选环境启用 Browser Agent。填写 HTTPS 源地址，不要附加 /v1/browser-agent，SDK 会添加路径。中国区请显式设置 https://api.lexmount.cn。调用这些 Agent API 无需安装 Playwright。

执行任务

运行示例前，为一次逻辑任务选择并持久保存 RUN_IDEMPOTENCY_KEY。恢复这次任务时复用同一个键和相同请求；只有明确创建新任务时才使用新键。

export RUN_IDEMPOTENCY_KEY='<persistent-unique-key-for-this-task>'

保存为 agent.py，运行 python agent.py：

import os
from lexmount import APIError, Lexmount, RunWaitTimeout

key = os.environ["RUN_IDEMPOTENCY_KEY"]
try:
    with Lexmount() as client:
        receipt = client.browser_agent.runs.create(
            query="Read the page title and summarize the page.",
            start_url="https://example.com/",
            idempotency_key=key,
        )
        print("Accepted:", receipt.id)  # Save this Run ID before waiting.
        run = client.browser_agent.runs.wait(receipt.id, timeout=300)
        print(run.id, run.status, run.output)
        if run.status != "succeeded":
            print(run.stop_reason, run.error)
            raise SystemExit(1)
except RunWaitTimeout as error:
    print("Resume with runs.get/wait:", error.run_id)
    raise SystemExit(1)
except APIError as error:
    print(error.status_code, error.code, error.request_id)
    raise SystemExit(1)

创建响应只是已排队回执，不表示任务完成。保存 receipt.id，进程重启后通过 runs.get(id) 或 runs.wait(id) 恢复查询。关闭 SDK 客户端不会取消正在运行的任务。

复用已保存的 Agent

另外持久保存两个键：AGENT_IDEMPOTENCY_KEY 用于创建 Agent，SAVED_RUN_IDEMPOTENCY_KEY 用于本次 Run。保存返回的 Agent ID，后续 Run 直接使用它；创建 Agent 是一次性设置步骤。

import os
from lexmount import Lexmount

with Lexmount() as client:
    agent = client.browser_agent.agents.create(
        name="Page summary",
        instruction="Summarize the page in {{language}}.",
        start_url="https://example.com/",
        output_schema={
            "type": "object",
            "properties": {"summary": {"type": "string"}},
            "required": ["summary"],
            "additionalProperties": False,
        },
        idempotency_key=os.environ["AGENT_IDEMPOTENCY_KEY"],
    )
    receipt = client.browser_agent.runs.create(
        agent_id=agent.id,
        variables={"language": "English"},
        metadata={"source": "docs"},
        idempotency_key=os.environ["SAVED_RUN_IDEMPOTENCY_KEY"],
    )
    print("Agent:", agent.id, "Run:", receipt.id)
    # Recover this Run later with runs.get(receipt.id) or runs.wait(receipt.id).

变量必须与指令模板中的变量完全匹配。Run 被接受后会冻结 Agent 版本、展开的指令和输出结构，后续编辑 Agent 不影响已接受的 Run。metadata 返回给应用，不发送给模型。使用结构定义时，成功的输出是对象；自由输出是文本。

管理资源

以下方法均位于 client.browser_agent。使用 AsyncLexmount 时，每次调用都使用 await。

操作	调用
读取 / 列出 Agent	`agents.get(id)` / `agents.list()`
编辑 Agent	`agents.update(id, revision=current.revision, name="Updated name")`
删除 Agent	`agents.delete(id)`
读取 / 列出 Run	`runs.get(id)` / `runs.list()`
请求取消	`runs.cancel(id)`
等待终态	`runs.wait(id)`

编辑前先读取 Agent 并传入当前 revision；版本冲突时重新读取，更新操作不会自动重试。取消与完成可能同时发生，应检查返回状态；若仍为 cancelling，继续查询直到终态。

恢复与错误处理

未提供幂等键时 SDK 会自动生成。需要跨进程恢复时，请自行持久保存并传入：关键字参数 idempotency_key。发送前持久保存键和请求体；键使用 1–128 个不含空格的可见 ASCII 字符。相同键和请求体至少七天内返回原回执，不同请求体会冲突。超过该窗口后，先核对历史记录再决定是否新建。
succeeded、failed、timed_out、cancelled 都是 Run 终态。HTTP 成功不代表任务成功，应检查 status、stop_reason、error 和 output。
RunWaitTimeout 仅停止本地轮询。用已知 Run ID 继续查询，或显式请求取消。timeout、poll_interval、request_timeout 单位为秒。
APIError 包含 status_code、code、request_id，排查时保留请求 ID。传入 with_response=True 可读取成功响应的元信息。
创建 Run 时使用的 API Key 应保持有效，直到执行和清理完成。撤销它可能让活动 Run 以 credential_unavailable 失败；更换客户端 Key 不会替换原 Run 的执行凭证。
计量尚未确认时，usage.credits 可能为 None，不表示零费用。执行可能产生外部操作，结果未知时不要盲目重新提交。

异步客户端

使用 AsyncLexmount 访问相同资源。此例读取已有 Run，请将 RUN_ID 设置为之前保存的 ID。

import asyncio
import os
from lexmount import AsyncLexmount

async def main():
    async with AsyncLexmount() as client:
        run = await client.browser_agent.runs.get(os.environ["RUN_ID"])
        print(run.id, run.status, run.output)

asyncio.run(main())

可运行示例覆盖基础执行、已保存 Agent 和恢复流程。