CapSkip के साथ agent में LangChain कैप्चा tool कैसे जोड़ें

LangChain कैप्चा हल करने के लिए agent को ऐसा tool दें जो CapSkip को call करे, और token को बातचीत से बाहर रखें। सीधा-सादा tool token को model को लौटा देता है, जिससे अगली model call में करीब 1,700 अक्षर चले जाते हैं और उसके बाद की हर call में उसकी दूसरी copy। बेहतर tool एक छोटा handle लौटाता है, token को आपकी तरफ़ रखता है और उसे दूसरे tool में खर्च करता है। solve tool को CapSkip के exceptions पकड़ने चाहिए, क्योंकि tool से बाहर निकला exception पूरा run ख़त्म कर देता है। CapSkip आपके अपने Windows PC पर चलता है, इसलिए दोबारा कोशिश करने वाले agent पर प्रति-हल कोई बिल नहीं आता। यह गाइड LangChain 1.4 के साथ tools की यह जोड़ी बनाती है और उसे बिना किसी API key के चलाती है।

आपको क्या चाहिए

• किसी Windows PC पर चल रहा CapSkip, जिसमें reCAPTCHA सॉल्वर चालू हो। पहली बार शुरू करने के चरण सेटअप गाइड में समझाए गए हैं।
• Python 3.10 या उसके बाद का वर्ज़न। LangChain 1.4 और capskip 1.3.0 package, दोनों को यही चाहिए। नीचे सब कुछ 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 के साथ चलाया गया। ये blocks Python 3.11.7 पर भी पास हुए। Python 3.10, 3.12 और 3.13 पर नहीं चलाए गए।
• एक PowerShell prompt। कमांड PowerShell syntax में लिखी गई हैं।
• आपका अपना sign-up पेज, जिस पर reCAPTCHA v2 checkbox हो। उदाहरण के form में _csrf नाम का एक hidden field और email नाम का एक field है। submit tool token के साथ यही दो भेजता है, इसलिए नाम अपने form के हिसाब से बदलें।
• सॉल्वर के लिए एक पता। Local mode में CapSkip केवल उसी डिवाइस के लिए 127.0.0.1 पर जवाब देता है। Server mode में वह आपके नेटवर्क पते या पब्लिक IP पर सुनता है, ताकि दूसरे PC, VPS या hosted platform पर चल रहा agent API के ज़रिए उसे कॉल कर सके। दोनों यहाँ सेट किए जाते हैं: कनेक्शन सेटिंग्स.

इस गाइड को लिखते समय किसी language model को call नहीं किया गया। agent के run एक scripted model इस्तेमाल करते हैं जो tool calls एक तय क्रम में करता है, इसलिए wiring सचमुच चलती है और यह कुछ नहीं बताती कि असली model कैसा बर्ताव करेगा। माप उसी PC पर चल रहे mock CapSkip और mock sign-up पेज से आए हैं, जिनके साथ असली LangChain और असली SDK इस्तेमाल हुए। यह गाइड इन्हें mock runs और mock पेज कहती है। mock पेज के accept और reject के नियम, mock का हल queue में लगाने का तरीक़ा और हर समय इस PC पर mock के अपने हैं। चल रहे CapSkip के सामने Google की public test sitekey और default settings के साथ एक हल में 12.9 सेकंड लगे और 1,700 अक्षरों का token मिला। हल का समय हर run में बदलता रहता है।

tool को script से ज़्यादा सावधानी क्यों चाहिए

script में arguments आप चुनते हैं और नतीजा आप पढ़ते हैं। agent में arguments model चुनता है, नतीजा model पढ़ता है, और जो पेज agent ने पढ़ा है वह model को भटकाने की कोशिश कर सकता है। इससे चार चीज़ें बदल जाती हैं। नतीजा conversation में पहुँचता है और हर अगली call पर दोबारा भेजा जाता है। tool के अंदर का exception आपके लिए अपने आप नहीं संभाला जाता। model जितनी बार चाहे एक और हल माँग सकता है। और model से आए sitekey तथा पते को CapSkip तक पहुँचने से पहले जाँचना पड़ता है। जो agent browser को ख़ुद चलाता है, उसके सामने भी यही दिक्कतें आती हैं, और Browser Use गाइड उसी रास्ते को कवर करती है। यह वाला एक सादा function है जिसे model call कर सकता है।

