Как добавить в агент LangChain инструмент для решения капчи с помощью CapSkip

Чтобы решить капчу в LangChain, дайте агенту инструмент, который вызывает CapSkip, и не пускайте токен в диалог. Очевидный инструмент возвращает токен модели, из-за чего примерно 1700 символов попадают в следующий вызов модели и ещё одна копия в каждый последующий вызов. Лучший инструмент возвращает короткий дескриптор, хранит токен на вашей стороне и расходует его во втором инструменте. Инструмент решения должен перехватывать исключения CapSkip, потому что исключение, вырвавшееся из инструмента, завершает весь запуск. CapSkip работает на вашем собственном ПК с Windows, поэтому у агента с повторными попытками нет платы за каждое решение. В этом руководстве такая пара инструментов собирается на LangChain 1.4 и запускается без API-ключа.

Что понадобится

• CapSkip, запущенный на ПК с Windows, с включённым решателем reCAPTCHA. Пошаговое руководство по настройке проведёт вас через первый запуск.
• Python 3.10 или новее. Он нужен и LangChain 1.4, и пакету capskip 1.3.0. Всё ниже запускалось на Windows 11 с Python 3.14.0, langchain 1.4.6, langchain-core 1.6.12, langgraph 1.2.16, pydantic 2.14.1 и requests 2.34.2. Блоки прошли и на Python 3.11.7. Python 3.10, 3.12 и 3.13 не запускались.
• Окно PowerShell. Команды написаны в синтаксисе PowerShell.
• Ваша собственная страница регистрации с чекбоксом reCAPTCHA v2. В примере у формы есть скрытое поле _csrf и поле email. Инструмент отправки передаёт эти два поля и токен, поэтому поменяйте имена под свою форму.
• Адрес решателя. В режиме Local CapSkip отвечает на 127.0.0.1 и только для этого устройства. В режиме Server он слушает ваш сетевой адрес или публичный IP, поэтому агент на другом ПК, VPS или хостинговой платформе может обращаться к нему через API. Оба режима задаются в разделе Настройки подключения.

Языковая модель при подготовке этого руководства не вызывалась. Запуски агента используют сценарную модель, которая делает вызовы инструментов в фиксированном порядке, поэтому связка работает по-настоящему, но ничего не говорит о поведении настоящей модели. Замеры получены на имитации CapSkip и имитации страницы регистрации на том же ПК, с настоящим LangChain и настоящим SDK. В руководстве они называются прогонами на имитации и страницей-имитацией. Правила приёма и отклонения на странице-имитации, то, как имитация ставит решения в очередь, и все тайминги относятся только к самой имитации на этом ПК. Одно решение на работающем CapSkip с публичным тестовым sitekey от Google и настройками по умолчанию заняло 12.9 секунды и вернуло токен из 1700 символов. Время решения меняется от запуска к запуску.

Почему инструменту нужно больше внимания, чем скрипту

В скрипте аргументы выбираете вы и результат читаете вы. В агенте аргументы выбирает модель, результат читает модель, а страница, которую прочитал агент, может попытаться управлять моделью. Из-за этого меняются четыре вещи. Результат попадает в диалог и отправляется заново при каждом следующем вызове. Исключение внутри инструмента никто не обрабатывает за вас. Модель может просить новое решение сколько угодно раз. А sitekey и адрес, пришедшие от модели, нужно проверить, прежде чем они дойдут до CapSkip. У агента, который сам управляет браузером, те же проблемы. Руководство по Browser Use разбирает этот путь. В этом руководстве используется простая функция, которую может вызвать модель.

Шаг 1: создаём проект

Используйте виртуальное окружение в новой папке. Команды запускают Python из этого окружения по его пути, так что активировать окружение не нужно. Пакет langchain подтягивает langgraph, langchain-core и pydantic, всего около 40 пакетов. Провайдеры моделей поставляются отдельными пакетами, которые добавляются только тогда, когда вы используете настоящую модель.

# pip install langchain capskip
mkdir langchain-captcha
cd langchain-captcha
python -m venv .venv
.\.venv\Scripts\python -m pip install langchain capskip
.\.venv\Scripts\python -c "import capskip, langchain; print(langchain.__version__)"

