如何将 CapSolver 与 Composio AI 浏览器自动化集成

运行脚本
  -> OpenAI代理SDK决定调用哪个工具
  -> Composio自定义工具: complete_recaptcha_v2
       -> Playwright打开页面
       -> capsolver.solve(...)返回gRecaptchaResponse
       -> 将令牌应用到g-recaptcha-response
       -> Playwright提交并等待页面
       -> 读取页面并判断是否通过
  -> 工具返回 {"accepted": ..., "message": ...}
  -> 代理报告accepted的结果

pip install composio composio-openai-agents openai-agents capsolver pydantic playwright
playwright install chromium

# API密钥。
COMPOSIO_API_KEY = "ak_..."
OPENAI_API_KEY = "sk-..."          # 您的官方OpenAI API密钥。
os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY   # OpenAI SDK从环境变量中读取密钥。

# 配置CapSolver和Composio。
capsolver.api_key = "CAP-..."
composio = Composio(
    api_key=COMPOSIO_API_KEY,
    provider=OpenAIAgentsProvider(),
)

import os
from typing import List, cast
import capsolver
from agents import Agent, Runner, SQLiteSession
from composio import Composio
from composio.core.models.custom_tool import CustomTool
from composio.core.models.tool_router import ToolRouterExperimentalConfig
from composio_openai_agents import OpenAIAgentsProvider
from playwright.sync_api import sync_playwright
from pydantic import BaseModel, Field

# API密钥。
COMPOSIO_API_KEY = "ak_..."
OPENAI_API_KEY = "sk-..."          # 您的官方OpenAI API密钥。
os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY

# 配置CapSolver和Composio。
capsolver.api_key = "CAP-..."
composio = Composio(
    api_key=COMPOSIO_API_KEY,
    provider=OpenAIAgentsProvider(),
)


# 自定义工具的输入模式；Composio要求此处为Pydantic BaseModel。
class CompleteRecaptchaInput(BaseModel):
    target_url: str = Field(
        default="https://www.google.com/recaptcha/api2/demo",
        description="包含reCAPTCHA v2演示的页面URL",
    )
    website_key: str = Field(
        default="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
        description="当前页面的reCAPTCHA v2网站密钥",
    )


# 将整个流程注册为一个Composio工具，代理可调用。
# 第一个参数的类型注解是Composio推断模式所必需的。
@composio.experimental.tool(preload=True)
def complete_recaptcha_v2(input: CompleteRecaptchaInput, _ctx):
    """使用Playwright打开页面，解决reCAPTCHA v2，提交并验证。"""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)  # 设置headless=True以隐藏窗口。
        page = browser.new_page()
        try:
            page.goto(input.target_url)
            # 请求CapSolver解决reCAPTCHA v2挑战。
            solution = capsolver.solve(
                {
                    "type": "ReCaptchaV2TaskProxyLess",
                    "websiteURL": input.target_url,
                    "websiteKey": input.website_key,
                }
            )
            token = solution.get("gRecaptchaResponse")
            page.evaluate(
                """
                (token) => {
                    const textarea = document.getElementById('g-recaptcha-response');
                    if (textarea) {
                        textarea.value = token;
                    }
                }
                """,
                token,
            )
            page.click("#recaptcha-demo-submit")
            page.wait_for_load_state("networkidle")
            result_page = page.content()
            # 仅当页面实际显示成功文本时才视为成功
            accepted = "Verification Success" in result_page
            return {
                "accepted": accepted,
                "message": (
                    "Verification Success"
                    if accepted
                    else "页面未报告Verification Success"
                ),
            }
        finally:
            browser.close()
