CapSkip ile Bir Ajana LangChain CAPTCHA Aracı Nasıl Eklenir

LangChain ile CAPTCHA çözmek için ajana CapSkip’i çağıran bir araç verin ve token’ı konuşmanın dışında tutun. Bariz araç token’ı modele döndürür; bu da bir sonraki model çağrısına yaklaşık 1.700 karakter koyar ve ondan sonraki her çağrıya ikinci bir kopya ekler. Daha iyi bir araç kısa bir handle (token’ın yerini tutan kısa bir kimlik) döndürür, token’ı sizin tarafınızda tutar ve onu ikinci bir araçta harcar. Çözüm aracı CapSkip’in istisnalarını yakalamalıdır, çünkü araçtan kaçan bir istisna tüm çalıştırmayı sonlandırır. CapSkip kendi Windows bilgisayarınızda çalışır, dolayısıyla yeniden deneyen bir ajanın çözüm başına faturası olmaz. Bu rehber bu araç çiftini LangChain 1.4 ile oluşturur ve API anahtarı olmadan çalıştırır.

Neye ihtiyacınız var

• reCAPTCHA çözücüsü açık hâlde bir Windows bilgisayarında çalışan CapSkip. İlk başlatmayı kurulum kılavuzu adım adım anlatıyor.
• Python 3.10 veya üzeri. LangChain 1.4 de capskip 1.3.0 paketi de bunu gerektirir. Aşağıdaki her şey Windows 11 üzerinde Python 3.14.0, langchain 1.4.6, langchain-core 1.6.12, langgraph 1.2.16, pydantic 2.14.1 ve requests 2.34.2 ile çalıştırıldı. Bloklar ayrıca Python 3.11.7 üzerinde de geçti. Python 3.10, 3.12 ve 3.13 denenmedi.
• Bir PowerShell istemi. Komutlar PowerShell sözdizimini kullanır.
• Sahibi olduğunuz, reCAPTCHA v2 onay kutusu bulunan bir kayıt sayfası. Örnek formda _csrf adlı gizli bir alan ve email adlı bir alan vardır. Submit aracı bu ikisini ve token’ı gönderir, bu yüzden adları kendi formunuza uyacak şekilde değiştirin.
• Çözücü için bir adres. Local modunda CapSkip yalnızca o cihaz için 127.0.0.1 üzerinden yanıt verir. Server modunda ağ adresinizi veya genel IP’nizi dinler, böylece başka bir bilgisayardaki, bir VPS’teki ya da barındırılan bir platformdaki bir ajan onu API üzerinden çağırabilir. İkisi de şurada ayarlanır: bağlantı ayarları.

Bu rehber yazılırken hiçbir dil modeli çağrılmadı. Ajan çalıştırmaları, araç çağrılarını sabit bir sırayla yapan senaryolu bir model (scripted model) kullanır; böylece bağlantılar gerçekten çalışır, ama bu, gerçek bir modelin nasıl davrandığı hakkında hiçbir şey söylemez. Ölçümler, aynı bilgisayardaki sahte (mock) bir CapSkip ve sahte bir kayıt sayfasından, gerçek LangChain ve gerçek SDK ile alındı. Bu rehber onlara mock çalıştırmaları ve mock sayfa der. Mock sayfanın kabul ve ret kuralları, mock’un çözümleri sıraya alma biçimi ve tüm süreler, bu bilgisayarda mock’un kendisine aittir. Çalışan bir CapSkip’e karşı, Google’ın herkese açık test sitekey’i ve varsayılan ayarlar kullanılarak yapılan bir çözüm 12,9 saniye sürdü ve 1.700 karakterlik bir token döndürdü. Çözüm süreleri çalıştırmadan çalıştırmaya değişir.

Bir araç neden bir betikten daha fazla özen gerektirir

Bir betikte argümanları siz seçersiniz ve sonucu siz okursunuz. Bir ajanda argümanları model seçer, sonucu model okur ve ajanın okuduğu bir sayfa modeli yönlendirmeye çalışabilir. Bu dört şeyi değiştirir. Sonuç konuşmaya düşer ve sonraki her çağrıda yeniden gönderilir. Araç içindeki bir istisna sizin için ele alınmaz. Model, istediği kadar sık yeni bir çözüm isteyebilir. Ve modelden gelen bir sitekey ile adresin, CapSkip’e ulaşmadan önce kontrol edilmesi gerekir. Tarayıcıyı kendi başına süren bir ajan aynı sorunlara sahiptir ve o yolu Browser Use rehberi ele alıyor. Buradaki ise modelin çağırabileceği düz bir fonksiyon.