चरण 1: प्रोजेक्ट बनाएँ

नए folder में virtual environment इस्तेमाल करें। कमांड environment का अपना Python उसके path से चलाते हैं, इसलिए उसे activate करने की ज़रूरत नहीं। langchain package अपने साथ langgraph, langchain-core और pydantic खींच लाता है, कुल मिलाकर करीब 40 packages। model providers अलग packages हैं जिन्हें आप तभी जोड़ते हैं जब असली model इस्तेमाल करें।

# 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__)"

agent के लिए langchain.agents से create_agent इस्तेमाल करें। पुराने helpers हट चुके हैं। AgentExecutor, initialize_agent और create_react_agent को langchain.agents से import करने पर ImportError उठता है। langgraph.prebuilt वाला create_react_agent अब भी चलता है, और वह LangGraphDeprecatedSinceV10 warning प्रिंट करता है जो कहती है कि langchain.agents से create_agent import करें। इनमें से किसी का भी इस्तेमाल करने वाला tutorial पुराना है।

चरण 2: पहला tool, और वह ग़लत ढाँचे का क्यों है

इसे naive_tool.py के नाम से सेव करें। यह सबसे छोटा काम का tool है: कैप्चा हल करो और token लौटा दो। SDK कोई environment variable नहीं पढ़ता, इसलिए फ़ाइल host, port और key ख़ुद पढ़ती है। key तभी मायने रखती है जब आपने CapSkip में API key validation चालू की हो। हर फ़ाइल पेज का पता PAGE_URL से और sitekey SITEKEY से पढ़ती है, इसलिए अपने sign-up पेज से दोनों एक बार अपनी PowerShell window में सेट करें। जिस पेज पर reCAPTCHA v2 widget नहीं है, उसके लिए CapSkip error से जवाब देता है।

# 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")

यह चलता है और token की लंबाई प्रिंट करता है। यही लंबाई समस्या है। Google की public test sitekey के सामने हल किया गया reCAPTCHA token करीब 1,700 अक्षरों का निकला। tool उसे लौटाता है तो token tool के नतीजे में एक बार जमा होता है। फिर model को उसे अगले tool को argument के रूप में देना पड़ता है, इसलिए वह model की अपनी tool call में दूसरी बार जमा होता है। उसके बाद की हर call में दोनों copies जाती हैं। एक mock run ने तीन calls में से हर एक पर model को भेजे गए अक्षर गिने, message text और tool call arguments को मिलाकर। इसमें scripted model, एक अक्षर का system prompt, 2,000 अक्षरों का mock token और, पहली पंक्ति के लिए, token को argument लेने वाला दूसरा tool इस्तेमाल हुआ। असली system prompt और tool schemas दोनों पंक्तियों में उतना ही जोड़ते हैं:

अंदाज़े के लिए, 2,000 अक्षरों की एक random string OpenAI tokenizer से करीब 1,400 tokens गिनी गई। दूसरे models अलग गिनते हैं। यह लागत और आकार की दलील है, यह दावा नहीं कि model लंबी string को कितनी बार ग़लत copy करता है। यह एक और जगह भी है जहाँ copy की ग़लती ऐसे token को तोड़ सकती है जो बस एक बार चलता है। इसका हल है token को अपनी तरफ़ रखना और model को handle देना।

चरण 3: असली tools की settings

असली tools एक ही फ़ाइल captcha_tools.py में रहते हैं, जिसे आप चार चिपकाए हुए हिस्सों से बनाते हैं। यह पहला हिस्सा सेव करें। यह settings पढ़ता है, client बनाता है और वह थोड़ी-सी state घोषित करता है जो tools साझा करते हैं। ALLOWED_HOSTS आपके अपने PAGE_URL से आता है, model से कभी नहीं।

# 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: किसी भी चीज़ से पहले host जाँचें

model वह पेज का पता चुनता है जो CapSkip को जाता है, और जिस पेज को agent ने पढ़ा उसका text उसे किसी दूसरे पते की ओर धकेल सकता है। पते की जाँच tool के अंदर लागू करें, क्योंकि prompt में लिखे नियम को बातों में उलझाकर तोड़ा जा सकता है। इसे 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