def main():
    experimental: ToolRouterExperimentalConfig = {
        "custom_tools": cast(List[CustomTool], [complete_recaptcha_v2]),
    }
    session = composio.sessions.create(
        user_id="playwright-recaptcha-demo-user",
        experimental=experimental,
        sandbox={"enable": False},  # 在此进程中运行工具，而非沙箱。
    )
    agent = Agent(
        name="Playwright reCAPTCHA助手",
        instructions=(
            "当用户要求运行演示时，使用默认值调用complete_recaptcha_v2 "
            "。仅当accepted为true时报告成功。"
        ),
        model="gpt-5.2",
        tools=session.tools(),
    )
    # 多轮对话的内存
    memory = SQLiteSession("conversation")
    print("Composio + Playwright reCAPTCHA v2演示正在运行...")
    user_input = (
        "现在使用默认的target_url和website_key调用complete_recaptcha_v2 "
        "。不要询问确认。"
    )
    result = Runner.run_sync(
        starting_agent=agent,
        input=user_input,
        session=memory,
    )
    print(f"助手: {result.final_output}\n")
if __name__ == "__main__":
    main()

image_src = page.locator("#demoCaptcha_CaptchaImage").get_attribute("src")
if not image_src or "," not in image_src:
    raise RuntimeError("未找到有效的CAPTCHA图片Data URL")
base64_image = image_src.split(",", 1)[1]  # 去除"data:image/...;base64,"前缀。

class CompleteImageCaptchaInput(BaseModel):
    target_url: str = Field(
        default="https://captcha.com/demos/features/captcha-demo.aspx",
        description="图片CAPTCHA演示页面URL",
    )
    module: str = Field(
        default="common",
        description="CapSolver ImageToTextTask识别模块",
    )

@composio.experimental.tool(preload=True)
def complete_image_captcha(input: CompleteImageCaptchaInput, _ctx):
    """使用Playwright打开页面，识别图片CAPTCHA，提交并验证。"""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()
        try:
            page.goto(input.target_url)
            page.wait_for_selector("#demoCaptcha_CaptchaImage", state="visible")

            # 图片src已经是数据URL；去除前缀以获取Base64。
            image_src = page.locator("#demoCaptcha_CaptchaImage").get_attribute("src")
            if not image_src or "," not in image_src:
                raise RuntimeError("未找到有效的CAPTCHA图片Data URL")
            base64_image = image_src.split(",", 1)[1]
            solution = capsolver.solve(
                {
                    "type": "ImageToTextTask",
                    "websiteURL": input.target_url,
                    "module": input.module,
                    "body": base64_image,
                }
            )
            captcha_text = solution.get("text")
            if not isinstance(captcha_text, str) or not captcha_text:
                raise RuntimeError("CapSolver未返回识别文本")
            page.fill("#captchaCode", captcha_text)      # 填写识别文本。
            page.click("#validateCaptchaButton")
            page.wait_for_load_state("networkidle")
            result_page = page.content()
            # 演示页面在成功时显示"Correct!"，失败时显示"Incorrect!"。
            accepted = "Correct!" in result_page
            return {
                "accepted": accepted,
                "recognized_text": captcha_text,
                "message": "Correct!" if accepted else "页面未报告Correct!",
            }
        finally:
            browser.close()

Playwright打开CAPTCHA页面
  -> 等待#demoCaptcha_CaptchaImage可见
  -> 读取src（数据URL）并去除前缀以获取Base64
  -> capsolver.solve(ImageToTextTask)返回文本
  -> page.fill将结果写入#captchaCode
  -> page.click激活#validateCaptchaButton
  -> page.content检查Correct!或Incorrect!
  -> finally关闭浏览器

solution = capsolver.solve({
    "type": "ImageToTextTask",
    "module": "number",
    "images": [base64_image],
})

answers = solution["answers"]

experimental.tool: "complete_recaptcha_v2"的第一个参数必须
标注为Pydantic BaseModel子类。得到: <class 'inspect._empty'>

403 APIKey_InsufficientPermissions
此路由需要 "sessions" 写入权限

Composio 工具 = Playwright 页面操作 + CapSolver 结果 + 页面状态验证