How to Run a FastAPI CAPTCHA Solver Service on Windows

To run a FastAPI CAPTCHA solver service, create one AsyncCapSkip client when the app starts. Expose a POST endpoint that takes a sitekey and a page URL. Wrap each solve in asyncio.timeout and a Semaphore, and map the SDK’s exceptions to HTTP status codes. Scrapers, workers and scripts in any language then call one internal address and never see CapSkip’s port or key. CapSkip runs on your own Windows machine, so the service adds no per-solve bill. This guide builds it in about 100 lines and shows what each part does when CapSkip is slow, closed or silent.

What you need

• CapSkip running on a Windows PC, with the reCAPTCHA solver switched on. The setup guide walks through the first start.
• Python 3.11 or later, because the code uses asyncio.timeout. Everything below ran on Windows 11 with Python 3.14.0, capskip 1.3.0, fastapi 0.150.0 and uvicorn 0.54.0.
• A PowerShell prompt. The commands use PowerShell syntax.
• A page with a reCAPTCHA v2 checkbox. The service needs only its sitekey and its address.
• An address for CapSkip. In Local mode it answers on 127.0.0.1 for that device only. In Server mode it listens on your network address or public IP, so a service on another machine can call it over the API. Both are set under connection settings.

The numbers in this guide come from a mock CapSkip on the same PC. FastAPI, uvicorn and the SDK are the real ones. The mock holds each solve for a set time, usually two seconds, so your real solves will differ. Nothing here was run against a live CapSkip or a live site.

Why put a service in front of CapSkip

If one Python script solves CAPTCHAs, call the SDK from that script. The Python CAPTCHA solver page covers that, and the parallel solving guide covers many solves in one process. A service pays off when several callers share one CapSkip. Four jobs then belong in one place.

• One deadline. The Python SDK waits up to 300 seconds by default. Its plain client sets no HTTP timeout at all. Callers should not each carry their own timer.
• One limit. CapSkip runs a fixed number of solves at once, and extra jobs wait inside it. A queue you control is easier to reason about.
• One key. Callers get a service key. CapSkip’s address and API key stay in the service’s settings.
• Any language. A C# worker, a PHP site and a PowerShell task all speak HTTP and JSON.

Step 1: install the packages

Create a folder and a virtual environment first if you keep one per project. The capskip package pulls in httpx, which the async client uses.

pip install capskip fastapi uvicorn

Step 2: set the settings as environment variables

The SDK reads no environment variables itself, so the service reads them and passes each value in. Set them in the PowerShell window that will start the service. Replace both placeholder keys. YOUR_CAPSKIP_KEY is the API key from CapSkip’s settings, and any text works while key validation is off in CapSkip. YOUR_API_KEY is a key you invent for your callers.

# Where CapSkip listens and which key it expects.
$env:CAPSKIP_HOST = "127.0.0.1"
$env:CAPSKIP_PORT = "8080"
$env:CAPSKIP_API_KEY = "YOUR_CAPSKIP_KEY"
# Match Max. Threads in the CapSkip settings.
$env:CAPSKIP_MAX_THREADS = "10"
# The key your callers must send to this service.
$env:SOLVER_API_KEY = "YOUR_API_KEY"
# Seconds a caller may wait, queue time included.
$env:SOLVE_DEADLINE = "120"

Use 127.0.0.1 for the host. With this client, a refused connection took about 2.4 seconds on 127.0.0.1 and about 2.7 seconds on localhost. The plain requests client took about 4 seconds on localhost in an earlier test.

Step 3: read the settings

Create main.py. The blocks in Steps 3 to 10 are consecutive parts of that one file. The full file at the end merges their imports at the top.

import asyncio
import os
from contextlib import asynccontextmanager

from capskip import AsyncCapSkip
from fastapi import FastAPI