Для агента используйте create_agent из langchain.agents. Старых помощников больше нет. AgentExecutor, initialize_agent и create_react_agent вызывают ImportError при импорте из langchain.agents. create_react_agent из langgraph.prebuilt всё ещё работает и печатает предупреждение LangGraphDeprecatedSinceV10, которое советует импортировать create_agent из langchain.agents. Любой учебник, в котором используется что-то из этого, устарел.

Шаг 2: первый инструмент и почему у него неправильная форма

Сохраните это как naive_tool.py. Это самый короткий полезный инструмент: решить капчу и вернуть токен. SDK не читает переменные окружения, поэтому файл сам читает хост, порт и ключ. Ключ важен, только если вы включили в CapSkip проверку API-ключа. Каждый файл читает адрес страницы из PAGE_URL и sitekey из SITEKEY, поэтому один раз задайте обе переменные в окне PowerShell, взяв значения со своей страницы регистрации. Для страницы без виджета reCAPTCHA v2 CapSkip отвечает ошибкой.

# PowerShell: set these once per window, before you run any file
$env:PAGE_URL = "https://your-site.example/signup"   # your own page
$env:SITEKEY = "YOUR_SITEKEY"                        # its data-sitekey

Затем запустите naive_tool.py командой .\.venv\Scripts\python naive_tool.py.

# naive_tool.py: the first version, token straight back to the model
import os
from capskip import CapSkip
from langchain.tools import tool

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


@tool
def solve_recaptcha(sitekey: str, page_url: str) -> str:
    """Solve a reCAPTCHA v2 checkbox and return the token."""
    return solver.recaptcha(sitekey, page_url)["code"]


if __name__ == "__main__":
    args = {
        "sitekey": os.getenv("SITEKEY", "YOUR_SITEKEY"),
        "page_url": os.getenv(
            "PAGE_URL", "https://example.com/page-with-recaptcha"),
    }
    token = solve_recaptcha.invoke(args)
    print(len(token), "characters")

Он работает и печатает длину токена. Именно эта длина и есть проблема. Токен reCAPTCHA, решённый для публичного тестового sitekey от Google, получился длиной около 1700 символов. Когда инструмент возвращает его, токен один раз сохраняется в результате инструмента. Затем модель должна передать его следующему инструменту аргументом, и он сохраняется второй раз в собственном вызове инструмента модели. Каждый следующий вызов несёт обе копии. Прогон на имитации считал символы, отправленные модели в каждом из трёх вызовов, как текст сообщений плюс аргументы вызовов инструментов. В нём использовались сценарная модель, системный промпт из одного символа, токен-имитация длиной 2000 символов и, для первой строки, второй инструмент, принимающий токен аргументом. Настоящий системный промпт и схемы инструментов добавляют одинаковое количество к обеим строкам:

Для масштаба: случайная строка из 2000 символов насчитала около 1400 токенов в токенизаторе OpenAI. Другие модели считают иначе. Это довод о стоимости и размере, а не утверждение о том, как часто модель неверно копирует длинную строку. Это ещё и одно место, где ошибка копирования могла бы испортить токен, который работает один раз. Решение: хранить токен на своей стороне и давать модели дескриптор.

Шаг 3: настройки для настоящих инструментов

Настоящие инструменты находятся в одном файле, captcha_tools.py, который вы собираете из четырёх вставляемых частей. Сохраните первую часть. Она читает настройки, создаёт клиент и объявляет небольшое состояние, общее для инструментов. ALLOWED_HOSTS берётся из вашего собственного PAGE_URL, а не от модели.

# captcha_tools.py, part 1: settings
import logging
import os
import re
import secrets
import threading
import time
from typing import Literal
from urllib.parse import urljoin, urlparse

import requests
from capskip import ApiException, CapSkip, CapSkipError
from capskip import NetworkException, TimeoutException
from langchain.tools import tool

log = logging.getLogger("captcha_tools")
solver = CapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
)
PAGE_URL = os.getenv("PAGE_URL", "https://example.com/page-with-recaptcha")
SITEKEY = os.getenv("SITEKEY", "YOUR_SITEKEY")
ALLOWED_HOSTS = {urlparse(PAGE_URL).hostname}  # set by you, not the model
MAX_SOLVES = 3        # CapSkip calls allowed per run
TOKEN_TTL = 100       # seconds a stored token may wait for its submit
_slots = threading.Semaphore(2)  # at or below Max. Threads in CapSkip
_vault = {}           # handle -> (token, page_url, born)
_used = [0]
_lock = threading.Lock()

