ブラウザ内で CAPTCHA を解決するエージェントを使用する

# pip install browser-use capsolver-agent
# export CAPSOLVER_API_KEY=CAP-XXXXXX
# export OPENAI_API_KEY=sk-XXXXXX

import asyncio
import json
import os
from browser_use import ActionResult, Agent, Browser, ChatOpenAI, Tools
from capsolver_core import Capsolver
from capsolver_core.capsolver import SolveOnPageOptions
from _env import load_example_env  # demo only: reads repo-root .env

load_example_env()

# 1. Adapt Browser Use's live page to capsolver-core's PageDriver (evaluate + url).
class PageDriver:
    def __init__(self, session):
        self.session = session

    async def url(self) -> str:
        return await self.session.get_current_page_url()

    async def evaluate(self, script: str, arg=None):
        cdp = await self.session.get_or_create_cdp_session()
        result = await cdp.cdp_client.send.Runtime.evaluate(
            params={
                "expression": f"({script})({json.dumps(arg)})",
                "returnByValue": True,
                "awaitPromise": True,
            },
            session_id=cdp.session_id,
        )
        return result.get("result", {}).get("value")

# 2. Register one action: detect + solve + autofill on the page the agent controls.
tools = Tools()
capsolver = Capsolver(api_key=os.environ["CAPSOLVER_API_KEY"])

@tools.action(
    description="Solve and autofill every supported captcha (reCaptchaV2 / reCaptchaV3 / "
                "cloudflare) on the current page. Never solve a captcha yourself."
)
async def solve_page_captchas(browser_session) -> ActionResult:
    results = await capsolver.solve_on_page(
        PageDriver(browser_session),
        options=SolveOnPageOptions(autofill=True, throw_on_error=False),
    )
    if not results:
        return ActionResult(extracted_content="No supported captcha on this page.")
    filled = sum(1 for r in results if r.filled)
    if not filled:
        return ActionResult(error="; ".join(r.error or "not filled" for r in results))
    return ActionResult(
        extracted_content=f"Filled {filled}/{len(results)} captcha tokens. Continue the "
                          "workflow now, e.g. click Submit.",
    )

# 3. Hand the action to the agent; the LLM decides when to call it.
model = os.environ.get("OPENAI_MODEL", "gpt-4o")
# OpenAI-compatible gateways (DeepSeek, ...) reject OpenAI's structured-output format.
compat = {} if "openai.com" in os.environ.get("OPENAI_BASE_URL", "openai.com") else {
    "dont_force_structured_output": True,
    "add_schema_to_system_prompt": True,
}

agent = Agent(
    task="Solve the captcha on this page, then click Submit and report the result.",
    llm=ChatOpenAI(model=model, **compat),
    browser=Browser(headless=False),
    tools=tools,
    initial_actions=[{"navigate": {"url": "https://www.google.com/recaptcha/api2/demo"}}],
)

# 4. Run.
asyncio.run(agent.run())