# The SDK reads no env vars, so pass each setting in by hand.
HOST = os.getenv("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.getenv("CAPSKIP_PORT", "8080"))
KEY = os.getenv("CAPSKIP_API_KEY", "capskip")
MAX_THREADS = int(os.getenv("CAPSKIP_MAX_THREADS", "10"))
DEADLINE = float(os.getenv("SOLVE_DEADLINE", "120"))

Step 4: build one client and one gate when the app starts

A lifespan function runs once at startup. It creates the client and a Semaphore, and puts both on the app state, so every request shares them. Give the Semaphore one slot per CapSkip thread, which is the Max. Threads setting in CapSkip (it defaults to 10).

AsyncCapSkip is a genuine async client in Python. It builds on httpx and polls with asyncio.sleep, so the waiting between polls does not block the event loop. The calls themselves do cost some CPU, as the section called One cost to know about explains. The plain CapSkip client calls requests with no timeout, so a silent CapSkip would hold a thread forever. Setting recaptchaTimeout to the deadline keeps the SDK’s own clock equal to ours. When the two tie, asyncio.timeout fires first.

@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client and one gate, shared by every request.
    app.state.cap = AsyncCapSkip(
        apiKey=KEY, host=HOST, port=PORT,
        recaptchaTimeout=DEADLINE,
    )
    app.state.slots = asyncio.Semaphore(MAX_THREADS)
    yield


app = FastAPI(lifespan=lifespan)

Step 5: check the service key

An unprotected solve endpoint on a network address is a free solver for everyone who can reach the port. This dependency compares the X-API-Key header with secrets.compare_digest, which takes the same time wherever the bytes differ. A missing, empty, wrong, longer and non-ASCII key all got 401 on the mock. The right key got 200.

The check at the top stops the service from starting with an empty key. Without it, a request with no header would default to an empty string and match. The dependency runs before the body is checked, so a request with no key and a body that fails validation gets 401, not 422. A body that is not valid JSON still gets 422, because FastAPI decodes the JSON first. Callers who lack the key learn little about your schema.

import secrets

from fastapi import Depends, Header, HTTPException

SERVICE_KEY = os.environ["SOLVER_API_KEY"]
if not SERVICE_KEY:
    raise SystemExit("SOLVER_API_KEY must not be empty")


def require_key(x_api_key: str = Header(default="")):
    good = secrets.compare_digest(
        x_api_key.encode(), SERVICE_KEY.encode())
    if not good:
        raise HTTPException(
            401, "missing or wrong X-API-Key",
            headers={"WWW-Authenticate": "APIKey"})

Step 6: validate the request and set the deadline

The request model rejects an empty sitekey or a URL that does not start with http. FastAPI answers a bad body with its own 422 and lists each problem.

The deadline is the important part. asyncio.timeout wraps the wait for a slot and the solve together. A lightly loaded service answers within DEADLINE seconds in total. If the timeout wrapped only the solve, a request could queue for minutes before its clock even started. Under heavy load the answer can come later, as the section called One cost to know about explains.

from fastapi import Request
from pydantic import BaseModel, Field


class SolveRequest(BaseModel):
    sitekey: str = Field(min_length=1)
    url: str = Field(pattern=r"^https?://")


async def solve_v2(state, sitekey: str, url: str) -> str:
    # The deadline also covers the wait for a free slot.
    async with asyncio.timeout(DEADLINE):
        async with state.slots:
            result = await state.cap.recaptcha(sitekey, url)
    return result["code"]

Step 7: add the solve endpoint

The endpoint is three lines. The key dependency runs first, then the solve, then the token goes back as JSON.

@app.post("/solve", dependencies=[Depends(require_key)])
async def solve(body: SolveRequest, request: Request):
    token = await solve_v2(request.app.state, body.sitekey, body.url)
    return {"token": token}

Step 8: map the SDK’s exceptions to status codes

The SDK raises four exception classes, all under CapSkipError, and all importable from capskip. One handler on the base class turns each into a status code and a short error name. The detail goes to the log and never to the caller, so CapSkip’s messages stay private.

import logging

from capskip import ApiException, CapSkipError, NetworkException
from capskip import TimeoutException, ValidationException
from fastapi.responses import JSONResponse

STATUS = {
    NetworkException: (503, "capskip_unreachable"),
    TimeoutException: (504, "capskip_timeout"),
    ValidationException: (422, "bad_parameters"),
    ApiException: (502, "capskip_error"),
}

@app.exception_handler(CapSkipError)
async def capskip_error(request: Request, exc: CapSkipError):
    # Log the detail here, keep it away from the caller.
    logging.getLogger("uvicorn.error").warning("%r", exc)
    status, code = STATUS.get(type(exc), (502, "capskip_error"))
    if "UNSOLVABLE" in str(exc):
        status, code = 422, "unsolvable"
    return JSONResponse({"error": code}, status)

Each row below was provoked against the mock.

The 502 is deliberate. A rejected CapSkip key is the service’s mistake, not the caller’s. The 5.2 second row is httpx’s default timeout of 5 seconds, which the async client inherits. A silent answer to a poll is different. The SDK treats a poll that fails as not ready yet and tries again, so only the deadline ends it. A ValidationException cannot happen with just a sitekey and a URL. It needs extra options such as an action on a v2 solve, which this service never sends.

Step 9: catch the deadline separately

asyncio.timeout raises Python’s built-in TimeoutError. The SDK’s TimeoutException is a different class, and catching the built-in does not catch it. The handler in Step 8 already maps the SDK’s TimeoutException to a 504 named capskip_timeout. This second handler covers the deadline, and its error name tells the caller which clock ran out.

@app.exception_handler(TimeoutError)
async def deadline_hit(request: Request, exc: TimeoutError):
    # asyncio.timeout raises the built-in TimeoutError. The SDK's
    # TimeoutException is a different class, so it needs its own entry.
    return JSONResponse({"error": "deadline_exceeded"}, 504)

Step 10: add a health check

CapSkip has no ping call, so the check opens a TCP connection to its port and gives up after three seconds. With CapSkip closed it answered 503 after about 2 seconds, and with the mock running it answered 200 at once. It tells a monitoring script that the service is up and CapSkip is not. It does not prove that a solve works.

@app.get("/health")
async def health():
    # CapSkip has no ping call, so check that its port accepts a
    # connection.
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(HOST, PORT), timeout=3)
    except (OSError, TimeoutError):
        return JSONResponse({"ok": False}, 503)
    writer.close()
    return {"ok": True}