Шаг 4: сначала проверьте хост

Адрес страницы, который уходит в CapSkip, выбирает модель, а текст на странице, прочитанной агентом, может подтолкнуть её к другому адресу. Проверяйте адрес внутри инструмента, потому что правило в промпте можно обойти уговорами. Добавьте это в конец captcha_tools.py.

# captcha_tools.py, part 2: guards (appended to part 1)
def host_ok(url):
    try:
        u = urlparse(url)
        u.port  # raises ValueError on a malformed port
    except ValueError:
        return False
    return (u.scheme in ("http", "https") and u.username is None
            and u.hostname in ALLOWED_HOSTS)


def reset_run():
    _vault.clear()
    _used[0] = 0

Проверка намеренно затрагивает порт. Проверка, сравнивающая только имя хоста, пропустила бы адрес вроде http://127.0.0.1:18512.evil.example/page, потому что парсер всё равно сообщает 127.0.0.1 как хост, когда порт записан неверно. Чтение порта в таком случае вызывает ValueError, и функция отклоняет адрес. Она отклоняет также адреса file:, javascript: и ftp:, пустые строки и адреса с именем пользователя. Сравнивается только хост, но не порт. Инструмент отправки в шаге 6 выполняет ту же проверку для адреса action формы.

Шаг 5: инструмент решения

Добавьте это в конец captcha_tools.py. В нём важны три решения. Инструмент перехватывает каждое исключение CapSkip и возвращает небольшой словарь, потому что исключение, вырвавшееся из инструмента, завершает весь запуск агента. Он считает каждый свой вызов CapSkip, включая неудачные, чтобы модель, которая продолжает просить, не могла держать CapSkip занятым. А токен он сохраняет в словаре и возвращает только дескриптор.

# captcha_tools.py, part 3: the solve tool (appended to part 2)
def fail(error, retryable=False):
    return {"ok": False, "error": error, "retryable": retryable}


@tool
def solve_recaptcha(
    sitekey: str, page_url: str, kind: Literal["v2", "v3"] = "v2"
) -> dict:
    """Solve a reCAPTCHA on a page and return a handle for submit_form.

    sitekey is the data-sitekey value on the page. page_url is the
    full URL of the page that shows the CAPTCHA. The token itself is
    kept out of the conversation; pass the returned handle on.
    """
    if not host_ok(page_url):
        return fail("host_not_allowed")
    with _lock:
        if _used[0] >= MAX_SOLVES:
            return fail("solve_budget_used_up")
        _used[0] += 1
    try:
        with _slots:
            result = solver.recaptcha(sitekey, page_url, version=kind)
    except TimeoutException:
        return fail("timeout", retryable=True)
    except NetworkException as exc:
        log.warning("CapSkip unreachable: %s", exc)
        return fail("capskip_unreachable")
    except ApiException as exc:
        log.warning("CapSkip refused the job: %s", exc)
        unsolvable = "UNSOLVABLE" in str(exc)
        return fail("unsolvable" if unsolvable else "capskip_rejected",
                    retryable=unsolvable)
    except CapSkipError as exc:
        return fail("bad_arguments: %s" % exc)
    handle = "cap_" + secrets.token_hex(3)
    _vault[handle] = (result["code"], page_url, time.monotonic())
    return {"ok": True, "handle": handle, "valid_for_s": TOKEN_TTL}

Модель читает docstring, поэтому опишите в нём, что представляет собой каждый аргумент и что токен остаётся вне диалога. Инструмент без docstring падает ещё при определении с ValueError: Function must have a docstring if description not provided. Тип Literal становится enum в схеме. Значение вне его, например v4, до функции не доходит, и CapSkip не вызывается. Модель видит сообщение, которое начинается с Error invoking tool ‘solve_recaptcha’, повторяет отправленные аргументы, называет kind: Input should be ‘v2’ or ‘v3’ и заканчивается словами Please fix the error and try again.

Что происходит, когда инструмент выбрасывает исключение

Поэтому инструмент решения возвращает словари. В прогонах на имитации create_agent превращал в сообщение для модели только ошибки неверных аргументов. Любое другое исключение выходило из agent.invoke без изменений и завершало запуск. Каждое из этих исключений вышло из агента, а вместо CapSkip стояла имитация:

Установка handle_tool_error в True для инструмента здесь не помогает. Она охватывает только собственный ToolException из LangChain, а ни одно исключение CapSkip им не является. Таймаут сработал поздно, потому что SDK проверяет свой лимит только между опросами. Инструмент выше вместо этого возвращает небольшой словарь, где ok равен false, имя ошибки вроде unsolvable, а retryable равен true, и на это модель может опереться. Повторяемой считается только ошибка, текст которой содержит UNSOLVABLE. Любая другая ошибка API, например неверный API-ключ, возвращается как capskip_rejected и повторяемой не считается. Подробности, например текст ошибки соединения, попадают в журнал, а не к модели. Если хотите оставить инструменты сырыми, в библиотеке агентов есть два хука. ToolErrorMiddleware принимает функцию on_error, которая возвращает текст для модели, и если эта функция ничего не возвращает, исключение всё равно завершает запуск. Декоратор wrap_tool_call позволяет самому написать try и except. Оба варианта сохраняли запуск живым в прогонах на имитации.

Шаг 6: инструмент отправки

Добавьте последнюю часть captcha_tools.py. Инструмент отправки расходует дескриптор. Он вынимает токен из словаря, поэтому дескриптор работает один раз, и отказывается принимать дескриптор старше TOKEN_TTL. Токен истекает, поэтому файл разрешает 100 секунд, меньше двух минут, указанных в документации Google. Остаётся около 20 секунд на загрузку страницы и отправку. У инструмента отправки нет аргумента с адресом, потому что адрес страницы сохранён в момент решения токена. Собственный адрес action формы тоже является текстом страницы, поэтому инструмент выполняет для него ту же проверку хоста и не следует за перенаправлениями. Модель не может выбрать, куда отправляется токен.

# captcha_tools.py, part 4: the submit tool (appended to part 3)
@tool
def submit_form(handle: str, email: str) -> dict:
    """Submit the sign-up form with a handle from solve_recaptcha.

    A handle works once. After a failed submit, solve again.
    """
    entry = _vault.pop(handle, None)  # pop: a handle is single use
    if entry is None:
        return fail("unknown_or_used_handle")
    token, page_url, born = entry
    if time.monotonic() - born > TOKEN_TTL:
        return fail("handle_expired", retryable=True)
    try:
        session = requests.Session()
        page = session.get(page_url, timeout=15, allow_redirects=False)
        html = page.text
        csrf = re.search(r'name="_csrf" value="([^"]+)"', html)
        action = re.search(r'<form[^>]*action="([^"]*)"', html)
        if not (csrf and action):
            return fail("form_not_recognised")
        target = urljoin(page_url, action.group(1))
        if not host_ok(target):  # the form action is page text too
            return fail("host_not_allowed")
        reply = session.post(
            target,
            data={"_csrf": csrf.group(1), "email": email,
                  "g-recaptcha-response": token},
            headers={"Referer": page_url},  # requests sends none itself
            timeout=15,
        )
    except requests.RequestException:
        return fail("site_unreachable")
    said = re.sub(r"<[^>]+>", " ", reply.text).strip()[:80]
    return {"ok": reply.status_code == 200, "status": reply.status_code,
            "page_says": said}

Библиотека requests не отправляет заголовок Referer, а сайт, который его проверяет, отвечает 403. Инструмент задаёт его сам. Имена полей в коде выше принадлежат примеру формы, и несоответствующая форма возвращает form_not_recognised. Повторная отправка с тем же дескриптором возвращает unknown_or_used_handle, а сайт-имитация увидел ровно одну отправку. Инструмент возвращает модели первые 80 символов ответа страницы. Текст со страницы позволяет чужим словам попасть к модели, поэтому возвращайте только статус, если вам это не нужно. Двойная отправка самого токена заставила имитацию ответить 403 с token already used, и это правило самой имитации. Страница про срок действия токена reCAPTCHA объясняет, как долго живёт настоящий токен.

Шаг 7: сценарная модель, без API-ключа

Сохраните это как fake_model.py. Это небольшая чат-модель, которая возвращает заданные вами вызовы инструментов по порядку. Библиотека агентов вызывает bind_tools у модели, а эта модель принимает вызов и игнорирует схемы. Она доказывает, что ваши инструменты, их состояние и цикл агента работают. Она не может показать, как настоящая модель выбирает аргументы или читает ошибку.

