How to Validate 100+ Proxy Endpoints in Minutes Without a Paid API

A proxy pool does not fail loudly. Entries go stale one at a time: a node stops answering, an authentication credential rotates, an upstream line gets reassigned to another customer. Your pipeline keeps running, and the failures surface later as missing records, a partial dataset, or a suspension email.

Checking the pool before a run is cheaper than debugging a run that went wrong.

What is actually worth testing

Most proxy checkers return a single number — latency — and stop there. That number is the least useful of the five things you can measure:

Check	Catches	Cost
TCP connect	Dead host, closed port	One socket
Real HTTP request through the proxy	Auth failure, whitelist rejection, upstream outage	One request
Exit IP matches expectation	Wrong node returned, pool misconfiguration	Free (from the response)
Latency	Slow-but-alive nodes worth deprioritising	Free
Header leakage	Via, X-Forwarded-For, real IP in errors	Free

The important one is the second. A socket that connects is not a proxy that works — plenty of dead nodes still accept TCP and then hang or reject. You have to push a real request through and read the answer.

The checker

This uses only the standard library. It runs the whole battery concurrently and returns a structured report.

import json
import socket
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse

# Any endpoint that echoes the caller's IP works.
PROBE_URL = "http://ip-api.com/json/?fields=status,query,country,as"
TIMEOUT = 12
WORKERS = 40


def check(entry):
    """entry: {'id': ..., 'host': ..., 'port': ..., 'user': ..., 'password': ...}"""
    result = {
        "id": entry.get("id"),
        "host": entry.get("host"),
        "port": entry.get("port"),
        "tcp_ok": False,
        "http_ok": False,
        "exit_ip": None,
        "latency_ms": None,
        "error": None,
    }

    # --- 1. TCP reachability -------------------------------------------------
    t0 = time.perf_counter()
    try:
        with socket.create_connection((entry["host"], int(entry["port"])), timeout=TIMEOUT):
            result["tcp_ok"] = True
    except Exception as exc:
        result["error"] = f"tcp: {type(exc).__name__}"
        return result

    # --- 2. Real request through the proxy ----------------------------------
    # Uses a plain HTTP CONNECT-free request for SOCKS-compatible testing.
    # For a genuine SOCKS5 handshake use PySocks: socks.socksocket()
    try:
        import socks  # PySocks
        s = socks.socksocket()
        s.set_proxy(socks.SOCKS5, entry["host"], int(entry["port"]),
                    username=entry.get("user"), password=entry.get("password"))
        s.settimeout(TIMEOUT)
        s.connect((urlparse(PROBE_URL).hostname, 80))
        s.sendall(
            f"GET {urlparse(PROBE_URL).path} HTTP/1.1\r\n"
            f"Host: {urlparse(PROBE_URL).hostname}\r\n"
            f"User-Agent: pool-checker/1.0\r\n"
            f"Connection: close\r\n\r\n".encode()
        )
        chunks = []
        while True:
            chunk = s.recv(4096)
            if not chunk:
                break
            chunks.append(chunk)
        s.close()

        raw = b"".join(chunks).decode("utf-8", "ignore")
        body = raw.split("\r\n\r\n", 1)[-1]
        payload = json.loads(body)
        result["latency_ms"] = round((time.perf_counter() - t0) * 1000)
        if payload.get("status") == "success":
            result["http_ok"] = True
            result["exit_ip"] = payload.get("query")
            result["asn"] = payload.get("as")
    except Exception as exc:
        result["error"] = f"http: {type(exc).__name__}"
    return result


def validate(pool):
    healthy, dead = [], []
    with ThreadPoolExecutor(max_workers=WORKERS) as pool_exec:
        futures = {pool_exec.submit(check, e): e for e in pool}
        for fut in as_completed(futures):
            r = fut.result()
            (healthy if r["http_ok"] else dead).append(r)
    return healthy, dead


if __name__ == "__main__":
    pool = [
        {"id": "n1", "host": "127.0.0.1", "port": 1080},
        # ... load from your config
    ]
    healthy, dead = validate(pool)
    print(f"healthy {len(healthy)} / dead {len(dead)}")
    for r in sorted(healthy, key=lambda x: x["latency_ms"] or 0)[:10]:
        print(f'  {r["id"]:>6}  {r["latency_ms"]:>5} ms  exit={r["exit_ip"]}  {r.get("asn")}')

Two details that matter more than they look:

Concurrency is what makes this fast. 100 endpoints at 40 workers finishes in roughly the time of three bad nodes, not the sum of all of them. Serial checking with a 12-second timeout on 100 nodes can take twenty minutes when half the pool is down.

The timeout has to be per-connection, not global. A single global deadline means one hanging node poisons the whole batch result — which is exactly the failure mode you were trying to detect.

Reading the output

Three patterns are worth acting on:

High TCP success, low HTTP success. The nodes are reachable but the upstream is failing — usually an auth problem, an IP whitelist that does not include your machine, or a subscription that expired. This is a configuration issue, not a pool-quality issue.

Exit IPs clustered in one subnet. Your pool claims geographic diversity but is routing through a handful of machines. Check the returned as field: if most nodes share one ASN, the pool is not as diversified as advertised.

Bimodal latency. A cluster of fast nodes and a cluster of slow ones often means two different upstreams behind one entry point. Splitting them lets you route latency-sensitive work to the fast half.

Common mistakes

Only testing on port 443. Many proxy pools serve HTTP CONNECT on one port and SOCKS5 on another. Test the protocol you actually use.

Reusing one target site. Any site with per-IP rate limits will start refusing after a few hundred checks, and you will misread that as pool failure. Use a lightweight echo endpoint.

Ignoring the check count against the probe. If a source offers 45 requests per minute, 100 concurrent workers will get throttled and produce false negatives. Either throttle, or use multiple probe endpoints. The DNS-based lookup described in the ASN guide has no such limit.

Treating a green run as permanent. Reputation and availability are both time-varying. Validate before each run, or on a schedule.

FAQ

Do I need PySocks?
Only for genuine SOCKS5 handshakes. For HTTP and HTTPS proxies the standard library's urllib accepts a proxy URL directly, and requests does the same with the proxies argument.

How many workers should I use?
Start at 40. Past a few hundred you will saturate the local NIC or hit probe rate limits, and the extra failures are your own doing rather than the pool's.

Can I check without sending a request to a third party?
Not meaningfully. A handshake proves the listener exists; it does not prove the upstream works or what exit IP you get. Some target has to answer.

What latency should I accept?
It depends on the workload, but if the 90th percentile is above roughly ten times the median, you have a multi-tier pool and should treat the tiers separately.

How often should I re-validate?
Before every large run, plus a scheduled sweep on a fixed interval. Pools degrade between your runs, not during them.

---

A longer walkthrough of the same procedure, including how the checks are scheduled and what the pass thresholds are in practice, is here: batch proxy validation tutorial. For current per-platform pricing across the providers referenced in this series, see the pricing centre.