1. Adım: projeyi oluşturun

Yeni bir klasörde bir sanal ortam kullanın. Komutlar ortamın kendi Python’unu yoluyla çalıştırır, bu yüzden onu etkinleştirmeniz gerekmez. langchain paketi langgraph, langchain-core ve pydantic’i de çeker; toplamda yaklaşık 40 paket. Model sağlayıcıları ayrı paketlerdir ve yalnızca gerçek bir model kullandığınızda eklersiniz.

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

Ajan için langchain.agents içindeki create_agent fonksiyonunu kullanın. Eski yardımcılar kalktı. AgentExecutor, initialize_agent ve create_react_agent, langchain.agents içinden import edildiğinde ImportError fırlatır. langgraph.prebuilt içindeki create_react_agent hâlâ çalışır ve create_agent’ı langchain.agents içinden import etmenizi söyleyen bir LangGraphDeprecatedSinceV10 uyarısı yazdırır. Bunlardan birini kullanan bir öğretici güncel değildir.

2. Adım: ilk araç ve neden yanlış biçimde olduğu

Bunu naive_tool.py olarak kaydedin. En kısa yararlı araçtır: CAPTCHA’yı çözer ve token’ı geri verir. SDK hiçbir ortam değişkenini okumaz, bu yüzden dosya host, port ve anahtarı kendisi okur. Anahtar yalnızca CapSkip’te API key validation’ı açtıysanız önemlidir. Her dosya sayfa adresini PAGE_URL değişkeninden, sitekey’i SITEKEY değişkeninden okur; bu yüzden ikisini de kendi kayıt sayfanızdan, PowerShell pencerenizde bir kez ayarlayın. CapSkip, reCAPTCHA v2 widget’ı taşımayan bir sayfa için bir hatayla yanıt verir.

# 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

Sonra naive_tool.py dosyasını .\.venv\Scripts\python naive_tool.py ile çalıştırın.

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

Çalışır ve token’ın uzunluğunu yazdırır. Sorun o uzunluktur. Google’ın herkese açık test sitekey’ine karşı çözülmüş bir reCAPTCHA token’ı yaklaşık 1.700 karakter tuttu. Araç onu döndürdüğünde token, araç sonucunda bir kez saklanır. Model sonra onu sonraki araca argüman olarak geçirmek zorundadır, bu yüzden modelin kendi araç çağrısında ikinci kez saklanır. Bundan sonraki her çağrı iki kopyayı da taşır. Bir mock çalıştırması, üç çağrının her birinde modele gönderilen karakterleri, mesaj metni artı araç çağrısı argümanları olarak saydı. Senaryolu bir model, bir karakterlik sistem istemi, 2.000 karakterlik bir mock token ve ilk satır için token’ı argüman olarak alan ikinci bir araç kullandı. Gerçek bir sistem istemi ve araç şemaları her iki satıra da aynı miktarı ekler:

Ölçek için: rastgele 2.000 karakterlik bir dize, bir OpenAI tokenizer’ıyla yaklaşık 1.400 token saydı. Diğer modeller farklı sayar. Bu, bir maliyet ve boyut argümanıdır; bir modelin uzun bir dizeyi ne sıklıkla yanlış kopyaladığına dair bir iddia değildir. Aynı zamanda, bir kopyalama hatasının bir kez çalışan bir token’ı bozabileceği bir yer daha. Düzeltme, token’ı kendi tarafınızda tutmak ve modele bir handle vermektir.

3. Adım: gerçek araçlar için ayarlar

Gerçek araçlar, yapıştıracağınız dört parçadan oluşturduğunuz tek bir dosyada, captcha_tools.py içinde yaşar. Bu ilk parçayı kaydedin. Ayarları okur, istemciyi oluşturur ve araçların paylaştığı küçük miktardaki durumu tanımlar. ALLOWED_HOSTS sizin kendi PAGE_URL değerinizden gelir, asla modelden gelmez.

# 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. Adım: her şeyden önce host’u kontrol edin

