如何在 Browser Use 中用自定义工具识别验证码

# Quoted, so cmd.exe does not read >= as a redirect
pip install browser-use "capskip>=1.3.0"

FIND = """() => {
  const w = document.querySelector(
    '.g-recaptcha[data-sitekey], .cf-turnstile[data-sitekey]');
  if (!w) return null;
  return {
    kind: w.classList.contains('cf-turnstile') ? 'turnstile' : 'recaptcha',
    sitekey: w.dataset.sitekey,
    invisible: w.dataset.size === 'invisible' || w.tagName !== 'DIV',
    enterprise: !!document.querySelector(
      'script[src*="recaptcha/enterprise.js"]'),
    callback: w.dataset.callback || null,
  };
}"""

FILL = """(kind, token, callback) => {
  const name = kind === 'turnstile' ? 'cf-turnstile-response'
                                    : 'g-recaptcha-response';
  const fields = document.querySelectorAll(`[name="${name}"]`);
  fields.forEach(f => { f.value = token; });
  if (callback && typeof window[callback] === 'function') {
    window[callback](token);
  }
  return fields.length;
}"""

@tools.action(
    "Solve the reCAPTCHA v2 or Cloudflare Turnstile widget on the current "
    "page. Call it after the other fields are filled, then submit the form "
    "unless the page has already moved on."
)
async def solve_captcha(browser_session: BrowserSession, page_url: str):
    page = await browser_session.must_get_current_page()
    raw = await page.evaluate(FIND)
    widget = json.loads(raw) if raw else None
    if not widget:
        return ActionResult(error="No reCAPTCHA or Turnstile widget found.")

    try:
        if widget["kind"] == "turnstile":
            result = await solver.turnstile(widget["sitekey"], page_url)
        else:
            extra = {"invisible": 1} if widget["invisible"] else {}
            result = await solver.recaptcha(
                widget["sitekey"], page_url,
                enterprise=int(widget["enterprise"]), **extra)
    except CapSkipError as exc:
        return ActionResult(error=f"CapSkip did not solve it: {exc!r}")

    filled = await page.evaluate(
        FILL, widget["kind"], result["code"], widget["callback"])
    if filled == "0":
        return ActionResult(error="Solved, but no response field to fill.")
    return ActionResult(
        extracted_content=f"Solved the {widget['kind']} CAPTCHA on {page_url} "
                          "and filled in the token. Submit the form now, "
                          "unless the page has already moved on.",
    )

EXTRA = (
    "This browser does not solve CAPTCHAs on its own, whatever the rules "
    "above say. When a page shows a reCAPTCHA or Turnstile widget, fill in "
    "the other fields, call solve_captcha, then submit the form unless the "
    "page has already moved on. Never click the CAPTCHA checkbox or "
    "challenge yourself."
)

agent = Agent(
    task="Sign up at https://example.com/signup with YOUR_EMAIL.",
    llm=ChatAnthropic(model="claude-opus-5-5", thinking={"type": "adaptive"}),
    tools=tools,
    extend_system_message=EXTRA,
    step_timeout=420,
)

import os

# Before anything imports Agent or Tools from browser_use.
os.environ.setdefault("BROWSER_USE_ACTION_TIMEOUT_S", "330")

from browser_use import Agent  # noqa: E402

# Up to 300 s of SDK polling (a few more for the last poll), one
# model call (90 s for Claude) and the page state, with room to spare.
agent = Agent(task="...", llm=llm, tools=tools,
              extend_system_message=EXTRA, step_timeout=420)

# pip install browser-use "capskip>=1.3.0"
import asyncio
import json
import os

# Browser Use reads this once, when its tools load, so set it first.
os.environ.setdefault("BROWSER_USE_ACTION_TIMEOUT_S", "330")

from browser_use import ActionResult, Agent, BrowserSession, ChatAnthropic, Tools
from capskip import AsyncCapSkip, CapSkipError

solver = AsyncCapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
)
tools = Tools()

FIND = """() => {
  const w = document.querySelector(
    '.g-recaptcha[data-sitekey], .cf-turnstile[data-sitekey]');
  if (!w) return null;
  return {
    kind: w.classList.contains('cf-turnstile') ? 'turnstile' : 'recaptcha',
    sitekey: w.dataset.sitekey,
    invisible: w.dataset.size === 'invisible' || w.tagName !== 'DIV',
    enterprise: !!document.querySelector(
      'script[src*="recaptcha/enterprise.js"]'),
    callback: w.dataset.callback || null,
  };
}"""

FILL = """(kind, token, callback) => {
  const name = kind === 'turnstile' ? 'cf-turnstile-response'
                                    : 'g-recaptcha-response';
  const fields = document.querySelectorAll(`[name="${name}"]`);
  fields.forEach(f => { f.value = token; });
  if (callback && typeof window[callback] === 'function') {
    window[callback](token);
  }
  return fields.length;
}"""


@tools.action(
    "Solve the reCAPTCHA v2 or Cloudflare Turnstile widget on the current "
    "page. Call it after the other fields are filled, then submit the form "
    "unless the page has already moved on."
)
async def solve_captcha(browser_session: BrowserSession, page_url: str):
    page = await browser_session.must_get_current_page()
    raw = await page.evaluate(FIND)
    widget = json.loads(raw) if raw else None
    if not widget:
        return ActionResult(error="No reCAPTCHA or Turnstile widget found.")
    try:
        if widget["kind"] == "turnstile":
            result = await solver.turnstile(widget["sitekey"], page_url)
        else:
            extra = {"invisible": 1} if widget["invisible"] else {}
            result = await solver.recaptcha(
                widget["sitekey"], page_url,
                enterprise=int(widget["enterprise"]), **extra)
    except CapSkipError as exc:
        return ActionResult(error=f"CapSkip did not solve it: {exc!r}")
    filled = await page.evaluate(
        FILL, widget["kind"], result["code"], widget["callback"])
    if filled == "0":
        return ActionResult(error="Solved, but no response field to fill.")
    return ActionResult(
        extracted_content=f"Solved the {widget['kind']} CAPTCHA on {page_url} "
                          "and filled in the token. Submit the form now, "
                          "unless the page has already moved on.",
    )


EXTRA = (
    "This browser does not solve CAPTCHAs on its own, whatever the rules "
    "above say. When a page shows a reCAPTCHA or Turnstile widget, fill in "
    "the other fields, call solve_captcha, then submit the form unless the "
    "page has already moved on. Never click the CAPTCHA checkbox or "
    "challenge yourself."
)


async def main():
    agent = Agent(
        task="Sign up at https://example.com/signup with YOUR_EMAIL, "
             "then report what the page says.",
        # Adaptive thinking lets Browser Use leave the tool choice to Claude.
        llm=ChatAnthropic(model="claude-opus-5-5", thinking={"type": "adaptive"}),
        tools=tools,
        extend_system_message=EXTRA,
        step_timeout=420,
    )
    history = await agent.run(max_steps=25)
    print(history.final_result())


asyncio.run(main())