如何用 CapSkip 给智能体添加 LangChain 验证码工具

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

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

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

# 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

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

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

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

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

# 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

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

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

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

# 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