CapSkip’e giden sayfa adresini model seçer ve ajanın okuduğu bir sayfadaki metin onu başka bir adrese doğru itebilir. Adresi araç içinde zorunlu kılın, çünkü istemdeki bir kural ikna edilerek aşılabilir. Bunu captcha_tools.py dosyasına ekleyin.

# 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

Kontrol portu bilerek okur. Yalnızca host adını karşılaştıran bir kontrol, http://127.0.0.1:18512.evil.example/page gibi bir adresi geçirirdi, çünkü port bozuk olduğunda bile ayrıştırıcı host olarak 127.0.0.1’i bildirir. Portu okumak bu durumda ValueError fırlatır ve fonksiyon adresi reddeder. Ayrıca file:, javascript: ve ftp: adreslerini, boş dizeleri ve içinde kullanıcı adı olan adresleri de reddeder. Yalnızca host’u karşılaştırır, portu değil. 6. Adım’daki submit aracı, formun action adresinde aynı kontrolü çalıştırır.

5. Adım: çözüm aracı

Bunu captcha_tools.py dosyasına ekleyin. İçindeki üç karar önemlidir. Araç her CapSkip istisnasını yakalar ve küçük bir sözlük döndürür, çünkü araçtan kaçan bir istisna tüm ajan çalıştırmasını sonlandırır. Yaptığı her CapSkip çağrısını, başarısız olanlar dahil, sayar; böylece sürekli isteyen bir model CapSkip’i meşgul tutamaz. Ve token’ı bir sözlükte saklar, yalnızca bir handle döndürür.

# 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}

Modelin okuduğu şey docstring’dir; bu yüzden orada her argümanın ne olduğunu ve token’ın konuşmanın dışında kaldığını yazın. Docstring’i olmayan bir araç, tanım anında ValueError: Function must have a docstring if description not provided hatasıyla başarısız olur. Literal tipi şemada bir enum olur. Onun dışındaki bir değer, örneğin v4, fonksiyona hiç ulaşmaz ve CapSkip çağrılmaz. Model, Error invoking tool ‘solve_recaptcha’ ile başlayan, gönderdiği argümanları tekrar eden, kind: Input should be ‘v2’ or ‘v3’ diyen ve Please fix the error and try again ile biten bir mesaj görür.

Bir araç hata fırlattığında ne olur

Çözüm aracının sözlük döndürmesinin nedeni budur. Mock çalıştırmalarında create_agent yalnızca hatalı argüman hatalarını model için bir mesaja çevirdi. Diğer her istisna agent.invoke’tan değişmeden çıktı ve çalıştırmayı sonlandırdı. Bunların her biri, CapSkip’in yerine mock konulmuşken ajandan çıktı:

Bir araçta handle_tool_error değerini True yapmak burada yardımcı olmaz. Bu yalnızca LangChain’in kendi ToolException hatasını kapsar ve CapSkip’in istisnalarından hiçbiri o değildir. Zaman aşımı geç tetiklendi, çünkü SDK sınırını yalnızca yoklamalar arasında kontrol eder. Yukarıdaki araç bunun yerine küçük bir sözlük döndürür: ok değeri false, unsolvable gibi bir hata adı ve modelin buna göre davranabileceği retryable değeri true. Yalnızca UNSOLVABLE içeren bir hata mesajı yeniden denenebilir sayılır. Yanlış API anahtarı gibi diğer her API hatası capskip_rejected olarak döner ve yeniden denenemez. Bağlantı hatası metni gibi ayrıntılar modele değil, log’a gider. Araçları ham tutmayı tercih ederseniz, ajan kütüphanesinin iki kancası vardır. ToolErrorMiddleware, model için metin döndüren bir on_error fonksiyonu alır ve o fonksiyon hiçbir şey döndürmezse istisna yine çalıştırmayı sonlandırır. wrap_tool_call dekoratörü try ve except bloklarını kendiniz yazmanızı sağlar. İkisi de mock çalıştırmalarında çalıştırmayı ayakta tuttu.

6. Adım: submit aracı