जाँच जानबूझकर port को छूती है। जो जाँच सिर्फ़ host का नाम मिलाती, वह http://127.0.0.1:18512.evil.example/page जैसे पते को निकल जाने देती, क्योंकि port गड़बड़ होने पर भी parser 127.0.0.1 को ही host बताता है। port पढ़ने पर उस मामले में ValueError उठता है, और function पता ठुकरा देता है। वह file:, javascript: और ftp: पते, ख़ाली strings और ऐसे पते भी ठुकराता है जिनमें user name हो। वह सिर्फ़ host मिलाता है, port नहीं। चरण 6 का submit tool form के action address पर यही जाँच चलाता है।

चरण 5: solve tool

इसे captcha_tools.py में जोड़ें। इसमें तीन फ़ैसले मायने रखते हैं। tool हर CapSkip exception पकड़ता है और एक छोटा dictionary लौटाता है, क्योंकि tool से बाहर निकला exception पूरा agent run ख़त्म कर देता है। वह CapSkip की हर call गिनता है, फेल हुई भी, ताकि बार-बार माँगने वाला model CapSkip को व्यस्त न रख सके। और वह token को एक dictionary में रखता है और सिर्फ़ handle लौटाता है।

# 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 वही है जो model पढ़ता है, इसलिए वहीं लिखें कि हर argument क्या है और token conversation से बाहर रहता है। docstring के बिना tool परिभाषा के समय ValueError: Function must have a docstring if description not provided के साथ फेल हो जाता है। Literal type schema में enum बन जाता है। उससे बाहर का मान, जैसे v4, function तक पहुँचता ही नहीं, और CapSkip को call नहीं जाती। model को एक संदेश दिखता है जो Error invoking tool ‘solve_recaptcha’ से शुरू होता है, भेजे गए arguments दोहराता है, kind: Input should be ‘v2’ or ‘v3’ बताता है और Please fix the error and try again पर ख़त्म होता है।

जब कोई tool exception उठाता है तब क्या होता है

इसीलिए solve tool dictionaries लौटाता है। mock runs में create_agent ने सिर्फ़ ग़लत-argument वाले errors को model के लिए संदेश में बदला। हर दूसरा exception agent.invoke से बिना बदले निकल गया और run ख़त्म हो गया। इनमें से हर एक agent से बाहर आया, जबकि CapSkip की जगह mock था:

किसी tool पर handle_tool_error को True करना यहाँ मदद नहीं करता। वह LangChain के अपने ToolException को कवर करता है, और CapSkip का कोई भी exception वह नहीं है। timeout देर से इसलिए चला क्योंकि SDK अपनी सीमा सिर्फ़ polls के बीच जाँचता है। ऊपर वाला tool इसके बजाय एक छोटा dictionary लौटाता है, जिसमें ok को false पर, unsolvable जैसा error नाम और retryable को true पर रखा गया है, जिस पर model अमल कर सकता है। सिर्फ़ वह error संदेश retryable गिना जाता है जिसमें UNSOLVABLE हो। कोई भी दूसरा API error, जैसे ग़लत API key, capskip_rejected बनकर आता है और retryable नहीं होता। connection error के text जैसा ब्योरा log में जाता है, model को नहीं। अगर आप tools को कच्चा रखना चाहें, तो agent library के पास दो hooks हैं। ToolErrorMiddleware एक on_error function लेता है जो model के लिए text लौटाता है, और अगर वह function कुछ न लौटाए तो exception फिर भी run ख़त्म कर देता है। wrap_tool_call decorator आपको try और except ख़ुद लिखने देता है। mock runs में दोनों ने run को ज़िंदा रखा।

चरण 6: submit tool