Run it on Windows

Start uvicorn from the folder that holds main.py. Bound to 127.0.0.1, the service answers only on that PC.

python -m uvicorn main:app --host 127.0.0.1 --port 8000

For callers on other machines, change the host to 0.0.0.0, which means every network adapter. On the test PC, that answered on the PC’s LAN address, and a 127.0.0.1 bind refused it. Set a real service key first, and check what firewall rule Windows creates for the port. The service key travels in clear text over plain HTTP, so use this only on a network you trust. The firewall prompt was not observed here, and access from another machine was not tested.

Run one worker. With the workers option set to 2 and a limit of 3 each, the mock saw up to 6 solves at once. Each worker is a separate process with its own Semaphore, so the limit multiplies. The reload option is safe to use while you edit. Touching the file mid-solve did not drop the running request, and a new request 1.5 seconds later also succeeded.

The documentation pages are open by default. When the service listens on a network address, create the app with docs_url and openapi_url set to None, so the schema is not public.

Run the service away from CapSkip

The service does not have to sit on the same PC. Put CapSkip in Server mode on a Windows machine you control, preferably with a static public IP. Set CAPSKIP_HOST and CAPSKIP_PORT to that address before starting uvicorn. Three cautions apply. Turn on API key validation in CapSkip and use the same key in CAPSKIP_API_KEY. The SDK talks plain HTTP, so that key crosses the network in clear text. CapSkip also adds an allow-any rule for its port, so add a block rule for every address that should not reach it. The connection settings section of the setup guide lists both modes.

Call it from PowerShell

The caller sends the sitekey, the page URL and the service key. The answer is a JSON object with one field, token. Use the token within about two minutes, and use it once. The token expiry guide explains the clock. The address below is a placeholder for your service.