# fake_model.py: a scripted chat model, no API key needed
# It calls the tools exactly as scripted, so the wiring runs for
# real. It says nothing about how a real model behaves.
from typing import Any, List

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult


def calls(*items) -> AIMessage:
    """One AI turn asking for tool calls: (name, args, id) each."""
    return AIMessage(content="", tool_calls=[
        {"name": n, "args": a, "id": i, "type": "tool_call"}
        for n, a, i in items
    ])


def call(name: str, args: dict, cid: str) -> AIMessage:
    return calls((name, args, cid))


class ScriptedChat(BaseChatModel):
    script: List[Any]  # AIMessages, or functions(messages) -> AIMessage
    seen: List[Any] = []  # what each model call was sent
    i: int = 0

    @property
    def _llm_type(self) -> str:
        return "scripted"

    def bind_tools(self, tools, **kwargs):
        return self  # the agent calls this; we ignore the schemas

    def _generate(self, messages, stop=None, run_manager=None, **kw):
        self.seen.append(list(messages))
        item = self.script[self.i]
        self.i += 1
        msg = item(messages) if callable(item) else item
        return ChatResult(generations=[ChatGeneration(message=msg)])

Шаг 8: соберите агента и запустите его

Сохраните это как run_agent.py. Без CAPTCHA_AGENT_MODEL он использует сценарную модель. Если задать переменную с именем провайдера и модели, он создаёт настоящую модель через init_chat_model. Системный промпт велит модели остановиться, когда инструмент сообщает, что ошибка не подлежит повтору. Запустите его командой .\.venv\Scripts\python run_agent.py.

# run_agent.py: wire the tools into an agent and run it
import json
import os

from langchain.agents import create_agent
from langchain.messages import AIMessage
from captcha_tools import PAGE_URL, SITEKEY, reset_run
from captcha_tools import solve_recaptcha, submit_form

SYSTEM = (
    "Sign the user up on the page. Call solve_recaptcha, then "
    "submit_form with the handle it returns. Never invent a handle. "
    "If a tool says retryable is false, stop and report the error."
)


def build_model():
    name = os.getenv("CAPTCHA_AGENT_MODEL")  # "provider:model-id"
    if name:
        from langchain.chat_models import init_chat_model
        return init_chat_model(name)
    # No API key: a scripted model that calls the tools in order.
    from fake_model import ScriptedChat, call
    first = call("solve_recaptcha",
                 {"sitekey": SITEKEY, "page_url": PAGE_URL}, "c1")

    def second(msgs):  # read the handle out of the first tool result
        handle = json.loads(msgs[-1].content)["handle"]
        args = {"handle": handle, "email": "[email protected]"}
        return call("submit_form", args, "c2")

    return ScriptedChat(script=[first, second, AIMessage(content="Done.")])


agent = create_agent(build_model(), [solve_recaptcha, submit_form],
                     system_prompt=SYSTEM)

if __name__ == "__main__":
    reset_run()
    task = "Sign up [email protected] at %s. Sitekey: %s"
    out = agent.invoke({"messages": [("user", task % (PAGE_URL, SITEKEY))]})
    for m in out["messages"]:
        print(m.type, "|", str(m.content or m.tool_calls)[:100])

Запуск печатает диалог: сообщение пользователя, вызов решения, результат инструмента, в котором только дескриптор, вызов отправки, ответ страницы и Done. На странице-имитации результат инструмента содержал только дескриптор и срок его жизни, а страница приняла одну отправку.

Используйте настоящую модель

Установите пакет провайдера для вашей модели и задайте её ключ и имя модели. В инструментах ничего не меняется. Это запускалось только на локальной заглушке, которая понимает форматы запросов OpenAI и Anthropic, с фиктивными ключами, с langchain-openai 1.7.2 и langchain-anthropic 1.7.6. Настоящий провайдер не вызывался.

# pip install langchain-anthropic   (or langchain-openai for OpenAI)
.\.venv\Scripts\python -m pip install langchain-anthropic
$env:PAGE_URL = "https://your-site.example/signup"   # your own page
$env:SITEKEY = "YOUR_SITEKEY"
$env:ANTHROPIC_API_KEY = "YOUR_PROVIDER_KEY"
$env:CAPTCHA_AGENT_MODEL = "anthropic:YOUR_MODEL_ID"
.\.venv\Scripts\python run_agent.py