captcha_tools.py का आख़िरी हिस्सा जोड़ें। submit tool एक handle खर्च करता है। वह token को dictionary से pop कर लेता है, इसलिए handle एक ही बार चलता है, और TOKEN_TTL से पुराने handle को ठुकरा देता है। token expire हो जाता है, इसलिए फ़ाइल 100 सेकंड देती है, यानी उन दो मिनट से कम जो Google के दस्तावेज़में बताए गए हैं। इससे पेज fetch और post के लिए करीब 20 सेकंड बचते हैं। submit tool में address argument नहीं है, क्योंकि पेज का पता तब सहेजा गया था जब token हल हुआ। form का अपना action address भी पेज का text है, इसलिए tool उस पर वही host जाँच चलाता है और redirects का पीछा नहीं करता। model यह नहीं चुन सकता कि token कहाँ post हो।

# 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 header नहीं भेजता, और जो साइट उसे जाँचती है वह 403 से जवाब देती है। tool एक header सेट करता है। ऊपर के कोड में field के नाम उदाहरण वाले form के हैं, और जो form मेल नहीं खाता वह form_not_recognised लौटाता है। उसी handle से दूसरा submit unknown_or_used_handle लौटाता है, और mock साइट ने ठीक एक post देखा। tool पेज के जवाब के पहले 80 अक्षर model को वापस देता है। पेज का text बाहरी शब्दों के model तक पहुँचने का एक रास्ता है, इसलिए अगर आप यह नहीं चाहते तो सिर्फ़ status लौटाएँ। कच्चे token को दो बार post करने पर mock ने token already used के साथ 403 दिया, जो mock का नियम है। असली token कितनी देर चलता है, इसके लिए reCAPTCHA टोकन की एक्सपायरी देखें।

चरण 7: scripted model, बिना API key के

इसे fake_model.py के नाम से सेव करें। यह एक छोटा chat model है जो आपकी लिखी tool calls को क्रम से लौटाता है। agent library model पर bind_tools को call करती है, और यह उस call को स्वीकार कर लेता है और schemas को अनदेखा करता है। यह साबित करता है कि आपके tools, उनकी state और agent loop काम करते हैं। यह नहीं दिखा सकता कि असली model arguments कैसे चुनता है या error कैसे पढ़ता है।

# 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: agent बनाएँ और चलाएँ

इसे run_agent.py के नाम से सेव करें। CAPTCHA_AGENT_MODEL के बिना यह scripted model इस्तेमाल करती है। variable को provider और model के नाम पर सेट करने पर वह init_chat_model से असली model बनाती है। system prompt model से कहता है कि जब कोई tool बताए कि error retry के लायक नहीं है तब रुक जाए। इसे .\.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])

एक run conversation प्रिंट करता है: user का संदेश, एक solve call, tool का नतीजा जिसमें सिर्फ़ handle है, एक submit call, पेज का जवाब, और Done। mock पेज पर tool के नतीजे में सिर्फ़ handle और उसकी मियाद थी, और पेज ने एक post स्वीकार किया।

असली model इस्तेमाल करें

अपने model के लिए provider package इंस्टॉल करें और उसकी key तथा model का नाम सेट करें। tools में कुछ नहीं बदलता। इसे सिर्फ़ एक local stand-in के सामने चलाया गया जो OpenAI और Anthropic के request formats बोलता है, dummy keys के साथ, langchain-openai 1.7.2 और langchain-anthropic 1.7.6 इस्तेमाल करके। किसी असली provider को call नहीं किया गया।

# 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

stand-in ने दिखाया कि हर wrapper क्या भेजता है। ChatOpenAI tool को name, description और parameters वाले function के रूप में /v1/chat/completions पर post करता है। ChatAnthropic उसे name, description और input_schema के साथ /v1/messages पर post करता है। किसी ने default रूप से tool_choice, strict या parallel_tool_calls नहीं भेजा। schema में sitekey और page_url required थे, और kind एक enum था जिसका default था। strict चालू करने पर LangChain schema को इस तरह बदलता है कि हर property required हो जाए, और provider इसे स्वीकार करता है या नहीं, यह टेस्ट नहीं किया गया।

समानांतर calls और async tools

