1. 项目概述:Agno与CapSolver集成解决验证码挑战
在自动化领域,验证码(CAPTCHA)一直是开发者面临的主要障碍之一。无论是数据采集、自动化测试还是内容聚合,验证码都会显著降低自动化流程的效率。Agno作为新一代自主AI代理框架,通过与CapSolver的集成,为这个问题提供了优雅的解决方案。
Agno是一个开源的多代理框架,专注于高性能和隐私保护。它允许开发者在自己的基础设施上构建和运行AI系统,完全掌控数据和执行过程。而CapSolver则是专业的验证码解决服务,支持包括reCAPTCHA、Cloudflare Turnstile等多种验证码类型。
这种集成不仅仅是简单的API调用组合,而是将CapSolver的能力深度嵌入到Agno代理的工作流中。当代理在执行网络自动化任务时遇到验证码,可以自动识别类型并调用相应的CapSolver工具进行解决,整个过程无需人工干预。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 环境准备与基础配置
2.1 安装必要的Python包
要开始Agno与CapSolver的集成,首先需要安装基础依赖包:
bash复制pip install agno requests
这里requests库用于与CapSolver的API进行交互。Agno本身是一个轻量级的框架,安装过程简单快捷。
2.2 获取API密钥
在使用CapSolver服务前,需要注册账号并获取API密钥:
- 访问CapSolver官网创建账号
- 在用户面板中找到API密钥
- 将密钥保存在环境变量或配置文件中
建议将API密钥存储在环境变量中,避免硬编码在脚本里:
python复制import os
os.environ["CAPSOLVER_API_KEY"] = "your_api_key_here"
3. 核心工具实现
3.1 基础验证码解决工具
以下是实现通用验证码解决功能的基础工具:
python复制import requests
import time
from agno.tools import tool
@tool
def solve_captcha(
website_url: str,
website_key: str,
captcha_type: str = "ReCaptchaV2TaskProxyLess"
) -> str:
"""
通用验证码解决工具,支持多种验证码类型
参数:
website_url: 出现验证码的网页URL
website_key: 验证码的站点密钥
captcha_type: 验证码类型,默认为reCAPTCHA v2
返回:
验证码解决方案token,或错误信息
"""
api_key = os.getenv("CAPSOLVER_API_KEY")
payload = {
"clientKey": api_key,
"task": {
"type": captcha_type,
"websiteURL": website_url,
"websiteKey": website_key
}
}
try:
# 创建验证码解决任务
response = requests.post(
"https://api.capsolver.com/createTask",
json=payload,
timeout=10
)
result = response.json()
if result.get("errorId") != 0:
return f"错误: {result.get('errorDescription')}"
task_id = result["taskId"]
# 轮询获取结果
for _ in range(30): # 最多尝试30次,每次间隔2秒
time.sleep(2)
result = requests.post(
"https://api.capsolver.com/getTaskResult",
json={"clientKey": api_key, "taskId": task_id},
timeout=10
).json()
if result.get("status") == "ready":
return result["solution"].get("gRecaptchaResponse") or result["solution"].get("token")
elif result.get("status") == "failed":
return f"验证码解决失败: {result.get('errorDescription')}"
return "错误: 验证码解决超时"
except Exception as e:
return f"API请求异常: {str(e)}"
这个工具实现了与CapSolver API的基本交互流程:
- 创建验证码解决任务
- 定期轮询任务状态
- 返回解决方案或错误信息
3.2 专用验证码解决工具
针对不同类型的验证码,我们可以创建专用工具以提高使用便利性:
reCAPTCHA v2解决工具
python复制@tool
def solve_recaptcha_v2(website_url: str, website_key: str) -> str:
"""
专门解决reCAPTCHA v2验证码
参数:
website_url: 出现验证码的网页URL
website_key: reCAPTCHA的站点密钥
返回:
g-recaptcha-response token
"""
return solve_captcha(
website_url=website_url,
website_key=website_key,
captcha_type="ReCaptchaV2TaskProxyLess"
)
Cloudflare Turnstile解决工具
python复制@tool
def solve_turnstile(website_url: str, website_key: str) -> str:
"""
解决Cloudflare Turnstile验证码
参数:
website_url: 出现验证码的网页URL
website_key: Turnstile的站点密钥
返回:
Turnstile解决方案token
"""
return solve_captcha(
website_url=website_url,
website_key=website_key,
captcha_type="AntiTurnstileTaskProxyLess"
)
4. 创建Agno代理
4.1 基础代理配置
下面创建一个具备验证码解决能力的Agno代理:
python复制from agno.agent import Agent
from agno.models.openai import OpenAIChat
# 创建代理实例
captcha_agent = Agent(
name="CAPTCHA Solver Agent",
model=OpenAIChat(id="gpt-4"), # 使用GPT-4作为底层模型
tools=[solve_captcha, solve_recaptcha_v2, solve_turnstile],
description="专门用于解决各种类型验证码的AI代理",
instructions=[
"你是一个专业的验证码解决专家",
"当遇到验证码时,首先准确识别其类型",
"根据验证码类型选择合适的解决工具",
"确保提供的URL和站点密钥准确无误",
"如果解决失败,分析原因并尝试其他方法"
]
)
4.2 代理运行示例
python复制# 定义任务
task = """
请帮我解决以下reCAPTCHA v2验证码:
网站URL: https://example.com/protected-page
站点密钥: 6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-
解决后请返回获得的token。
"""
# 运行代理
response = captcha_agent.run(task)
print(response.content)
5. 高级功能实现
5.1 验证码自动检测
为了使代理能够自动检测验证码类型,我们可以扩展其能力:
python复制@tool
def detect_captcha_type(html_content: str) -> str:
"""
分析HTML内容,检测使用的验证码类型
参数:
html_content: 网页HTML内容
返回:
验证码类型标识字符串
"""
if 'recaptcha/api.js' in html_content:
return "ReCaptchaV2TaskProxyLess"
elif 'turnstile.js' in html_content:
return "AntiTurnstileTaskProxyLess"
# 可以添加更多验证码类型的检测逻辑
return "Unknown"
5.2 带自动检测的解决流程
python复制from agno.tools import tool
@tool
def solve_detected_captcha(website_url: str, html_content: str) -> str:
"""
自动检测并解决验证码
参数:
website_url: 网页URL
html_content: 网页HTML内容
返回:
验证码解决方案token
"""
# 检测验证码类型
captcha_type = detect_captcha_type(html_content)
if captcha_type == "Unknown":
return "错误: 无法识别的验证码类型"
# 获取站点密钥
if captcha_type == "ReCaptchaV2TaskProxyLess":
site_key = extract_recaptcha_sitekey(html_content)
elif captcha_type == "AntiTurnstileTaskProxyLess":
site_key = extract_turnstile_sitekey(html_content)
else:
return "错误: 不支持的验证码类型"
if not site_key:
return "错误: 无法提取站点密钥"
# 调用相应的解决工具
return solve_captcha(
website_url=website_url,
website_key=site_key,
captcha_type=captcha_type
)
6. 多代理协作系统
Agno的强大之处在于支持多代理协作。我们可以建立一个专门处理验证码的团队:
python复制from agno.team import Team
# 创建专门解决验证码的代理
captcha_specialist = Agent(
name="CAPTCHA Specialist",
model=OpenAIChat(id="gpt-4"),
tools=[solve_recaptcha_v2, solve_turnstile, detect_captcha_type],
description="验证码识别和解决专家",
instructions=[
"你专门负责识别和解决各种验证码",
"当收到验证码解决请求时,先准确识别类型",
"根据类型选择最合适的解决方法",
"确保解决方案的准确性和及时性"
]
)
# 创建负责数据提取的代理
data_extractor = Agent(
name="Data Extractor",
model=OpenAIChat(id="gpt-4"),
description="网页数据提取专家",
instructions=[
"你负责从网页中提取结构化数据",
"当遇到验证码时,请求CAPTCHA Specialist协助",
"验证码解决后继续数据提取工作",
"确保提取数据的准确性和完整性"
]
)
# 创建团队
scraping_team = Team(
name="Web Scraping Team",
agents=[captcha_specialist, data_extractor],
description="自动化网页抓取团队,具备验证码解决能力"
)
7. 实际应用示例
7.1 自动化登录流程
下面展示如何使用这套系统自动化处理带验证码的登录流程:
python复制from selenium import webdriver
from selenium.webdriver.common.by import By
def automated_login(url, username, password):
# 初始化浏览器
driver = webdriver.Chrome()
driver.get(url)
try:
# 填写登录表单
driver.find_element(By.ID, "username").send_keys(username)
driver.find_element(By.ID, "password").send_keys(password)
# 检测是否有验证码
if "recaptcha" in driver.page_source:
# 获取验证码信息
site_key = driver.find_element(
By.CSS_SELECTOR, "[data-sitekey]"
).get_attribute("data-sitekey")
# 使用Agno代理解决验证码
task = f"""
解决以下reCAPTCHA验证码:
网站URL: {url}
站点密钥: {site_key}
"""
response = captcha_agent.run(task)
token = response.content
# 注入token并提交表单
driver.execute_script(
"document.getElementById('g-recaptcha-response').value = arguments[0];",
token
)
# 提交登录表单
driver.find_element(By.ID, "login-btn").click()
# 验证登录是否成功
if "Welcome" in driver.title:
print("登录成功")
else:
print("登录失败")
finally:
driver.quit()
8. 性能优化与最佳实践
8.1 实现请求重试机制
python复制from functools import wraps
import time
def retry(max_attempts=3, delay=1):
"""
重试装饰器,用于处理暂时性失败
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_attempts - 1:
time.sleep(delay * (attempt + 1))
continue
raise last_exception
return wrapper
return decorator
@retry(max_attempts=3, delay=2)
def solve_captcha_with_retry(website_url, website_key, captcha_type):
"""
带重试机制的验证码解决函数
"""
return solve_captcha(website_url, website_key, captcha_type)
8.2 异步处理提高效率
python复制import asyncio
import aiohttp
async def solve_captchas_async(captcha_tasks):
"""
异步批量解决验证码
"""
async with aiohttp.ClientSession() as session:
tasks = []
for task in captcha_tasks:
tasks.append(
solve_single_captcha_async(
session,
task["website_url"],
task["website_key"],
task["captcha_type"]
)
)
return await asyncio.gather(*tasks)
async def solve_single_captcha_async(session, website_url, website_key, captcha_type):
"""
异步解决单个验证码
"""
payload = {
"clientKey": os.getenv("CAPSOLVER_API_KEY"),
"task": {
"type": captcha_type,
"websiteURL": website_url,
"websiteKey": website_key
}
}
async with session.post(
"https://api.capsolver.com/createTask",
json=payload
) as response:
result = await response.json()
if result.get("errorId") != 0:
raise Exception(result.get("errorDescription"))
task_id = result["taskId"]
for _ in range(30):
await asyncio.sleep(2)
async with session.post(
"https://api.capsolver.com/getTaskResult",
json={"clientKey": os.getenv("CAPSOLVER_API_KEY"), "taskId": task_id}
) as response:
result = await response.json()
if result.get("status") == "ready":
return result["solution"].get("gRecaptchaResponse") or result["solution"].get("token")
elif result.get("status") == "failed":
raise Exception(result.get("errorDescription"))
raise Exception("验证码解决超时")
9. 错误处理与日志记录
9.1 实现详细的错误处理
python复制import logging
from typing import Optional, Dict, Any
class CaptchaSolver:
def __init__(self):
self.logger = logging.getLogger("CaptchaSolver")
logging.basicConfig(level=logging.INFO)
def solve(self, website_url: str, website_key: str, captcha_type: str) -> Optional[str]:
"""
增强版的验证码解决方法,包含详细的错误处理
"""
try:
self.logger.info(f"开始解决验证码: {captcha_type}")
# 验证输入参数
if not all([website_url, website_key, captcha_type]):
raise ValueError("缺少必要参数")
# 调用API
result = self._call_capsolver_api(
website_url,
website_key,
captcha_type
)
if "error" in result:
self.logger.error(f"验证码解决失败: {result['error']}")
return None
self.logger.info("验证码解决成功")
return result["solution"]
except Exception as e:
self.logger.error(f"验证码解决过程中发生异常: {str(e)}", exc_info=True)
return None
def _call_capsolver_api(self, website_url: str, website_key: str,
captcha_type: str) -> Dict[str, Any]:
"""
调用CapSolver API的内部方法
"""
try:
payload = {
"clientKey": os.getenv("CAPSOLVER_API_KEY"),
"task": {
"type": captcha_type,
"websiteURL": website_url,
"websiteKey": website_key
}
}
response = requests.post(
"https://api.capsolver.com/createTask",
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
if result.get("errorId") != 0:
return {"error": result.get("errorDescription", "未知错误")}
task_id = result["taskId"]
return self._poll_task_result(task_id)
except requests.exceptions.RequestException as e:
return {"error": f"API请求失败: {str(e)}"}
def _poll_task_result(self, task_id: str) -> Dict[str, Any]:
"""
轮询任务结果的内部方法
"""
for attempt in range(30):
time.sleep(2)
try:
response = requests.post(
"https://api.capsolver.com/getTaskResult",
json={
"clientKey": os.getenv("CAPSOLVER_API_KEY"),
"taskId": task_id
},
timeout=10
)
response.raise_for_status()
result = response.json()
if result.get("status") == "ready":
return {"solution": result["solution"]}
elif result.get("status") == "failed":
return {"error": result.get("errorDescription", "验证码解决失败")}
except requests.exceptions.RequestException as e:
if attempt == 29: # 最后一次尝试
return {"error": f"轮询结果失败: {str(e)}"}
continue
return {"error": "验证码解决超时"}
10. 实际应用中的注意事项
10.1 验证码解决的最佳实践
-
合理控制请求频率:避免短时间内发送大量验证码解决请求,这可能导致服务限制或额外费用。
-
验证解决方案的有效性:在使用获得的token前,检查其格式和长度是否符合预期。
-
处理各种边界情况:
- 网络超时
- API配额不足
- 不支持的验证码类型
- 无效的站点密钥
-
实现本地缓存:对于短时间内重复出现的相同验证码,可以考虑缓存解决方案。
10.2 成本控制策略
- 监控API使用情况:定期检查CapSolver账户余额和使用统计。
python复制@tool
def check_capsolver_balance() -> dict:
"""
检查CapSolver账户余额和使用情况
"""
try:
response = requests.post(
"https://api.capsolver.com/getBalance",
json={"clientKey": os.getenv("CAPSOLVER_API_KEY")},
timeout=10
)
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
-
优化验证码解决流程:确保只在必要时才调用验证码解决服务。
-
使用合适的套餐:根据业务量选择合适的CapSolver套餐,大宗使用通常有折扣。
10.3 性能优化技巧
-
并行处理多个验证码:当需要解决多个验证码时,使用异步方法可以显著提高效率。
-
预解决验证码:对于已知会出现的验证码,可以提前解决并缓存结果。
-
连接池管理:对于高频使用场景,维护一个HTTP连接池可以减少连接建立的开销。
11. 安全与合规考虑
11.1 数据隐私保护
-
敏感信息处理:确保不在日志或错误信息中泄露API密钥等敏感数据。
-
安全存储凭证:使用安全的秘密管理工具存储API密钥,而不是硬编码在脚本中。
-
最小权限原则:为自动化脚本分配仅具有必要权限的API密钥。
11.2 合规使用建议
-
遵守目标网站条款:在使用自动化工具前,确保了解并遵守目标网站的使用条款。
-
尊重robots.txt:检查目标网站的robots.txt文件,尊重其爬虫政策。
-
合理设置请求间隔:在自动化流程中添加适当的延迟,避免对目标网站造成过大负载。
12. 扩展与定制
12.1 支持更多验证码类型
CapSolver支持多种验证码类型,可以根据需要扩展工具集:
python复制@tool
def solve_hcaptcha(website_url: str, website_key: str) -> str:
"""
解决hCaptcha验证码
"""
return solve_captcha(
website_url=website_url,
website_key=website_key,
captcha_type="HCaptchaTaskProxyLess"
)
@tool
def solve_recaptcha_v3(website_url: str, website_key: str, page_action: str) -> str:
"""
解决reCAPTCHA v3验证码
"""
api_key = os.getenv("CAPSOLVER_API_KEY")
payload = {
"clientKey": api_key,
"task": {
"type": "ReCaptchaV3TaskProxyLess",
"websiteURL": website_url,
"websiteKey": website_key,
"pageAction": page_action,
"minScore": 0.7 # 可调整的最低分数要求
}
}
response = requests.post(
"https://api.capsolver.com/createTask",
json=payload,
timeout=10
)
result = response.json()
if result.get("errorId") != 0:
return f"错误: {result.get('errorDescription')}"
task_id = result["taskId"]
for _ in range(30):
time.sleep(2)
result = requests.post(
"https://api.capsolver.com/getTaskResult",
json={"clientKey": api_key, "taskId": task_id},
timeout=10
).json()
if result.get("status") == "ready":
return result["solution"]["gRecaptchaResponse"]
elif result.get("status") == "failed":
return f"验证码解决失败: {result.get('errorDescription')}"
return "错误: 验证码解决超时"
12.2 自定义解决策略
根据业务需求,可以实现更复杂的解决策略:
python复制@tool
def solve_with_fallback(website_url: str, website_key: str) -> str:
"""
带备用策略的验证码解决方法
"""
# 首先尝试reCAPTCHA v2
result = solve_recaptcha_v2(website_url, website_key)
if not result.startswith("错误"):
return result
# 如果失败,尝试hCaptcha
result = solve_hcaptcha(website_url, website_key)
if not result.startswith("错误"):
return result
return "错误: 所有验证码解决方法均失败"
13. 监控与维护
13.1 实现健康检查
python复制@tool
def check_service_health() -> dict:
"""
检查CapSolver服务健康状况
"""
try:
start_time = time.time()
response = requests.post(
"https://api.capsolver.com/getBalance",
json={"clientKey": os.getenv("CAPSOLVER_API_KEY")},
timeout=10
)
response_time = time.time() - start_time
if response.status_code == 200:
return {
"status": "healthy",
"response_time": response_time,
"balance": response.json().get("balance")
}
else:
return {
"status": "unhealthy",
"status_code": response.status_code,
"response_time": response_time
}
except Exception as e:
return {
"status": "error",
"error": str(e)
}
13.2 建立监控系统
建议实现以下监控指标:
- 验证码解决成功率
- 平均解决时间
- API错误率
- 账户余额变化
可以使用Prometheus、Grafana等工具建立可视化监控面板。
14. 常见问题排查
14.1 验证码解决失败常见原因
-
无效的API密钥:
- 检查环境变量是否正确设置
- 验证密钥是否过期或被撤销
-
账户余额不足:
- 检查账户余额
- 确保有足够的额度解决当前类型的验证码
-
不支持的验证码类型:
- 确认CapSolver支持目标验证码类型
- 检查文档获取最新支持列表
-
网络问题:
- 检查网络连接
- 确保没有防火墙阻止API请求
-
参数错误:
- 验证website_url和website_key是否正确
- 检查captcha_type是否使用正确的标识符
14.2 调试技巧
- 启用详细日志:
python复制import logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
- 检查原始API响应:
python复制print(response.json()) # 在关键步骤打印API响应
- 验证输入参数:
python复制print(f"URL: {website_url}, Key: {website_key}, Type: {captcha_type}")
- 测试简单案例:
使用已知有效的测试网站和验证码进行隔离测试。
15. 总结与建议
Agno与CapSolver的集成为自动化流程中的验证码问题提供了强大的解决方案。通过将CapSolver的能力封装为Agno工具,我们可以创建能够自主处理验证码的智能代理。
在实际应用中,建议:
- 逐步实施:从简单的用例开始,逐步扩展到更复杂的场景。
- 充分测试:在各种条件下全面测试解决方案。
- 监控优化:持续监控性能指标并根据需要进行优化。
- 保持更新:关注Agno和CapSolver的更新,及时利用新功能和改进。
这种集成特别适合以下场景:
- 大规模数据采集项目
- 需要处理多种验证码的自动化流程
- 对隐私和控制有严格要求的环境
- 需要高可靠性和弹性的生产系统
通过合理的设计和实现,Agno与CapSolver的组合可以显著提高自动化项目的成功率和效率。