Заглушка показала, что отправляет каждая обёртка. ChatOpenAI отправляет инструмент как функцию с name, description и parameters на /v1/chat/completions. ChatAnthropic отправляет его с name, description и input_schema на /v1/messages. Ни одна из них по умолчанию не отправляла tool_choice, strict или parallel_tool_calls. В схеме sitekey и page_url были обязательными, а kind был enum со значением по умолчанию. При включённом strict LangChain переписывает схему так, что все свойства становятся обязательными, и принимает ли это провайдер, не проверялось.

Параллельные вызовы и асинхронные инструменты

Модель может запросить несколько инструментов за один ход, и LangChain запускает их одновременно. Одно решение в 2 секунды занимает в прогонах на имитации 3.8 секунды, потому что SDK ждёт 0.25, 0.5, 1 и 2 секунды между опросами. Три таких решения за один ход тоже закончились за 3.8 секунды, когда у имитации было место для всех трёх. При ограничении имитации одним потоком она ставила лишние в очередь, и тот же ход занял 7.8 секунды. Семафор _slots в captcha_tools.py ограничивает, сколько заданий этот процесс держит открытыми одновременно. В прогонах на имитации он времени не сэкономил, а лимит ниже числа потоков имитации только стоил времени: при 2 потоках у имитации лимит 1 занял 11.3 секунды против 7.5 секунды при лимите 2. Его цель в том, чтобы всплеск параллельных вызовов не поставил в очередь больше заданий, чем позволяет Max. Threads в CapSkip, которое для reCAPTCHA по умолчанию равно 10, потому что задание, ждущее дольше Wait Timeout в 250 секунд, завершается ошибкой. Эта часть не проверялась. Задайте лимит 2 или выше, вплоть до Max. Threads.

Обычный инструмент на def служит безопасным вариантом по умолчанию. В agent.ainvoke LangChain запускает его в рабочем потоке, поэтому цикл событий остаётся свободным: на этом ПК цикл ни разу не замирал дольше чем на 0.02 секунды во время решения в 2 секунды. Инструмент async def, который внутри вызывает синхронный клиент CapSkip, делает обратное. Он блокировал цикл все 3.7 секунды. Если нужен async, используйте AsyncCapSkip, который останавливал цикл до 0.3 секунды за раз, потому что для каждого запроса создаёт новый HTTP-клиент, примерно по 0.15 секунды на каждый. Сохраните это как async_tool.py. Другие файлы ему не нужны. Один инструмент может обслуживать оба пути:

# async_tool.py: one tool that works under .invoke and .ainvoke
import asyncio
import os

from capskip import AsyncCapSkip, CapSkip
from langchain_core.tools import StructuredTool

