Install
openclaw skills install @thcjp/bsession-tool-proopenclaw skills install @thcjp/bsession-tool-pro定时任务+会话持久化+Webhook通知+批量管理+监控告警。企业级会话管理全功能覆盖。
将复杂的浏览器会话管理任务交给专业工具处理。专业版在免费版单次抓取能力之上,新增定时任务、会话持久化、Webhook通知、批量会话管理、企业级监控告警、Cloudflare自动规避、会话调试增强七大高级能力,满足企业级场景对浏览器会话的可靠性、可观测性与可扩展性要求.
| 能力维度 | 免费版 | 专业版 |
|---|---|---|
| 单次抓取(fetch) | 支持 | 支持 |
| 定时任务(recurring) | 不支持 | 支持(cron调度) |
| 会话持久化 | 不支持 | 支持(conf+py脚本) |
| Webhook通知 | 不支持 | 支持(n8n/飞书/Slack/邮件) |
| 批量会话管理 | 不支持 | 支持(多会话并发) |
| 监控告警 | 不支持 | 支持(健康检查+失败告警) |
| Cloudflare规避 | 不支持 | 支持(自动识别+等待) |
| 调试模式 | 基础 | 增强(断点+详细日志) |
| 技术支持 | 社区 | 优先工单响应 |
详细代码示例已移至
references/detail.md
处理: 解析定时任务(recurring循环执行)的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回定时任务(recurring循环执行)的响应数据,包含状态码、结果和日志.
处理: 解析会话持久化(保存为可复用脚本)的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回会话持久化(保存为可复用脚本)的响应数据,包含状态码、结果和日志.
处理: 解析Webhook通知(多渠道)的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回Webhook通知(多渠道)的响应数据,包含状态码、结果和日志.
处理: 解析批量会话管理的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回批量会话管理的响应数据,包含状态码、结果和日志.
| 参数名 | 类型 | 必填 | 说明 |
|---|---|---|---|
| input | string | 是 | 浏览器会话(专业版)处理的输入数据或指令 |
| options | object | 否 | 附加配置选项,如模式选择、格式偏好等 |
| callback_url | string | 否 | 异步处理完成后的回调通知URL |
class SessionMonitor:
"""会话监控告警(专业版)"""
# ...
def __init__(self, check_interval=300, max_failures=3):
self.check_interval = check_interval
self.max_failures = max_failures
self.failure_counts = {}
self.notifier = WebhookNotifier()
# ...
def check_health(self, session_name):
"""检查会话健康状态"""
try:
result = subprocess.run(
["docker", "exec", "agent-browser", "bsession", "show", session_name],
capture_output=True, text=True, timeout=10
)
if result.returncode == 0:
output = result.stdout
if "running" in output:
return {"status": "healthy", "details": output}
elif "stopped" in output:
return {"status": "stopped", "details": output}
else:
return {"status": "unknown", "details": output}
return {"status": "error", "error": result.stderr}
except Exception as e:
return {"status": "error", "error": str(e)}
# ...
def monitor_loop(self, sessions):
"""监控循环"""
while True:
for name in sessions:
health = self.check_health(name)
if health["status"] != "healthy":
self.failure_counts[name] = self.failure_counts.get(name, 0) + 1
if self.failure_counts[name] >= self.max_failures:
"slack-alerts",
"会话异常告警",
f"会话 {name} 连续失败 {self.failure_counts[name]} 次\n状态:{health.get('status')}\n详情:{health.get('error', '')[:200]}"
)
_restart_session(name)
else:
failure_counts[name] = 0
# ...
time.sleep(self.check_interval)
# ...
def _restart_session(self, name):
"""重启会话"""
print(f"尝试重启会话:{name}")
try:
subprocess.run(
["docker", "exec", "agent-browser", "bsession", "restart", name],
capture_output=True, text=True, timeout=30
)
except Exception as e:
print(f"重启失败:{e}")
# ...
monitor = SessionMonitor(check_interval=300, max_failures=3)
monitor.notifier.register("slack-alerts", "https://hooks.slack.com/services/未指定", "slack")
处理: 解析企业级监控告警的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回企业级监控告警的响应数据,包含状态码、结果和日志. 能力覆盖范围:本skill的核心能力覆盖以下场景关键词:企业级浏览器会话、含定时任务、批量会话管理与监、浏览器会话助手专、业版是面向企业级、场景的完整浏览器、会话管理工具、在免费版单次抓取、能力之上、新增定时任务、Cloudflare、自动规避、会话调试增强七大、高级能力、Use、when、需要项目管理、任务规划、进度跟踪、团队协作时使用、不适用于实际人员、绩效评估等。这些关键词对应description中声明的使用场景,均已在上述能力点中提供对应的操作支持.
input_params参数,支持创建/查询/导出操作场景描述:监控竞争对手官网价格变化,变化时通知团队.
session = RecurringSession(
name="competitor-price",
interval=3600, # 每小时检查
webhook_url="https://hooks.slack.com/services/未指定"
)
session.start()
场景描述:每天凌晨2点巡检多个内部系统,发现异常立即告警.
manager = BatchSessionManager(max_workers=5)
manager.register_session("api-health", {"timeout": 60})
manager.register_session("db-status", {"timeout": 90})
manager.register_session("cache-monitor", {"timeout": 30})
manager.register_session("log-scanner", {"timeout": 120})
# ...
import schedule
schedule.every().day.at("02:00").do(manager.run_all)
场景描述:抓取需要登录的目标站点,使用持久化会话保存登录态.
persister = SessionPersister()
persister.save_session(
name="authenticated-crawler",
config={"CDP_PORT": "9222", "PERSIST_PROFILE": "true"},
script_code="# 抓取脚本..."
)
docker exec agent-browser bsession new price-monitor
# ...
echo 'N8N_WEBHOOK_URL=https://n8n.example.com/webhook/未指定' >> $HOME/.bsession/workspace/conf/price-monitor.conf
# ...
docker exec agent-browser bsession run price-monitor
# ...
docker exec agent-browser bsession logs price-monitor -n 50
mkdir -p $HOME/.bsession/workspace/conf
cat > $HOME/.bsession/workspace/conf/batch-monitor.conf <<EOF
[env]
CDP_PORT=9222
CHECK_INTERVAL=1800
N8N_WEBHOOK_URL=https://n8n.example.com/webhook/未指定
EOF
# ...
cat > $HOME/.bsession/workspace/(请参考skill目录中的脚本文件) <<'PYEOF'
import os, sys, time
sys.path.insert(0, "/app")
from lib.browser import ab, find_ref, send_webhook, make_logger
# ...
logger = make_logger("batch-monitor")
# ...
def run():
port = int(os.environ.get("CDP_PORT", 9222))
webhook = os.environ.get("N8N_WEBHOOK_URL", "")
# ...
urls = [
"https://example.com/page1",
]
# ...
for url in urls:
ab(port, "open", url)
time.sleep(3)
snap = ab(port, "snapshot")
send_webhook(webhook, {"url": url, "status": "checked"})
logger.info(f"已检查:{url}")
# ...
if __name__ == "__main__":
run()
PYEOF
# ...
docker exec agent-browser bsession run batch-monitor
响应解析: 完成完成后,查看输出响应确认任务状态。成功时输出包含解析摘要和响应数据;失败时根据错误信息排查问题,查阅错误解析章节获取恢复步骤.
sessions:
- name: price-monitor
interval: 1800
webhook: https://hooks.slack.com/services/未指定
timeout: 60
persistent: true
# ...
- name: stock-check
interval: 3600
webhook: https://qyapi.weixin.qq.com/cgi-(请参考skill目录中的脚本文件)?key=未指定
timeout: 90
# ...
monitoring:
check_interval: 300
max_failures: 3
alert_channels:
- type: slack
url: https://hooks.slack.com/services/未指定
- type: feishu
url: https://open.feishu.cn/open-apis/bot/v2/hook/未指定
# ...
batch:
max_workers: 5
timeout: 300
session_names = [
"ecommerce-price-monitor-prod",
"finance-stock-check-prod",
"internal-api-health-staging",
]
class ResilientSession(RecurringSession):
"""带错误恢复的会话"""
def _loop(self):
while self.running:
try:
super()._loop()
except Exception as e:
print(f"会话异常,{60}s后重试:{e}")
time.sleep(60)
manager = BatchSessionManager(max_workers=3)
专业版完全兼容免费版的所有功能。免费版的fetch命令、基础交互、容器配置在专业版中均可直接使用。升级后原有脚本无需修改,仅新增高级能力可用.
专业版支持最小60秒间隔。对于需要更短间隔的场景(如实时监控),建议使用其他工具(如Prometheus+Alertmanager)。注意:间隔过短可能触发目标站点的反爬机制.
专业版支持:n8n(通用工作流)、Slack、飞书、企业微信、钉钉、邮件(通过SMTP API)、自定义HTTP端点。所有渠道均通过Webhook URL方式调用,无需安装额外客户端.
持久化保存两个文件:(1) conf文件(环境变量配置);(2) py脚本文件(执行逻辑)。保存后可通过 bsession run <name> 直接运行,无需重复创建。conf文件中的Webhook URL、检查间隔等参数可随时修改.
理论无上限,实际受限于:(1) Docker容器资源(建议单机不超过20个并发会话);(2) 目标站点承压能力;(3) 本地网络带宽。建议通过 max_workers 参数控制并发数.
专业版采用"连续失败N次才告警"策略(默认N=3),避免单次网络抖动触发误报。同时支持告警抑制(同一会话5分钟内不重复告警)和告警恢复通知(恢复正常后发送"已恢复"消息).
专业版提供增强调试模式:(1) bsession logs <name> --debug 查看详细日志;(2) bsession debug <name> 进入断点调试;(3) bsession trace <name> 追踪执行流程;(4) 支持VSCode远程调试容器内Python脚本.
| 依赖项 | 类型 | 是否必需 | 获取方式 |
|---|---|---|---|
| Docker | 运行时 | 必需 | 官网下载安装 |
| Docker Compose | 工具 | 必需 | 随Docker Desktop安装 |
| Python 3.8+ | 运行时 | 必需 | 容器内已内置 |
| requests | Python库 | 必需 | pip install requests(Webhook通知) |
| schedule | Python库 | 可选 | pip install schedule(定时任务) |
| concurrent.futures | Python库 | 必需 | Python标准库(批量管理) |
| LLM API | API | 必需 | 由Agent平台内置LLM提供 |
本专业版相比免费版新增以下能力:
此外,专业版还提供:
| 版本 | 价格 | 功能 | 适用场景 |
|---|---|---|---|
| 免费体验版 | ¥0 | 单次抓取 + 基础交互 + 容器隔离 + 调试模式 | 个人试用、单次任务 |
| 收费专业版 | ¥39/月 | 定时任务 + 持久化 + Webhook + 批量管理 + 监控告警 + CF规避 + 调试增强 + 优先支持 | 团队/企业、长期任务 |
专业版通过SkillHub SkillPay发布.
| 错误场景 | 原因 | 处理方式 |
|---|---|---|
| 配置错误 | 参数缺失或格式错误 | 检查依赖说明中的配置要求 |
| 运行时错误 | 运行环境不满足 | 确认运行环境符合依赖说明 |
| 网络错误 | 连接超时或不可达 | 执行ping命令测试网络连通性,检查防火墙和代理设置连接后执行ping命令测试网络连通性,检查防火墙和代理设置连接后重新执行命令,参考国内替代方案 |
输出:返回执行结果,包含操作状态和输出数据
用户: 执行核心功能
Skill: 正在执行核心功能...
Skill: 执行完成,结果如下: 操作成功
{
"success": true,
"data": {
"result": "浏览器会话(专业版)处理结果",
"execution_time": "0.5s",
"metadata": {
"version": "1.0",
"processor": "bsession pro"
}
},
"execution_log": ["解析输入参数", "执行核心处理", "格式化输出结果"],
"error": null
}
| 风险类型 | 防范措施 |
|---|---|
| API密钥泄露 | 通过环境变量配置,禁止硬编码到代码或配置文件中 |
| 命令执行风险 | 仅执行白名单命令,避免拼接用户输入到命令行参数中 |
| 网络通信安全 | 使用HTTPS协议,验证SSL证书有效性 |
| 敏感数据暴露 | 输出结果中不包含密钥、令牌等敏感信息 |
使用前请确认已阅读依赖说明章节,确保运行环境满足安全要求。