如何在CrewAI代理中使用CapSolver解决reCAPTCHA v3
Ethan Collins
Pattern Recognition Specialist
简要
- CrewAI 的 reCAPTCHA v3 求解器应为一个窄类型工具，而不是通用浏览器或任意 URL 函数。
- 从服务器端注册表中解析目标 URL、站点密钥、预期的
pageAction和评分策略，而不是从模型生成的文本中解析。 - 使用 CapSolver 的文档中定义的 reCAPTCHA v3 任务参数，并明确 Enterprise、代理和会话设置。
- 在受信任的应用程序代码中提交返回的令牌；向 CrewAI 代理返回仅脱敏的状态对象。
- 将求解结果视为中间步骤，并在 Crew 继续之前验证预期页面状态是否已更改。
引言
在 CrewAI 中最安全的解决 reCAPTCHA v3 的方法是通过一个类型化、策略控制的工具暴露 CapSolver。CrewAI 应决定何时需要验证任务，但受信任的应用程序代码应解析已批准的目标、站点密钥、页面操作、代理模式和评分策略。CapSolver 的 reCAPTCHA v3 文档定义了支持的任务类型和参数，而用户提供的 Agent SDK 将结构化工具调用映射到 capsolver-core。该工具应服务器端提交令牌，验证结果页面状态，并向 Crew 返回一个小的结果，如 verified、review_required 或 stopped。这种设计可防止 URL 偏移、操作猜测、密钥泄露、重复求解和虚假成功信号。
为什么 reCAPTCHA v3 需要不同的 CrewAI 工具
reCAPTCHA v3 基于评分，并且通常在没有可见复选框的情况下运行。目标应用程序调用一个操作，接收一个令牌，并在服务器上评估该令牌。因此，CrewAI 工作流即使从未看到视觉挑战也可能失败。
常见的原因是操作性的而非对话性的：
- 工具使用了错误的站点密钥；
pageAction与页面的运行时操作不匹配；- 令牌被提交到不同的路由或浏览器状态；
- 任务类型未匹配 Standard 或 Enterprise；
- Crew 将任务完成视为应用程序验证；
- 同一步骤在没有状态变化的情况下创建了多个令牌。
CapSolver reCAPTCHA 博客 包含支持的实现指南，而 AI 和自动化 FAQ 帮助定义安全的代理边界。
将 Crew 角色与求解器权限分离
多代理 Crew 在责任明确时效果最佳。
| 角色 | 允许的责任 | 不得控制 |
|---|---|---|
| 导航员 | 观察已批准的应用程序状态 | API 密钥、代理凭证、原始令牌 |
| 验证计划员 | 决定已批准的步骤是否需要工具 | 任意 URL 或站点密钥 |
| CapSolver 工具 | 解析策略、创建一个任务、提交令牌 | 无限制重试或无关浏览 |
| 状态验证器 | 确认预期的路由和语义标记 | 业务批准决策 |
| 审查员 | 在失败时检查脱敏证据 | 秘密会话材料 |
模型可以选择注册的目标 ID。它不应构造目标 URL 或挑战参数。
使用官方的 CrewAI 工具模式
CrewAI 的 自定义工具文档 支持 BaseTool 与 Pydantic args_schema、@tool 装饰器、类型化结果以及用于 I/O 绑定操作的异步工具。
用户提供的 CapSolver Agent 文档说明 capsolver-agent 包装 capsolver-core：create_executor() 创建执行器，executor.execute("solve_captcha", args) 将类型化请求分派到核心引擎。
安装记录的包：
bash
pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install git+https://github.com/capsolver-ai/capsolver-agent.git
pip install crewai
export CAPSOLVER_API_KEY="CAP-..."将密钥保留在运行时密钥存储中。不要将它们放在 Crew 提示、任务描述或工具结果中。
定义受信任的目标注册表
python
from dataclasses import dataclass
@dataclass(frozen=True)
class RecaptchaV3Policy:
target_id: str
website_url: str
website_key: str
page_action: str
minimum_score: float
enterprise: bool
proxy_profile: str | None
allowed_crew_role: str
max_attempts: int = 1
TARGETS = {
"approved_login_test": RecaptchaV3Policy(
target_id="approved_login_test",
website_url="https://approved.example.com/login",
website_key="PUBLIC_SITE_KEY",
page_action="login",
minimum_score=0.7,
enterprise=False,
proxy_profile=None,
allowed_crew_role="verification_specialist",
)
}公共站点密钥不是账户密钥，但仍应来自受信任的配置，以防止模型重定向工具。
观察 pageAction 而非猜测它
CapSolver 的 reCAPTCHA v3 文档 将 pageAction 列为可选任务字段，并解释该值可以在页面的 grecaptcha.execute 调用中找到。
python
@dataclass(frozen=True)
class PageObservation:
target_id: str
current_url: str
observed_action: str
observed_site_key: str
form_state: str
observed_at: str
def validate_observation(
observation: PageObservation,
policy: RecaptchaV3Policy,
) -> None:
if observation.current_url != policy.website_url:
raise PermissionError("观察到的 URL 与策略不匹配")
if observation.observed_site_key != policy.website_key:
raise ValueError("观察到的站点密钥与策略不匹配")
if observation.observed_action != policy.page_action:
raise ValueError("观察到的页面操作与策略不匹配")
if observation.form_state != "READY_FOR_VERIFICATION":
raise ValueError("应用程序状态未准备好")工具应拒绝不匹配项，而不是为不确定的参数创建令牌。
理解 CapSolver v3 参数
| 参数 | 用途 | 策略规则 |
|---|---|---|
captcha_type |
在 Agent SDK 中选择 reCAPTCHA v3 | 固定为 reCaptchaV3 |
website_url |
承载挑战的页面 | 从注册表加载 |
website_key |
公共站点密钥 | 从注册表加载并检查页面 |
page_action |
运行时 v3 操作 | 必须与观察结果匹配 |
min_score |
请求的最小分数 | 由目标策略设置 |
enterprise |
标准或企业路径 | 由集成配置固定 |
proxy |
可选的网络身份 | 如果需要，从秘密配置加载 |
CapSolver 文档记录了标准、企业、代理和无代理任务变体。在启用相关模式时，其结果可能包含 gRecaptchaResponse、用户代理数据和会话值。
CapSolver 产品页面 可在实现前确认支持的任务家族。
创建类型化的 CrewAI 工具
python
import os
from typing import Literal
from crewai.tools import tool
from pydantic import BaseModel, Field
from capsolver_agent.schema import create_executor
executor = create_executor(api_key=os.environ["CAPSOLVER_API_KEY"])
class SolveRequest(BaseModel):
target_id: str = Field(description="注册的目标标识符")
crew_role: str = Field(description="请求验证的角色")
observed_action: str = Field(description="在实时页面上观察到的操作")
observed_site_key: str = Field(description="在实时页面上观察到的站点密钥")
state_id: str = Field(description="不透明的服务器端应用程序状态标识符")
class SolveResult(BaseModel):
status: Literal["verified", "review_required", "stopped"]
target_id: str
state_id: str
reason: str
task_attempted: bool结果模型故意排除了令牌、API 密钥、代理、cookies 和原始提供者响应。
解析密钥并服务器端提交令牌
python
PROXY_VAULT = {
"approved_proxy": os.environ.get("APPROVED_PROXY")
}
async def submit_token_and_verify(
*,
state_id: str,
token: str,
policy: RecaptchaV3Policy,
) -> bool:
"""应用程序拥有的提交和检查函数。"""
response = await application_sessions.submit_recaptcha_v3(
state_id=state_id,
token=token,
expected_action=policy.page_action,
)
return (
response.current_url.startswith("https://approved.example.com/account")
and response.semantic_marker == "AUTHENTICATED_ACCOUNT_PAGE"
and response.challenge_present is False
)application_sessions 代表您的授权浏览器或 HTTP 会话服务。求解器工具使用它，但模型不会收到其凭证。
实现异步工具
python
@tool("解决已批准的 reCAPTCHA v3", result_schema=SolveResult)
async def solve_approved_recaptcha_v3(
target_id: str,
crew_role: str,
observed_action: str,
observed_site_key: str,
state_id: str,
) -> dict:
"""解决一个已注册的 reCAPTCHA v3 步骤并验证应用程序状态。"""
policy = TARGETS.get(target_id)
if policy is None:
return SolveResult(
status="stopped",
target_id=target_id,
state_id=state_id,
reason="未知目标",
task_attempted=False,
).model_dump()
if crew_role != policy.allowed_crew_role:
return SolveResult(
status="stopped",
target_id=target_id,
state_id=state_id,
reason="角色不允许调用此工具",
task_attempted=False,
).model_dump()
if observed_action != policy.page_action:
return SolveResult(
status="review_required",
target_id=target_id,
state_id=state_id,
reason="观察到的操作与目标策略不匹配",
task_attempted=False,
).model_dump()
if observed_site_key != policy.website_key:
return SolveResult(
status="review_required",
target_id=target_id,
state_id=state_id,
reason="观察到的站点密钥与目标策略不匹配",
task_attempted=False,
).model_dump()
args = {
"captcha_type": "reCaptchaV3",
"website_url": policy.website_url,
"website_key": policy.website_key,
"page_action": policy.page_action,
"min_score": policy.minimum_score,
"enterprise": policy.enterprise,
}
if policy.proxy_profile:
args["proxy"] = PROXY_VAULT[policy.proxy_profile]
result = await executor.execute("solve_captcha", args)
if not result.get("success"):
return SolveResult(
status="review_required",
target_id=target_id,
state_id=state_id,
reason="CapSolver 任务未完成",
task_attempted=True,
).model_dump()
solution = result.get("solution") or {}
token = solution.get("token")
if not token:
return SolveResult(
status="review_required",
target_id=target_id,
state_id=state_id,
reason="任务结果中未包含令牌",
task_attempted=True,
).model_dump()
verified = await submit_token_and_verify(
state_id=state_id,
token=token,
policy=policy,
)
return SolveResult(
status="verified" if verified else "review_required",
target_id=target_id,
state_id=state_id,
reason="应用程序状态已验证" if verified else "应用程序状态未验证",
task_attempted=True,
).model_dump()此实现将凭证保留在受信任的代码中，并仅向 Crew 返回策略安全的状态。
将工具附加到一个代理
python
from crewai import Agent, Crew, Process, Task
verification_agent = Agent(
role="verification_specialist",
goal="仅完成已注册的验证步骤并报告已验证状态",
backstory=(
"您操作批准的验证工具。您从不发明目标 ID、站点密钥、操作、凭证或成功状态。"
),
tools=[solve_approved_recaptcha_v3],
allow_delegation=False,
verbose=True,
)
verification_task = Task(
description=(
"对于提供的观察中的注册目标，仅在 URL、站点密钥、操作和状态确认后调用工具。返回结构化的状态，不包含秘密。"
),
expected_output="结构化的 verified、review_required 或 stopped 结果。",
agent=verification_agent,
)
crew = Crew(
agents=[verification_agent],
tasks=[verification_task],
process=Process.sequential,
verbose=True,
)不要将求解器工具附加到每个代理。将其限制为一个角色可使授权和审计更清晰。
防止重复调用工具
python
from datetime import datetime, timedelta, timezone
ATTEMPTS: dict[tuple[str, str], datetime] = {}
def claim_attempt(target_id: str, state_id: str) -> bool:
key = (target_id, state_id)
now = datetime.now(timezone.utc)
prior = ATTEMPTS.get(key)
if prior and now - prior < timedelta(minutes=2):
return False
ATTEMPTS[key] = now
return True在 executor.execute() 之前调用 claim_attempt()。重复的 Crew 消息不应为同一应用程序状态创建第二个令牌。
将令牌保留在 Crew 内存之外
CrewAI 内存、跟踪和详细日志可能会保留工具输出。仅返回：
json
{
"status": "verified",
"target_id": "approved_login_test",
"state_id": "state_7c19",
"reason": "应用程序状态已验证",
"task_attempted": true
}永远不要返回令牌、CapSolver API 密钥、代理值、浏览器 cookie、原始 HTML、密码或个人表单数据。
CapSolver 错误和故障排除 FAQ 可在不向 Crew 暴露原始响应的情况下支持提供者错误分类。
提交后验证页面
验证器应要求多个独立信号：
python
@dataclass(frozen=True)
class StateCheck:
expected_path_prefix: str
required_marker: str
forbidden_markers: tuple[str, ...]
def is_verified(page, check: StateCheck) -> bool:
return (
page.url.path.startswith(check.expected_path_prefix)
and page.has_semantic_marker(check.required_marker)
and not any(page.contains(marker) for marker in check.forbidden_markers)
and page.http_status == 200
)仅更改 URL 不够。需要预期的路由、语义标记、状态和已知挑战或错误状态的缺失。
明确处理企业模式和会话模式
CapSolver文档记录了企业版变体和可选的会话模式。不要让团队推断这些选项。
python
@dataclass(frozen=True)
class RecaptchaV3Policy:
target_id: str
website_url: str
website_key: str
page_action: str
minimum_score: float
enterprise: bool
is_session: bool
proxy_profile: str | None
allowed_crew_role: str
max_attempts: int = 1如果批准的目标使用企业版，则在策略中记录该事实。如果需要会话模式，请在应用程序会话服务中处理返回的会话值，并将其排除在团队输出之外。
对比总结
| 设计 | 参数完整性 | 密钥安全性 | 页面状态保障 | 推荐 |
|---|---|---|---|---|
| 模型提供URL、密钥和操作 | 低 | 低 | 低 | 避免 |
| 工具返回令牌给团队 | 中 | 低 | 低 | 避免 |
| 注册解析的工具提交和验证 | 高 | 高 | 高 | 优先选择 |
| 人工验证步骤 | 高 | 高 | 高 | 用于敏感或不确定状态 |
首选设计将决策权交给模型，但将执行权保留在受信任的代码中。
添加操作指标
跟踪脱敏字段，例如：
- 目标ID；
- 团队角色；
- 观察到的操作匹配；
- 任务尝试；
- 提供商状态类别；
- 提交持续时间；
- 页面验证；
- 人工审核原因。
python
SAFE_FIELDS = {
"target_id",
"crew_role",
"action_match",
"task_attempted",
"provider_category",
"duration_ms",
"verified",
"review_reason",
}
def safe_event(event: dict) -> dict:
return {key: event[key] for key in SAFE_FIELDS if key in event}CapSolver状态页面可以帮助区分供应商可用性与应用程序特定的故障。
在生产前测试工具
python
import pytest
@pytest.mark.asyncio
async def test_unknown_target_stops_before_task():
result = await solve_approved_recaptcha_v3.run(
target_id="unknown",
crew_role="verification_specialist",
observed_action="login",
observed_site_key="x",
state_id="state-1",
)
assert result["status"] == "stopped"
assert result["task_attempted"] is False
@pytest.mark.asyncio
async def test_action_mismatch_requests_review():
result = await solve_approved_recaptcha_v3.run(
target_id="approved_login_test",
crew_role="verification_specialist",
observed_action="checkout",
observed_site_key="PUBLIC_SITE_KEY",
state_id="state-2",
)
assert result["status"] == "review_required"
assert result["task_attempted"] is False还需测试重复尝试阻止、密钥脱敏、缺失令牌处理、企业策略和页面验证失败。
附加代码：在CapSolver仪表板使用代码WEBS，每次充值可获得额外5%的奖励。
生产清单
- 在服务器端注册每个目标URL、站点密钥、操作、分数和企业设置。
- 让一个CrewAI角色调用求解工具。
- 在任务创建前验证观察到的URL、站点密钥、操作和状态。
- 使用CapSolver的文档化reCAPTCHA v3参数。
- 在受信任的应用程序代码中提交令牌。
- 仅向团队返回脱敏的结构化结果。
- 每个观察到的应用程序状态只允许一次尝试。
- 验证路由、语义标记、状态和挑战不存在。
- 将密钥、令牌、代理数据、cookies和表单数据保留在内存和追踪之外。
- 将不确定或敏感状态路由到人工审核员。
CapSolver CAPTCHA求解常见问题提供了额外的任务生命周期指导。
负责任的使用
仅在您拥有、测试或明确授权自动化的网站上使用CrewAI的reCAPTCHA v3求解器。遵守条款、速率限制、认证边界、隐私义务和内部访问策略。公共站点密钥不表示有权访问受保护的工作流。在单独的审批控制后处理重要提交、支付、账户更改和敏感数据决策。
结论
生产环境中的CrewAI reCAPTCHA v3求解器应具有针对性、类型化和策略控制。CrewAI可以识别验证需求，但受信任的代码必须解决目标、站点密钥、页面操作、分数、企业模式和网络设置。CapSolver应在验证状态后运行一次，令牌应在服务器端提交，团队在目标页面验证后仅接收脱敏结果。
通过CapSolver启动授权实现，在受控页面上进行测试，并在生产使用前添加参数、重复调用、脱敏和页面状态测试。
常见问题
CrewAI可以直接调用CapSolver吗？
CrewAI可以调用一个类型化工具，该工具将委托给文档化的CapSolver代理执行器。将目标解析、密钥、令牌提交和验证保留在受信任的应用程序代码中。
哪些reCAPTCHA v3参数是必需的？
目标URL和站点密钥是必需的。页面操作、最低分数、企业设置、会话模式和代理取决于批准的目标配置。
模型应选择页面操作吗？
不。从实时批准的页面观察操作，并与服务器端策略值进行比较。
应将CapSolver令牌返回给团队吗？
不。在受信任的代码中提交它，并仅返回脱敏的已验证、需要审核或已停止的状态。
一个CrewAI任务应进行多少次尝试？
默认情况下，每个观察到的应用程序状态进行一次尝试。第二次尝试需要新的观察和明确的策略决策。