एक model एक turn में कई tools माँग सकता है, और LangChain उन्हें एक साथ चलाता है। mock runs में 2 सेकंड का एक हल 3.8 सेकंड लेता है, क्योंकि SDK polls के बीच 0.25, 0.5, 1 और 2 सेकंड इंतज़ार करता है। जब mock के पास तीनों के लिए जगह थी, तब एक turn में तीन ऐसे हल भी 3.8 सेकंड में पूरे हुए। mock को एक thread तक सीमित करने पर mock ने बाक़ी को queue में रखा और वही turn 7.8 सेकंड में पूरा हुआ। captcha_tools.py में _slots नाम का semaphore तय करता है कि यह process एक साथ कितने jobs खुले रख सकता है। mock runs में इससे कोई समय नहीं बचा, और mock के thread count से नीचे की सीमा ने सिर्फ़ समय खाया: mock के 2 threads होने पर 1 की सीमा में 11.3 सेकंड लगे, जबकि 2 की सीमा में 7.5 सेकंड। इसका मक़सद यह है कि समानांतर calls की बौछार CapSkip के Max. Threads से ज़्यादा jobs queue में न डाल दे, जो reCAPTCHA के लिए default रूप से 10 है, क्योंकि Wait Timeout के 250 सेकंड से ज़्यादा इंतज़ार करने वाला job फेल कर दिया जाता है। वह हिस्सा टेस्ट नहीं किया गया। सीमा को 2 या उससे ऊपर रखें, Max. Threads तक।

सादा def tool सुरक्षित default है। agent.ainvoke के तहत LangChain उसे एक worker thread में चलाता है, इसलिए event loop खाली रहता है: इस PC पर 2 सेकंड के हल के दौरान loop कभी 0.02 सेकंड से ज़्यादा नहीं रुका। जो async def tool अपने अंदर sync CapSkip client बुलाता है, वह उल्टा करता है। उसने पूरे 3.7 सेकंड loop को block रखा। async चाहिए तो AsyncCapSkip इस्तेमाल करें, जिसने loop को एक बार में 0.3 सेकंड तक रोका, क्योंकि वह हर request के लिए नया HTTP client बनाता है, जिसमें हर बार करीब 0.15 सेकंड लगते हैं। इसे async_tool.py के नाम से सेव करें। इसे किसी और फ़ाइल की ज़रूरत नहीं। एक tool दोनों रास्तों को संभाल सकता है:

# 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 वाला tool जब agent.invoke से call हो तो NotImplementedError: StructuredTool does not support sync invocation के साथ फेल होता है। ऊपर के दोनों functions इससे बचाते हैं।

बार-बार माँगने वाले model को रोकें

हल का counter CapSkip को तीन jobs पर रखता है, पर वह model को माँगने से नहीं रोकता। और कुछ भी उसे जल्दी नहीं रोकता। create_agent के साथ default recursion limit 9,999 steps है। कभी न रुकने वाले scripted model ने इस PC पर एक मिनट से कम में करीब 5,000 model calls कर दीं, और फिर GraphRecursionError: Recursion limit of 9999 reached without hitting a stop condition आया। अपनी सीमा तय करने के लिए agent.invoke के config argument में recursion_limit को 7 के मान से पास करें। tool पर ही सीमा रखना ज़्यादा साफ़ है, और agent library में वह है। इसे limited_agent.py के नाम से captcha_tools.py और fake_model.py के बगल में सेव करें, जिन्हें वह import करती है, और .\.venv\Scripts\python limited_agent.py से चलाएँ। इसका scripted model हल माँगना कभी बंद नहीं करता।

# 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 पर रखने से run चौथी request पर रुक गया और CapSkip को तीन jobs मिले थे। आख़िरी संदेश में लिखा था कि solve_recaptcha tool call की सीमा पूरी हो गई, 3 में से 4 calls के साथ। error विकल्प इसके बजाय ToolCallLimitExceededError उठाता है। default, continue, tool को block करता है और model से कहता है कि उसे फिर call न करे, और जो model यह अनदेखा करे वह recursion limit तक चलता रहता है, इसलिए बाक़ी दो में से कोई एक सेट करें।

एक process में दो run