captcha_tools.py dosyasının son parçasını ekleyin. Submit aracı bir handle harcar. Token’ı sözlükten çıkarır (pop), bu yüzden bir handle yalnızca bir kez işe yarar ve TOKEN_TTL değerinden eski bir handle’ı reddeder. Token’ın süresi dolar, bu yüzden dosya 100 saniyeye izin verir; bu da Google’ın belgelediği iki dakikanın altındadır. Sayfayı almak ve göndermek için yaklaşık 20 saniye kalır. Submit aracının adres argümanı yoktur, çünkü sayfa adresi token çözüldüğünde saklanmıştır. Formun kendi action adresi de sayfa metnidir, bu yüzden araç onda aynı host kontrolünü çalıştırır ve yönlendirmeleri izlemez. Model token’ın nereye gönderileceğini seçemez.

# 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 başlığı göndermez ve onu kontrol eden bir site 403 ile yanıt verir. Araç bir tane ayarlar. Yukarıdaki koddaki alan adları örnek formunkilerdir ve eşleşmeyen bir form form_not_recognised döndürür. Aynı handle ile ikinci bir gönderim unknown_or_used_handle döndürür ve mock site tam olarak bir gönderi gördü. Araç, sayfanın yanıtının ilk 80 karakterini modele geri verir. Bir sayfadan gelen metin, dışarıdaki sözlerin modele ulaşmasının bir yoludur; bunu istemiyorsanız yalnızca durumu döndürün. Ham token’ı iki kez göndermek, mock’un token already used ile 403 yanıtı vermesine yol açtı; bu mock’un kuralıdır. Gerçek bir token’ın ne kadar süre geçerli kaldığını öğrenmek için reCAPTCHA token süresinin dolması sayfasına bakın.

7. Adım: senaryolu bir model, API anahtarı yok

Bunu fake_model.py olarak kaydedin. Senaryo olarak verdiğiniz araç çağrılarını sırayla döndüren küçük bir sohbet modelidir. Ajan kütüphanesi bir model üzerinde bind_tools çağırır; bu model çağrıyı kabul eder ve şemaları yok sayar. Araçlarınızın, durumlarının ve ajan döngüsünün çalıştığını kanıtlar. Gerçek bir modelin argümanları nasıl seçtiğini ya da bir hatayı nasıl okuduğunu gösteremez.

# 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. Adım: ajanı oluşturun ve çalıştırın

Bunu run_agent.py olarak kaydedin. CAPTCHA_AGENT_MODEL olmadan senaryolu modeli kullanır. Değişken bir sağlayıcı ve model adına ayarlıysa init_chat_model üzerinden gerçek bir model oluşturur. Sistem istemi, bir araç bir hatanın yeniden denenemez olduğunu söylediğinde modele durmasını söyler. .\.venv\Scripts\python run_agent.py ile çalıştırın.

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

Bir çalıştırma konuşmayı yazdırır: kullanıcı mesajı, bir çözüm çağrısı, yalnızca bir handle içeren araç sonucu, bir submit çağrısı, sayfanın yanıtı ve Done. Mock sayfada araç sonucu yalnızca bir handle ve ömrünü içeriyordu ve sayfa tek bir gönderimi kabul etti.

Gerçek bir model kullanın

Modeliniz için sağlayıcı paketini kurun ve anahtarını ile model adını ayarlayın. Araçlarda hiçbir şey değişmez. Bu yalnızca, OpenAI ve Anthropic istek biçimlerini konuşan yerel bir taklit sunucuya karşı, sahte anahtarlarla, langchain-openai 1.7.2 ve langchain-anthropic 1.7.6 kullanılarak çalıştırıldı. Hiçbir gerçek sağlayıcı çağrılmadı.

# 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

Taklit sunucu, her sarmalayıcının ne gönderdiğini gösterdi. ChatOpenAI aracı name, description ve parameters içeren bir function olarak /v1/chat/completions adresine gönderir. ChatAnthropic ise name, description ve input_schema ile /v1/messages adresine gönderir. İkisi de varsayılan olarak tool_choice, strict ya da parallel_tool_calls göndermedi. Şemada sitekey ve page_url zorunluydu, kind ise varsayılanı olan bir enum’du. strict açıkken LangChain, her özellik zorunlu olsun diye şemayı yeniden yazar ve bir sağlayıcının bunu kabul edip etmediği test edilmedi.

Paralel çağrılar ve async araçlar

