HTTPX + CaptchaAI Integration

HTTPX is a modern Python HTTP client with async support and HTTP/2. This guide shows how to use it with CaptchaAI for both sync and async CAPTCHA solving.

Requirements

Requirement	Details
Python	3.8+
httpx	0.24+
CaptchaAI API key	Get one here

pip install httpx

Synchronous Client

import httpx
import time
import os


class CaptchaAISync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.Client(timeout=30)

    def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = self.client.get(f"{self.base_url}/in.php", params=params)
        text = resp.text

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        task_id = text.split("|")[1]

        # Poll
        deadline = time.time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while time.time() < deadline:
            time.sleep(5)
            result = self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    def get_balance(self):
        resp = self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    def close(self):
        self.client.close()


# Usage
solver = CaptchaAISync(os.environ["CAPTCHAAI_API_KEY"])

token = solver.solve({
    "method": "userrecaptcha",
    "googlekey": "6Le-wvkS...",
    "pageurl": "https://example.com",
})
print(f"Token: {token[:50]}...")
solver.close()

Async Client

import httpx
import asyncio
import os


class CaptchaAIAsync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.AsyncClient(timeout=30)

    async def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = await self.client.get(
            f"{self.base_url}/in.php", params=params
        )
        text = resp.text

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        task_id = text.split("|")[1]

        # Poll
        deadline = asyncio.get_event_loop().time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)
            result = await self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    async def get_balance(self):
        resp = await self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    async def close(self):
        await self.client.aclose()


# Usage
async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])

    # Solve multiple concurrently
    tasks = [
        solver.solve({
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": f"https://example.com/page{i}",
        })
        for i in range(5)
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Page {i}: FAILED - {r}")
        else:
            print(f"Page {i}: solved ({len(r)} chars)")

    await solver.close()

asyncio.run(main())

HTTP/2 Support

HTTPX supports HTTP/2, reducing connection overhead:

pip install httpx[http2]

client = httpx.AsyncClient(http2=True, timeout=30)

HTTP/2 multiplexes requests over a single connection, improving performance when submitting and polling multiple CAPTCHAs.

Scraping Example with CAPTCHA Handling

import httpx
import re
import os

async def scrape_with_captcha(url, solver):
    async with httpx.AsyncClient() as client:
        # Fetch page
        resp = await client.get(url)
        html = resp.text

        # Check for reCAPTCHA
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        if not match:
            return html

        site_key = match.group(1)
        token = await solver.solve({
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit form with token
        resp = await client.post(url, data={
            "g-recaptcha-response": token,
        })
        return resp.text


async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
    content = await scrape_with_captcha("https://example.com", solver)
    print(f"Got {len(content)} chars")
    await solver.close()

asyncio.run(main())

Comparison: httpx vs requests vs aiohttp

Feature	httpx (sync)	httpx (async)	requests	aiohttp
Async support	❌	✅	❌	✅
HTTP/2	✅	✅	❌	❌
Connection pooling	✅	✅	✅	✅
API compatibility	requests-like	requests-like	—	Different
Best for	Drop-in replacement	Modern async code	Quick scripts	High concurrency

FAQ

Should I use httpx over requests?

For new projects, yes. httpx has a requests-compatible API plus async and HTTP/2 support. For existing code using requests, the migration is straightforward.