captcha_tools.py का vault (वह dictionary जिसमें tokens रखे जाते हैं) एक module variable है, इसलिए एक process में दो agent runs उसे साझा करते हैं, और reset_run दोनों के लिए उसे साफ़ कर देता है। एक बार में एक run के लिए यह ठीक है। एक साथ चलते runs के लिए vault को हर run के हिसाब से रखें। ToolRuntime टाइप वाला tool parameter LangChain भरता है और model को दिखने वाले schema में नहीं आता। हर invoke call को एक context object पास करें, और हर run को अपना vault मिलेगा। इसे runtime_tools.py के नाम से fake_model.py के बगल में सेव करें, जिसे वह import करती है, और .\.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)

यह जो पहली लाइन प्रिंट करती है उसमें वे दो properties हैं जो model को दिखती हैं, sitekey और page_url, runtime के बिना। run के बाद vault ख़ाली होता है। captcha_tools.py का solve counter भी एक module variable है, और runtime_tools.py छोटा रखने के लिए host जाँच, budget, समय सीमा और error handling छोड़ देती है, इसलिए कॉपी करें तो इन्हें साथ ले जाएँ। इस तरह बना tool सिर्फ़ agent के अंदर चलता है। captcha_tools.py की जोड़ी ToolNode और tools_condition वाले हाथ से बनाए LangGraph graph में भी बिना बदले चली, langgraph 1.2.16 पर बिना किसी warning के। ToolRuntime वाले संस्करण को graph पर context_schema=RunState और invoke पर context argument चाहिए।

किसी दूसरी मशीन पर चल रहे CapSkip के सामने agent चलाएँ

agent और सॉल्वर को एक ही PC पर होना ज़रूरी नहीं है। कनेक्शन सेटिंग्स में CapSkip को Server mode पर करें, और वह आपके नेटवर्क पते या पब्लिक IP पर सुनेगा। स्थिर रिमोट एक्सेस के लिए स्टैटिक पब्लिक IP सुझाया जाता है। फिर वे तीन variables सेट करें जिन्हें फ़ाइल पहले से पढ़ती है। सॉल्वर अब भी आपकी अपनी 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 key सादे text में जाती है। उस link को private network या VPN पर रखें, और API key validation चालू करें। Server mode शुरू होते ही CapSkip अपना port हर पते के लिए खोल देता है, इसलिए जिन पतों की आपको उम्मीद नहीं उनके लिए Windows Firewall का block rule ही उसे सीमित करने का तरीक़ा है। CAPSKIP_HOST में सिर्फ़ पता रखें, और IP address इस्तेमाल करें: इस PC पर refused connection 127.0.0.1 पर करीब 2 सेकंड और localhost पर करीब 4 सेकंड लगा। http:// से शुरू होने वाला मान NetworkException के साथ फेल होता है, जो कहता है कि वह host http को resolve नहीं कर सका। अंत में port वाला मान इस संदेश के साथ फेल होता है कि host या port वैध नहीं है। remote setup इस गाइड के लिए चलाया नहीं गया, सिर्फ़ variables पढ़े गए।

LangChain कैप्चा tool के आम errors

FAQ

संक्षेप में

LangChain कैप्चा tool को token नहीं, handle लौटाना चाहिए, हर CapSkip exception पकड़कर एक छोटा dictionary लौटाना चाहिए, function के अंदर पेज का पता जाँचना चाहिए, हल गिनने चाहिए, और tool call की सीमा के पीछे रहना चाहिए। token को अपनी तरफ़ एक dictionary में रखें और दूसरे tool में खर्च करें। async की ज़रूरत न हो तो सादा def tool इस्तेमाल करें।

• कैप्चा के वे सभी टाइप जिन्हें Python पैकेज संभालता है: Python कैप्चा सॉल्वर पेज.
• checkbox वाला सॉल्वर कैसे काम करता है: reCAPTCHA v2 सॉल्वर पेज.
• ऐसा agent जो अपना browser ख़ुद चलाता है: Browser Use में कैप्चा हल करना.
• Claude Code और दूसरे MCP clients के लिए: MCP कैप्चा सॉल्वर.

वह agent जो कामयाब होने तक बार-बार कोशिश करता है, तभी किफ़ायती है जब हर कोशिश मुफ़्त हो, और अपनी ही Windows मशीन पर चल रहे कैप्चा सॉल्वर के साथ हर कोशिश में कुछ सेकंड लगते हैं, और कोई ख़र्च नहीं आता।