如何使用CapSolver构建AI浏览器恢复工具
Ethan Collins
Pattern Recognition Specialist
- 标题: 如何使用 CapSolver 构建 AI 浏览器恢复工具
- 元描述: 使用 CapSolver、Playwright 固定装置、页面状态路由、检查点、有限重试、脱敏跟踪和 CI 测试构建 AI 浏览器恢复工具。
- 关键词: ai 浏览器恢复工具 capsolver, ai 代理浏览器工具, playwright capcha 恢复, 代理浏览器可靠性, capsolver solve_on_page
- 封面图片替代文本: AI 浏览器恢复工具通过 CapSolver 路由页面状态并恢复授权工作流
如何使用 CapSolver 构建 AI 浏览器恢复工具
TL;DR
- 将浏览器策略、页面状态分类、挑战恢复、检查点、遥测和清理放在模型之外的工具中。
- 为批准的任务保持一个 Playwright 上下文，并在确定性恢复边界使用 CapSolver Core 的
detect、get_captcha_info或solve_on_page。 - 允许一次有限的恢复尝试，验证预期页面返回，将循环或未知状态路由到操作员审查。
- 记录元数据和工件引用，但脱敏令牌、cookies、凭据、表单内容和代理值。
- 在将工具连接到任何 AI 代理框架之前，使用隔离的固定装置和受控的测试页面对其进行测试。
引言
AI 浏览器恢复工具是当页面状态意外变化时保持浏览器任务受控的运行时层。模型可以决定下一步的业务步骤，但工具应拥有 Playwright 上下文、批准的主机策略、导航检查点、页面分类、支持的挑战恢复、重试限制、跟踪、截图和关闭。CapSolver 作为确定性恢复功能嵌入此层：capsolver-core 可以检测支持的挑战、读取参数、解决它们并将结果填充回同一页面。然后工具验证预期的应用程序状态返回，然后再允许代理继续。本指南构建了策略模型、状态机、异步上下文管理器、恢复函数、工件记录器、OpenTelemetry 跨度、测试和生产控制，以实现可靠的授权浏览器自动化。
工具中应包含什么
工具不是模型也不是浏览器驱动程序本身。它是它们之间的控制平面。
text
代理的业务目标
↓
浏览器恢复工具
├─ 目标策略
├─ Playwright 上下文
├─ 状态分类器
├─ 检查点存储
├─ CapSolver 恢复
├─ 重试预算
├─ 跟踪 + 工件
└─ 清理
↓
批准的页面操作或操作员审查Playwright 的 固定装置文档 强调了隔离的页面和浏览器上下文固定装置、可重用的设置和清理、可组合性以及自动调试附件。这些属性直接转化为生产工具。
CapSolver Core SDK 文档 定义了四个有用的浏览器阶段：detect、get_captcha_info、solve 和 solve_on_page。
将代理决策与运行时决策分开
模型可能决定打开已知产品页面或读取公共状态。工具决定请求的主机是否被允许，当前页面是否符合预期，是否支持恢复，以及重试预算是否还存在。
| 决策 | 所有者 | 原因 |
|---|---|---|
| 下一步业务步骤 | 代理或工作流 | 需要任务上下文 |
| 主机和路径权限 | 工具策略 | 必须是确定性的 |
| 页面状态分类 | 工具分类器 | 必须使用受信任的 DOM/网络证据 |
| 挑战恢复调用 | 工具 | 需要密钥和浏览器对象 |
| 令牌/cookie 处理 | 工具 | 敏感的运行时数据 |
| 继续 vs 审查 | 工具状态机 | 强制有限恢复 |
| 最终提交 | 人类或专用服务 | 高影响操作 |
CapSolver AI 代理指南 解释了相同的分工：模型处理推理，而 CapSolver 的各层执行支持的挑战工作。
定义目标策略
从批准的主机、路径、操作和预算的窄策略开始。
python
from dataclasses import dataclass, field
from urllib.parse import urlparse
@dataclass(frozen=True)
class TargetPolicy:
allowed_hosts: set[str]
allowed_path_prefixes: tuple[str, ...]
max_navigations: int = 20
max_recovery_attempts: int = 1
capture_screenshots: bool = True
capture_html: bool = False
allow_form_submission: bool = False
def validate_url(self, url: str) -> None:
parsed = urlparse(url)
if parsed.scheme != "https":
raise PermissionError("仅允许 HTTPS 目标")
if parsed.hostname not in self.allowed_hosts:
raise PermissionError("主机超出批准的策略")
if not parsed.path.startswith(self.allowed_path_prefixes):
raise PermissionError("路径超出批准的策略")使用租户特定的策略。不要为不相关的客户或项目维护一个全局允许列表。
CapSolver AI 和自动化常见问题解答 提供了集成上下文，而 CapSolver 网络抓取常见问题解答 涵盖了负责任的公共数据工作流。
对浏览器状态机建模
恢复工具应使用显式状态，而不是无限制的“重试”循环。
python
from enum import Enum
class BrowserState(str, Enum):
EXPECTED_PAGE = "expected_page"
SUPPORTED_CHALLENGE = "supported_challenge"
UNKNOWN_PAGE = "unknown_page"
RECOVERING = "recovering"
RECOVERED = "recovered"
REVIEW_REQUIRED = "review_required"
FAILED = "failed"允许的转换可以表示为数据：
python
ALLOWED_TRANSITIONS = {
BrowserState.EXPECTED_PAGE: {
BrowserState.EXPECTED_PAGE,
BrowserState.SUPPORTED_CHALLENGE,
BrowserState.UNKNOWN_PAGE,
},
BrowserState.SUPPORTED_CHALLENGE: {
BrowserState.RECOVERING,
BrowserState.REVIEW_REQUIRED,
},
BrowserState.RECOVERING: {
BrowserState.RECOVERED,
BrowserState.REVIEW_REQUIRED,
BrowserState.FAILED,
},
BrowserState.RECOVERED: {
BrowserState.EXPECTED_PAGE,
BrowserState.REVIEW_REQUIRED,
},
}验证每个转换。这使循环可见且可测试。
创建浏览器检查点
检查点记录确定工作流是否正确恢复所需的安全元数据。
python
from dataclasses import dataclass
from datetime import datetime, timezone
@dataclass
class BrowserCheckpoint:
url: str
title: str
expected_selector: str | None
navigation_index: int
recovery_attempts: int
observed_at: str
async def checkpoint(page, expected_selector, nav_index, attempts):
return BrowserCheckpoint(
url=page.url,
title=await page.title(),
expected_selector=expected_selector,
navigation_index=nav_index,
recovery_attempts=attempts,
observed_at=datetime.now(timezone.utc).isoformat(),
)不要在检查点中存储存储状态、cookies、密码、令牌或完整表单值。
对当前页面进行分类
使用受信任的 DOM 证据、标题、URL 和预期选择器。永远不要让模型仅根据截图推断页面状态。
python
async def classify_page(page, expected_selector: str) -> BrowserState:
if await page.locator(expected_selector).count():
return BrowserState.EXPECTED_PAGE
title = (await page.title()).strip().lower()
html = (await page.content()).lower()
challenge_markers = (
"just a moment...",
"challenge-platform",
"cf-chl-",
)
if any(marker in title or marker in html for marker in challenge_markers):
return BrowserState.SUPPORTED_CHALLENGE
return BrowserState.UNKNOWN_PAGE使用目标特定的标记和受控固定装置。标记集是一种路由启发式方法，而不是访问权限。
CapSolver CAPTCHA 解决常见问题解答 解释了支持的挑战工作流，而 CapSolver 错误常见问题解答 帮助分类失败。
为每个工具初始化 CapSolver Core 一次
官方 Core SDK 建议使用其异步上下文管理器，以便正确重用和释放 HTTP 连接。
python
import os
from capsolver_core import create_capsolver
def create_recovery_client():
return create_capsolver(
api_key=os.environ["CAPSOLVER_API_KEY"],
default_timeout=120,
polling_interval=5,
request_timeout_ms=30000,
source="ai-browser-recovery-harness",
version="1.0.0",
)不要为每次 DOM 检查创建新客户端。为工具生命周期保持一个客户端，并在清理时关闭它。
实现有限的恢复函数
使用 detect 和 get_captcha_info 进行诊断，然后使用 solve_on_page 进行一站式浏览器流程。
python
from capsolver_core import SolveOnPageOptions
async def recover_supported_challenge(
cap,
page,
policy: TargetPolicy,
recovery_attempts: int,
) -> dict:
policy.validate_url(page.url)
if recovery_attempts >= policy.max_recovery_attempts:
return {
"success": False,
"state": BrowserState.REVIEW_REQUIRED,
"reason": "recovery budget exhausted",
}
detected = await cap.detect(page)
if not detected:
return {
"success": False,
"state": BrowserState.REVIEW_REQUIRED,
"reason": "no supported challenge detected",
}
infos = await cap.get_captcha_info(page)
results = await cap.solve_on_page(
page,
options=SolveOnPageOptions(
autofill=True,
throw_on_error=False,
timeout=120,
polling_interval=5,
),
)
errors = [item.error for item in results if item.error]
filled = bool(results) and all(item.filled for item in results)
return {
"success": filled and not errors,
"state": (
BrowserState.RECOVERED
if filled and not errors
else BrowserState.REVIEW_REQUIRED
),
"detected_count": len(detected),
"info_count": len(infos),
"result_count": len(results),
"errors": errors,
}保留原始 page 对象。solve_on_page 的目的是在现有浏览器会话内检测、解决并填充。
在继续之前验证恢复
成功的工具响应并不能证明预期的应用程序页面已返回。
python
async def verify_recovery(
page,
expected_selector: str,
timeout_ms: int = 15000,
) -> bool:
try:
await page.locator(expected_selector).wait_for(
state="visible",
timeout=timeout_ms,
)
return True
except Exception:
return False恢复后，再次对页面进行分类。如果挑战仍然存在或预期选择器缺失，请停止并请求审查。
python
async def recover_and_verify(cap, page, policy, expected_selector, attempts):
result = await recover_supported_challenge(
cap=cap,
page=page,
policy=policy,
recovery_attempts=attempts,
)
if not result["success"]:
return result
if not await verify_recovery(page, expected_selector):
return {
"success": False,
"state": BrowserState.REVIEW_REQUIRED,
"reason": "expected page did not return after recovery",
}
return {
"success": True,
"state": BrowserState.EXPECTED_PAGE,
"reason": "page recovered and verified",
}构建异步工具上下文
使用异步上下文管理器以确保清理。
python
from contextlib import asynccontextmanager
from playwright.async_api import async_playwright
@dataclass
class BrowserHarness:
policy: TargetPolicy
playwright: object
browser: object
context: object
page: object
capsolver: object
navigation_count: int = 0
recovery_attempts: int = 0
@asynccontextmanager
async def browser_recovery_harness(policy: TargetPolicy):
async with async_playwright() as playwright:
browser = await playwright.chromium.launch(headless=True)
context = await browser.new_context()
page = await context.new_page()
async with create_recovery_client() as cap:
harness = BrowserHarness(
policy=policy,
playwright=playwright,
browser=browser,
context=context,
page=page,
capsolver=cap,
)
try:
yield harness
finally:
await context.close()
await browser.close()模型或工作流接收受控方法，而不是原始的无限制浏览器访问。
暴露狭窄的工具操作
python
async def safe_navigate(
harness: BrowserHarness,
url: str,
expected_selector: str,
) -> dict:
harness.policy.validate_url(url)
if harness.navigation_count >= harness.policy.max_navigations:
return {
"success": False,
"state": BrowserState.REVIEW_REQUIRED,
"reason": "navigation budget exhausted",
}
harness.navigation_count += 1
await harness.page.goto(url, wait_until="domcontentloaded")
state = await classify_page(harness.page, expected_selector)
if state == BrowserState.EXPECTED_PAGE:
return {"success": True, "state": state}
if state == BrowserState.SUPPORTED_CHALLENGE:
result = await recover_and_verify(
cap=harness.capsolver,
page=harness.page,
policy=harness.policy,
expected_selector=expected_selector,
attempts=harness.recovery_attempts,
)
harness.recovery_attempts += 1
return result
return {
"success": False,
"state": BrowserState.REVIEW_REQUIRED,
"reason": "unknown page state",
}代理可以请求 safe_navigate，但工具拥有策略和恢复路径。
记录脱敏遥测
OpenTelemetry 的 GenAI 可观测性指南 描述了模型和工具操作的跟踪。它还指出完整提示和工具内容可能包含敏感数据。默认使用仅元数据的跨度。
python
from opentelemetry import trace
tracer = trace.get_tracer("capsolver.browser_harness")
async def traced_safe_navigate(harness, url, expected_selector):
with tracer.start_as_current_span("browser.safe_navigate") as span:
span.set_attribute("browser.target_host", url.split("/")[2])
span.set_attribute("browser.navigation_index", harness.navigation_count + 1)
span.set_attribute("browser.recovery_attempts", harness.recovery_attempts)
result = await safe_navigate(harness, url, expected_selector)
span.set_attribute("browser.outcome", str(result.get("state")))
span.set_attribute("browser.success", bool(result.get("success")))
return result不要将令牌、cookies、API密钥、代理凭据、存储状态、提示内容或完整页面HTML附加到跨度中。
仅在失败时捕获工件
截图和HTML可能包含个人或机密数据。仅在政策允许时捕获，尽可能进行脱敏，并存储短期引用。
python
from pathlib import Path
import secrets
async def capture_failure_artifacts(harness, directory: Path) -> dict:
artifact_id = secrets.token_hex(12)
screenshot = directory / f"{artifact_id}.png"
await harness.page.screenshot(
path=str(screenshot),
full_page=False,
)
return {
"artifact_id": artifact_id,
"screenshot_path": str(screenshot),
"url": harness.page.url,
"title": await harness.page.title(),
}使用保留限制和访问控制。当仅需要顶层状态时，避免捕获全页截图。
CapSolver浏览器自动化博客包含相关实现模式，CapSolver Chrome扩展指南可以帮助团队在开发过程中检查支持的控件参数。
使用fixture测试harness
使用隔离的浏览器上下文和受控页面。Playwright fixture提供可重用的设置和清理。
python
import pytest
@pytest.mark.asyncio
async def test_unknown_host_is_rejected():
policy = TargetPolicy(
allowed_hosts={"staging.example.com"},
allowed_path_prefixes=("/qa/",),
)
with pytest.raises(PermissionError):
policy.validate_url("https://other.example.net/qa/test")
@pytest.mark.asyncio
async def test_recovery_budget_is_bounded(fake_cap, fake_page):
policy = TargetPolicy(
allowed_hosts={"staging.example.com"},
allowed_path_prefixes=("/qa/",),
max_recovery_attempts=1,
)
result = await recover_supported_challenge(
cap=fake_cap,
page=fake_page,
policy=policy,
recovery_attempts=1,
)
assert result["state"] == BrowserState.REVIEW_REQUIRED
assert result["reason"] == "recovery budget exhausted"为无挑战、支持的挑战、未知的中间页、成功填充、解决失败、恢复后挑战循环和缺失的预期选择器创建fixture。
定义可靠性指标
| 指标 | 目的 |
|---|---|
| 预期页面率 | 衡量成功的正常导航 |
| 挑战遇到率 | 显示由批准主机产生的源摩擦 |
| 恢复成功率 | 衡量支持的恢复结果 |
| 挑战循环率 | 检测重复的中间页状态 |
| 未知页面率 | 发现布局、认证或策略变化 |
| P95恢复延迟 | 跟踪用户可见的延迟 |
| 操作员审查率 | 衡量未解决的工作流数量 |
| 工件捕获率 | 检测过多的失败日志 |
按目标策略、路由、浏览器版本、挑战类型和harness版本分解指标。永远不要将挑战恢复失败标记为业务任务失败，而不保留两个维度。
附加代码：在CapSolver仪表板使用代码WEBS，每次充值可额外获得5%的奖金。
比较摘要
| 方法 | 浏览器所有权 | 恢复控制 | 最佳用途 |
|---|---|---|---|
| 直接代理浏览器访问 | 代理运行时 | 依赖提示 | 仅限低风险原型 |
| 框架特定操作 | 代理框架 | 工具包装器 | 快速集成 |
| 专用恢复harness | 独立控制层 | 确定性状态机 | 生产可靠性与治理 |
| 仅人工恢复 | 操作员 | 手动 | 不支持或高风险工作流 |
专用harness需要更多工程，但它创建了一个策略和可观测层，可以为多个代理框架服务。
生产清单
- 每个租户使用批准的主机和路径策略。
- 将CapSolver和浏览器凭据保留在提示和追踪之外。
- 为每个受控任务使用一个浏览器上下文。
- 在恢复前后对页面进行分类。
- 除非经过审查的场景证明需要，否则只允许一次恢复尝试。
- 在未知页面、重复挑战或缺失的预期选择器时停止。
- 从遥测中脱敏工具和浏览器秘密。
- 仅在明确的保留策略下捕获失败工件。
- 在提交、购买、账户更改或其他高影响操作前要求确认。
CapSolver产品页面列出了支持的解决方案类别，CapSolver AI博客涵盖了可以调用harness操作的代理框架示例。
负责任使用
仅在您拥有、测试或有明确授权自动化系统的系统上使用浏览器恢复harness。成功的挑战解决方案不授予访问私人内容、忽略认证边界、超出速率限制或执行交易的权限。保持harness范围有限，默认只读，并可审计。将不确定性路由到人员，而不是动态扩展权限。
结论
AI浏览器恢复harness将挑战处理转化为受控的运行时能力。它拥有浏览器上下文，验证目标，分类页面状态，确定性地调用CapSolver Core，在边界处验证预期页面，记录脱敏遥测，并在有限尝试后停止。代理框架可以使用harness而无需获得直接访问秘密或不受限制的浏览器控制。
从CapSolver开始，针对批准的测试应用实现状态机，并在生产前添加隔离的fixture和可靠性门禁。
FAQ
浏览器恢复harness是代理框架吗？
不是。它是代理框架可以调用的独立运行时和策略层。harness拥有浏览器状态、恢复、检查点、遥测和清理。
为什么要使用solve_on_page？
solve_on_page在同一个Playwright页面上结合检测、参数提取、求解和DOM填充，这使其适合受控的浏览器恢复边界。
模型应该接收Playwright页面对象吗？
优先使用狭窄的harness操作，如safe_navigate和read_public_page。原始页面访问使得实施目标、导航和恢复策略更加困难。
应该允许多少次恢复尝试？
默认情况下使用一次尝试。重复的挑战或未知页面状态应路由到操作员审查，而不是创建不受控制的循环。
应该存储哪些遥测？
存储元数据，如目标主机、harness版本、状态转换、延迟、标准化错误和工件引用。不要存储解决方案令牌、cookies、API密钥、代理凭据、存储状态或私有页面内容。