KEY = os.getenv("CAPSKIP_API_KEY", "capskip")
HOST = os.getenv("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.getenv("CAPSKIP_PORT", "8080"))
PAGE_URL = os.getenv("PAGE_URL", "https://example.com/page-with-recaptcha")


def solve_sync(sitekey: str, page_url: str) -> dict:
    token = CapSkip(apiKey=KEY, host=HOST, port=PORT).recaptcha(
        sitekey, page_url)["code"]
    return {"ok": True, "token_chars": len(token)}


async def solve_async(sitekey: str, page_url: str) -> dict:
    client = AsyncCapSkip(apiKey=KEY, host=HOST, port=PORT)
    result = await client.recaptcha(sitekey, page_url)
    return {"ok": True, "token_chars": len(result["code"])}


solve = StructuredTool.from_function(
    func=solve_sync,
    coroutine=solve_async,
    name="solve_recaptcha",
    description="Solve a reCAPTCHA v2 on a page; report the token size.",
)

if __name__ == "__main__":
    args = {"sitekey": os.getenv("SITEKEY", "YOUR_SITEKEY"),
            "page_url": PAGE_URL}
    print(solve.invoke(args))
    print(asyncio.run(solve.ainvoke(args)))

Инструмент только с async, вызванный через agent.invoke, падает с NotImplementedError: StructuredTool does not support sync invocation. Две функции выше этого избегают.

Остановите модель, которая продолжает просить

Счётчик решений держит CapSkip на трёх заданиях, но не мешает модели просить дальше. Ничто другое тоже не остановит её скоро. В create_agent лимит рекурсии по умолчанию составляет 9999 шагов. Сценарная модель, которая никогда не останавливалась, сделала около 5000 вызовов модели меньше чем за минуту на этом ПК, прежде чем появилась GraphRecursionError: Recursion limit of 9999 reached without hitting a stop condition. Передайте recursion_limit со значением 7 в аргументе config у agent.invoke, чтобы задать свой лимит. Ограничение на сам инструмент чище, и в библиотеке агентов оно есть. Сохраните это как limited_agent.py рядом с captcha_tools.py и fake_model.py, которые он импортирует, и запустите командой .\.venv\Scripts\python limited_agent.py. Его сценарная модель всё время просит новое решение.

# limited_agent.py: stop a model that keeps asking for solves
from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware
from captcha_tools import PAGE_URL, SITEKEY, reset_run
from captcha_tools import solve_recaptcha, submit_form
from fake_model import ScriptedChat, call

# A scripted model that never stops asking for a solve.
args = {"sitekey": SITEKEY, "page_url": PAGE_URL}
model = ScriptedChat(script=[
    (lambda msgs: call("solve_recaptcha", args, "c%d" % len(msgs)))
] * 50)

agent = create_agent(
    model,
    [solve_recaptcha, submit_form],
    middleware=[ToolCallLimitMiddleware(
        tool_name="solve_recaptcha", run_limit=3, exit_behavior="end")],
)

if __name__ == "__main__":
    reset_run()
    out = agent.invoke({"messages": [("user", "go")]})
    print(len(model.seen), "model calls")
    print(out["messages"][-1].content)

При exit_behavior, равном end, запуск остановился на четвёртом запросе, а CapSkip получил три задания. В последнем сообщении говорилось, что достигнут лимит вызовов инструмента solve_recaptcha: 4 из 3 вызовов. Вариант error вместо этого вызывает ToolCallLimitExceededError. Вариант по умолчанию, continue, блокирует инструмент и просит модель больше его не вызывать, а модель, которая это игнорирует, продолжает до лимита рекурсии, поэтому выберите один из двух других.

Два запуска в одном процессе

Хранилище в captcha_tools.py является переменной модуля, поэтому два запуска агента в одном процессе делят его, а reset_run очищает его для обоих. Для одного запуска за раз это нормально. Для одновременных запусков держите хранилище на каждый запуск. Параметр инструмента с типом ToolRuntime заполняет LangChain, и в схеме, которую видит модель, он не появляется. Передавайте объект контекста в каждый вызов invoke, и у каждого запуска будет своё хранилище. Сохраните это как runtime_tools.py рядом с fake_model.py, который он импортирует, и запустите командой .\.venv\Scripts\python runtime_tools.py:

# runtime_tools.py: a per-run vault through ToolRuntime (hidden arg)
import json
import os
from dataclasses import dataclass, field

from capskip import CapSkip
from langchain.agents import create_agent
from langchain.messages import AIMessage
from langchain.tools import ToolRuntime, tool
from langchain_core.utils.function_calling import convert_to_openai_tool

from fake_model import ScriptedChat, call

solver = CapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
)
PAGE_URL = os.getenv("PAGE_URL", "https://example.com/page-with-recaptcha")


@dataclass
class RunState:  # one instance per agent.invoke(..., context=...)
    vault: dict = field(default_factory=dict)  # handle -> token


@tool
def solve_recaptcha(
    sitekey: str, page_url: str, runtime: ToolRuntime[RunState]
) -> dict:
    """Solve a reCAPTCHA and return a handle for use_token."""
    token = solver.recaptcha(sitekey, page_url)["code"]
    handle = "cap_%d" % (len(runtime.context.vault) + 1)
    runtime.context.vault[handle] = token
    return {"ok": True, "handle": handle}


@tool
def use_token(handle: str, runtime: ToolRuntime[RunState]) -> dict:
    """Spend a handle from solve_recaptcha. It works once."""
    token = runtime.context.vault.pop(handle, None)
    if token is None:
        return {"ok": False, "error": "unknown_or_used_handle"}
    return {"ok": True, "token_chars": len(token)}