Invoke-RestMethod -Method Post -Uri http://192.168.1.50:8000/solve `
  -Headers @{ "X-API-Key" = "YOUR_API_KEY" } `
  -ContentType "application/json" `
  -Body '{"sitekey":"YOUR_SITEKEY","url":"https://example.com/page-with-recaptcha"}'

A POST to /solve/ with a trailing slash gets a 307 redirect. Invoke-RestMethod on Windows PowerShell 5.1 followed it and kept the body and the header. If you prefer curl.exe, feed the JSON through stdin. The usual single-quoted JSON argument fails on PowerShell 5.1, because the inner double quotes are stripped, and the service answered with a JSON error.

# curl.exe reads the JSON from stdin, so no quote escaping is needed.
$body = '{"sitekey":"YOUR_SITEKEY","url":"https://example.com/page-with-recaptcha"}'
$body | curl.exe -s http://192.168.1.50:8000/solve `
  -H "X-API-Key: YOUR_API_KEY" `
  -H "Content-Type: application/json" `
  --data-binary "@-"

What the gate does and does not do

The mock’s thread limit was 3, and ten requests arrived at once, each needing a 2 second solve. The gate kept the mock at three solves at once, and all ten came back 200. Without a working gate, all ten jobs reached CapSkip and queued there. In the mock, four of them gave up after waiting past its three second wait limit and came back as unsolvable. That reply is the mock’s own model, and it was not confirmed on a real CapSkip.

The wait for a slot sits inside the deadline for a reason. With a 5 second deadline, a 1 second solve and ten requests at once, six came back 200 and four came back 504. One request never reached CapSkip, and three of the nine that did were abandoned mid-solve.

A slot is held a little longer than the solve itself. The SDK polls at once, then waits 0.25, 0.5, 1, 2 and 4 seconds, and then 5 seconds between polls. That puts the polls at 0, 0.25, 0.75, 1.75, 3.75, 7.75 and 12.75 seconds, and the slot stays taken until a poll reads the answer. A result that appears just after a poll waits for the next one. The poll at 7.75 s misses an 8 second solve by a quarter of a second. On the test PC the slow calls described in the next section pushed that poll past 8 s, so the answer was read at about 8.4 to 8.9 s. With that cost removed, the same solve waited for the poll at 12.75 s and returned after 12.9 s with the default pollingInterval of 5, and after 8.9 s with a pollingInterval of 1. If solves are long and the queue is long, a lower pollingInterval frees slots sooner.

Two limits remain. A caller that hangs up does not cancel its handler, so the orphaned solve keeps its slot until it ends. With one slot and a 4 second solve, the next request took 8.3 seconds in total, against 4.7 seconds with a free slot. The mock also keeps working on an abandoned job after the deadline passes, and the SDK has no cancel call. The slot count can therefore run ahead of CapSkip’s real threads for a few seconds. That is modelled in the mock. The real app was not tested.

One cost to know about

AsyncCapSkip builds a new httpx client for every request it sends, including every poll. On the test PC that cost about 141 ms of blocking CPU each time, mostly loading the certificate bundle, which a plain HTTP address never needs. Ten simultaneous 2 second solves finished in about 6 s, and a request to /openapi.json sent during the burst waited up to about 1.9 s. The same ten through a synchronous endpoint took 3.8 s with a worst case of a few milliseconds, but that route cannot end a hung call. A deadline wrapped around a thread leaves the thread stuck. After 41 hung requests the service held 41 stuck threads, and they stay until the process restarts.

The blocked loop also makes clocks slip under load. Against a silent CapSkip with a 2 second deadline, 12 simultaneous requests gave 8 answers at 2.6 to 2.9 s and 4 answers named capskip_timeout at 8.5 s. With 41 requests the slowest took 13.9 s. A single request met its deadline. The cause of the 8.5 s answers was not isolated, but the blocked loop is part of it, and the SDK alone showed the same slip. For a handful of callers the stall is harmless. It grows with the number of solves in flight, so keep the slot count at what CapSkip can really run.

The whole main.py

This is Steps 3 to 10 joined, with the imports gathered at the top. Save it as main.py.

import asyncio
import logging
import os
import secrets
from contextlib import asynccontextmanager