Bir model tek turda birkaç araç isteyebilir ve LangChain bunları aynı anda çalıştırır. 2 saniyelik bir çözüm mock çalıştırmalarında 3,8 saniye sürer, çünkü SDK yoklamalar arasında sırasıyla 0,25 saniye, 0,5 saniye, 1 saniye ve 2 saniye bekler. Bunlardan üçü tek turda, mock üçüne de yer verdiğinde yine 3,8 saniyede bitti. Mock tek thread ile sınırlandığında mock fazlalıkları sıraya aldı ve aynı tur 7,8 saniye sürdü. captcha_tools.py içindeki _slots adlı semaphore, bu sürecin aynı anda kaç iş açık tutabileceğini sınırlar. Mock çalıştırmalarında zaman kazandırmadı ve mock’un thread sayısının altındaki bir sınır yalnızca zamana mal oldu: mock 2 thread ile çalışırken 1 sınırı 11,3 saniye sürdü, 2 sınırı ise 7,5 saniye. Amacı, bir paralel çağrı patlamasının CapSkip’in Max. Threads değerinden (reCAPTCHA için varsayılan 10) fazla işi sıraya almasını önlemektir, çünkü 250 saniyelik Wait Timeout değerini aşan bir iş başarısız sayılır. Bu kısım test edilmedi. Sınırı 2 veya daha yüksek, en çok Max. Threads kadar ayarlayın.

Düz bir def aracı güvenli varsayılandır. agent.ainvoke altında LangChain onu bir worker thread’inde çalıştırır, bu yüzden olay döngüsü serbest kalır: bu bilgisayarda, 2 saniyelik bir çözüm sırasında döngü hiçbir zaman 0,02 saniyeden uzun takılmadı. İçinde senkron CapSkip istemcisini çağıran bir async def aracı ise tersini yapar. Döngüyü 3,7 saniyenin tamamı boyunca bloke etti. Async istiyorsanız, döngüyü bir seferde 0,3 saniyeye kadar takan AsyncCapSkip’i kullanın; çünkü her istek için, her biri yaklaşık 0,15 saniye süren yeni bir HTTP istemcisi kurar. Bunu async_tool.py olarak kaydedin. Başka dosya gerektirmez. Tek bir araç iki yola da hizmet edebilir:

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

Yalnızca async olan bir araç agent.invoke ile çağrılırsa NotImplementedError: StructuredTool does not support sync invocation hatasıyla başarısız olur. Yukarıdaki iki fonksiyon bunu önler.

Sürekli isteyen bir modeli durdurun

Çözüm sayacı CapSkip’i üç işte tutar ve modelin istemesini engellemez. Başka hiçbir şey onu yakında durdurmaz. create_agent ile varsayılan özyineleme sınırı 9.999 adımdır. Hiç durmayan senaryolu bir model, GraphRecursionError: Recursion limit of 9999 reached without hitting a stop condition hatasından önce bu bilgisayarda bir dakikadan kısa sürede yaklaşık 5.000 model çağrısı yaptı. Kendi sınırınızı ayarlamak için agent.invoke’un config argümanında recursion_limit değerini 7 olarak verin. Aracın kendisindeki bir sınır daha temizdir ve ajan kütüphanesinde bir tane var. Bunu captcha_tools.py ve fake_model.py dosyalarının yanına limited_agent.py olarak kaydedin (ikisini de import eder) ve .\.venv\Scripts\python limited_agent.py ile çalıştırın. Senaryolu modeli hiç durmadan bir çözüm ister.

# 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 değeri end iken çalıştırma dördüncü istekte durdu ve CapSkip üç iş almıştı. Son mesaj, solve_recaptcha araç çağrısı sınırına ulaşıldığını, 3 çağrılık sınıra karşı 4 çağrı olduğunu söylüyordu. error seçeneği bunun yerine ToolCallLimitExceededError fırlatır. Varsayılan olan continue aracı engeller ve modele onu bir daha çağırmamasını söyler; bunu yok sayan bir model özyineleme sınırına kadar devam eder, bu yüzden diğer ikisinden birini ayarlayın.

Tek süreçte iki çalıştırma