if __name__ == "__main__":
    # the model-facing schema has no "runtime" property
    schema = convert_to_openai_tool(solve_recaptcha)
    print(list(schema["function"]["parameters"]["properties"]))
    args = {"sitekey": os.getenv("SITEKEY", "YOUR_SITEKEY"),
            "page_url": PAGE_URL}
    def second(msgs):  # read the handle out of the first tool result
        handle = json.loads(msgs[-1].content)["handle"]
        return call("use_token", {"handle": handle}, "c2")

    model = ScriptedChat(script=[
        call("solve_recaptcha", args, "c1"), second, AIMessage(content="Done."),
    ])
    agent = create_agent(model, [solve_recaptcha, use_token],
                         context_schema=RunState)
    run = RunState()
    out = agent.invoke({"messages": [("user", "go")]}, context=run)
    print(out["messages"][-2].content, "| vault left:", run.vault)

Первая выведенная строка перечисляет два свойства, которые видит модель, sitekey и page_url, без runtime. После запуска хранилище пусто. Счётчик решений в captcha_tools.py тоже является переменной модуля, а runtime_tools.py для краткости опускает проверку хоста, бюджет решений, лимит времени и обработку ошибок, поэтому перенесите их, если копируете его. Инструмент, собранный так, работает только внутри агента. Пара из captcha_tools.py также без изменений запускалась в собранном вручную графе LangGraph с ToolNode и tools_condition, без предупреждений на langgraph 1.2.16. Версии с ToolRuntime нужны context_schema=RunState у графа и аргумент context у invoke.

Запуск агента с CapSkip на другой машине

Агенту и решателю не обязательно работать на одном ПК. Переключите CapSkip в режим Server в настройках подключения, и он будет слушать ваш сетевой адрес или публичный IP. Для стабильного удалённого доступа рекомендуется статический публичный IP. Затем задайте три переменные, которые файл уже читает. Решателем по-прежнему остаётся ваша собственная машина с Windows, и решения по-прежнему не тарифицируются.

# CapSkip on another Windows PC, in Server mode: set these first
$env:PAGE_URL = "https://your-site.example/signup"   # your own page
$env:SITEKEY = "YOUR_SITEKEY"
$env:CAPSKIP_HOST = "192.168.1.50"
$env:CAPSKIP_PORT = "8080"
$env:CAPSKIP_API_KEY = "YOUR_API_KEY"   # only if API key validation is on
.\.venv\Scripts\python run_agent.py

Обратите внимание на три вещи. SDK обращается к CapSkip по обычному HTTP, поэтому API-ключ передаётся открытым текстом. Держите это соединение в частной сети или VPN и включите проверку API-ключа. При запуске режима Server CapSkip открывает свой порт для всех адресов, поэтому сузить доступ можно правилом Windows Firewall, блокирующим адреса, которых вы не ожидаете. Указывайте в CAPSKIP_HOST только адрес и используйте IP-адрес: отказ в соединении занимал около 2 секунд для 127.0.0.1 и около 4 для localhost на этом ПК. Значение, которое начинается с http://, приводит к NetworkException с сообщением о том, что не удалось разрешить хост http. Значение с портом на конце приводит к сообщению о том, что хост или порт недействительны. Удалённая настройка для этого руководства не запускалась, читались только переменные.

Частые ошибки при использовании инструмента капчи в LangChain

FAQ

Коротко

Инструмент капчи для LangChain должен возвращать дескриптор, а не токен, перехватывать каждое исключение CapSkip и возвращать небольшой словарь, проверять адрес страницы внутри функции, считать решения и находиться за лимитом вызовов инструмента. Храните токен в словаре на своей стороне и расходуйте его вторым инструментом. Используйте обычный инструмент на def, если вам не нужен async.

• Все типы капчи, которые поддерживает пакет для Python, перечислены на странице сервиса распознавания капч для Python.
• Как работает решатель reCAPTCHA с чекбоксом, описано на странице сервиса распознавания reCAPTCHA v2.
• Агент, который сам управляет браузером, описан в руководстве по решению капчи в Browser Use.
• Для Claude Code и других MCP-клиентов подойдёт MCP-решатель капчи.

Агент, который повторяет попытки до успеха, оказывается по карману, только если каждая попытка бесплатна, а распознавание капчи на вашем собственном компьютере с Windows обходится в несколько секунд на каждую попытку и больше ни во что.