from capskip import ApiException, AsyncCapSkip, CapSkipError
from capskip import NetworkException, TimeoutException, ValidationException
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

# The SDK reads no env vars, so pass each setting in by hand.
HOST = os.getenv("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.getenv("CAPSKIP_PORT", "8080"))
KEY = os.getenv("CAPSKIP_API_KEY", "capskip")
MAX_THREADS = int(os.getenv("CAPSKIP_MAX_THREADS", "10"))
DEADLINE = float(os.getenv("SOLVE_DEADLINE", "120"))


@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client and one gate, shared by every request.
    app.state.cap = AsyncCapSkip(
        apiKey=KEY, host=HOST, port=PORT,
        recaptchaTimeout=DEADLINE,
    )
    app.state.slots = asyncio.Semaphore(MAX_THREADS)
    yield


app = FastAPI(lifespan=lifespan)

SERVICE_KEY = os.environ["SOLVER_API_KEY"]
if not SERVICE_KEY:
    raise SystemExit("SOLVER_API_KEY must not be empty")


def require_key(x_api_key: str = Header(default="")):
    good = secrets.compare_digest(
        x_api_key.encode(), SERVICE_KEY.encode())
    if not good:
        raise HTTPException(
            401, "missing or wrong X-API-Key",
            headers={"WWW-Authenticate": "APIKey"})


class SolveRequest(BaseModel):
    sitekey: str = Field(min_length=1)
    url: str = Field(pattern=r"^https?://")


async def solve_v2(state, sitekey: str, url: str) -> str:
    # The deadline also covers the wait for a free slot.
    async with asyncio.timeout(DEADLINE):
        async with state.slots:
            result = await state.cap.recaptcha(sitekey, url)
    return result["code"]


@app.post("/solve", dependencies=[Depends(require_key)])
async def solve(body: SolveRequest, request: Request):
    token = await solve_v2(request.app.state, body.sitekey, body.url)
    return {"token": token}


STATUS = {
    NetworkException: (503, "capskip_unreachable"),
    TimeoutException: (504, "capskip_timeout"),
    ValidationException: (422, "bad_parameters"),
    ApiException: (502, "capskip_error"),
}


@app.exception_handler(CapSkipError)
async def capskip_error(request: Request, exc: CapSkipError):
    # Log the detail here, keep it away from the caller.
    logging.getLogger("uvicorn.error").warning("%r", exc)
    status, code = STATUS.get(type(exc), (502, "capskip_error"))
    if "UNSOLVABLE" in str(exc):
        status, code = 422, "unsolvable"
    return JSONResponse({"error": code}, status)


@app.exception_handler(TimeoutError)
async def deadline_hit(request: Request, exc: TimeoutError):
    # asyncio.timeout raises the built-in TimeoutError. The SDK's
    # TimeoutException is a different class, so it needs its own entry.
    return JSONResponse({"error": "deadline_exceeded"}, 504)


@app.get("/health")
async def health():
    # CapSkip has no ping call, so check that its port accepts a
    # connection.
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(HOST, PORT), timeout=3)
    except (OSError, TimeoutError):
        return JSONResponse({"ok": False}, 503)
    writer.close()
    return {"ok": True}

Common errors in a FastAPI CAPTCHA solver

FastAPI’s own lifespan documentation covers the startup hook, and every CapSkip error code is listed in the API docs.

FAQ

The short version

To run a FastAPI CAPTCHA solver service, keep one AsyncCapSkip, one Semaphore and one asyncio.timeout around the wait and the solve. Check the service key first. Catch the SDK’s TimeoutException separately from the built-in TimeoutError. Run a single worker. Put a block rule in front of CapSkip if it runs in Server mode.

• Every CAPTCHA type the Python package handles: the Python CAPTCHA solver page.
• How the checkbox solver works: the reCAPTCHA v2 solver page.
• Many solves in one script: solving CAPTCHAs in parallel in Python.

A shared service can serve a whole team without anyone counting solves. CapSkip handles captcha bypass on your own Windows machine, so every solve behind the service stays free of a per-call bill.