captcha_tools.py içindeki vault (token kasası) bir modül değişkenidir, bu yüzden tek süreçteki iki ajan çalıştırması onu paylaşır ve reset_run onu ikisi için de temizler. Aynı anda tek bir çalıştırma için bu sorun değildir. Eşzamanlı çalıştırmalar için vault’u çalıştırma başına tutun. ToolRuntime tipindeki bir araç parametresini LangChain doldurur ve modelin gördüğü şemada görünmez. Her invoke çağrısına bir context nesnesi geçirin; böylece her çalıştırma kendi vault’unu alır. Bunu fake_model.py dosyasının yanına runtime_tools.py olarak kaydedin (onu import eder) ve .\.venv\Scripts\python runtime_tools.py ile çalıştırın:

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

Yazdırdığı ilk satır, modelin gördüğü iki özelliği, sitekey ve page_url değerlerini, runtime olmadan listeler. Çalıştırmadan sonra vault boştur. captcha_tools.py içindeki çözüm sayacı da bir modül değişkenidir ve runtime_tools.py kısa kalmak için host kontrolünü, bütçeyi, zaman sınırını ve hata işlemeyi dışarıda bırakır; kopyalıyorsanız bunları da taşıyın. Bu şekilde oluşturulan bir araç yalnızca bir ajanın içinde çalışır. captcha_tools.py çifti, langgraph 1.2.16 üzerinde hiçbir uyarı olmadan, ToolNode ve tools_condition içeren elle kurulmuş bir LangGraph grafiğinde de değişmeden çalıştı. ToolRuntime sürümü, grafikte context_schema=RunState ve invoke üzerinde bir context argümanı gerektirir.

Ajanı başka bir makinedeki CapSkip’e karşı çalıştırın

Ajan ve çözücünün aynı bilgisayarı paylaşması gerekmez. Bağlantı ayarlarından CapSkip’i Server moduna alın; ağ adresinizi veya genel IP’nizi dinler. Kararlı uzaktan erişim için statik bir genel IP önerilir. Ardından dosyanın zaten okuduğu üç değişkeni ayarlayın. Çözücü hâlâ sizin kendi Windows makinenizdir ve çözüm başına ücretlendirme hâlâ yoktur.

# 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

Üç şeye dikkat edin. SDK, CapSkip ile düz HTTP üzerinden konuşur, bu yüzden API anahtarı açık metin olarak taşınır. Bu bağlantıyı özel bir ağda ya da bir VPN içinde tutun ve API key validation’ı açın. CapSkip, Server modu başladığında portunu her adrese açar; bu yüzden beklemediğiniz adresler için bir Windows Firewall engelleme kuralı onu daraltmanın yoludur. CAPSKIP_HOST içine yalnızca adresi yazın ve bir IP adresi kullanın: bu bilgisayarda reddedilen bir bağlantı 127.0.0.1’e yaklaşık 2 saniye, localhost’a yaklaşık 4 saniye sürdü. http:// ile başlayan bir değer, http adlı host’un çözümlenemediğini söyleyen bir NetworkException ile başarısız olur. Sonunda port bulunan bir değer, host ya da portun geçerli olmadığını söyleyen bir mesajla başarısız olur. Uzaktan kurulum bu rehber için çalıştırılmadı, yalnızca değişkenler okundu.

LangChain CAPTCHA aracında sık görülen hatalar

FAQ

Kısa özet

Bir LangChain CAPTCHA aracı token yerine bir handle döndürmeli, her CapSkip istisnasını yakalayıp küçük bir sözlük döndürmeli, sayfa adresini fonksiyonun içinde kontrol etmeli, çözümleri saymalı ve bir araç çağrısı sınırının arkasında durmalıdır. Token’ı kendi tarafınızda bir sözlükte tutun ve ikinci bir araçta harcayın. Async’e ihtiyacınız olmadıkça düz bir def aracı kullanın.

• Python paketinin desteklediği tüm CAPTCHA türleri: Python CAPTCHA çözücü sayfası.
• Onay kutulu çözücünün nasıl çalıştığı: reCAPTCHA v2 çözücü sayfası.
• Kendi tarayıcısını çalıştıran bir ajan: Browser Use’da CAPTCHA çözme.
• Claude Code ve diğer MCP istemcileri için: MCP CAPTCHA çözücü.

Başarılı olana kadar yeniden deneyen bir ajan ancak her deneme ücretsizse karşılanabilir olur. Kendi Windows makinenizde çalışan bir captcha çözücü kullandığınızda her yeniden deneme birkaç saniyeye mal olur ve başka hiçbir şeye mal olmaz.