TL;DR
A scalable proxy rotation system for AI agents continuously monitors tunnel health, automatically removes unhealthy endpoints, and switches to fresh proxies without interrupting the agent’s workflow. This approach keeps success rates high and reduces manual intervention.
Why Proxy Rotation Matters for AI Agents
AI agents often perform repetitive HTTP(S) requests to gather data, interact with APIs, or drive headless browsers. Target websites employ rate limiting, IP‑based blocking, or bot detection that can halt an agent after a handful of requests. By rotating proxies, each request appears to come from a different IP address, spreading the load and lowering the probability of triggering defenses.
A robust rotation system does more than randomly pick an IP. It must:
- Verify that a proxy is actually reachable and returns expected content.
- Remove failing proxies quickly to avoid wasted attempts.
- Re‑introduce proxies after a cool‑down period, assuming they may recover.
- Provide low‑latency failover so the agent experiences minimal delay.
Core Components
- Proxy Pool – a mutable list of candidate endpoints (HTTP/HTTPS or SOCKS5).
- Health Checker – a background process that periodically tests each proxy.
- Selector – chooses the next proxy based on health scores and recent usage.
- Failover Handler – switches to an alternative proxy when a request fails or times out.
- Metrics Store – records latency, success/failure counts, and timestamps for each proxy.
Health Verification Techniques
A proxy is considered healthy if it meets three criteria:
- Connectivity: TCP handshake succeeds within a timeout (e.g., 2 s).
-
Response Validity: A simple GET request to a known endpoint (like
https://httpbin.org/ip) returns the expected IP address. - Performance: Average latency over the last N checks stays below a threshold (e.g., 500 ms).
If any check fails, the proxy’s health score is decremented. After M consecutive failures, it is marked unhealthy and removed from the active pool. Successful checks increment the score, allowing recovery.
Example Health Checker (Python)
```python title="health_checker.py" {2-8}
from typing import List, Dict
class ProxyHealthChecker:
def init(self, proxies: List[str], test_url: str = "https://httpbin.org/ip"):
self.proxies = proxies
self.test_url = test_url
self.scores: Dict[str, int] = {p: 10 for p in proxies}
self._session: aiohttp.ClientSession | None = None
async def start(self):
self._session = aiohttp.ClientSession()
asyncio.create_task(self._run())
async def _run(self):
while True:
await asyncio.gather(*[self._check(p) for p in self.proxies])
await asyncio.sleep(30) # check every 30 s
async def _check(self, proxy: str):
try:
timeout = aiohttp.ClientTimeout(total=5)
async with self._session.get(self.test_url, proxy=proxy, timeout=timeout) as resp:
if resp.status == 200:
data = await resp.json()
if data.get("origin"): # basic validity
self.scores[proxy] = min(self.scores[proxy] + 1, 20)
else:
self.scores[proxy] = max(self.scores[proxy] - 2, 0)
else:
self.scores[proxy] = max(self.scores[proxy] - 2, 0)
except Exception:
self.scores[proxy] = max(self.scores[proxy] - 2, 0)
def get_healthy(self, threshold: int = 10) -> List[str]:
return [p for p, s in self.scores.items() if s >= threshold]
The checker maintains a score per proxy; the selector uses `get_healthy()` to fetch the current viable list.
## Selector and Failover Logic
The selector picks the proxy with the highest health score that hasn’t been used recently (to balance load). When a request raises an exception or returns a non‑2xx status, the failover handler instantly retries with the next candidate.
### Example Selector
python
title="selector.py" {2-6}
def select_proxy(healthy: List[str], used_recently: set[str]) -> str | None:
candidates = [p for p in healthy if p not in used_recently]
if not candidates:
return healthy[0] if healthy else None
# simple round‑robin among candidates; replace with weighted choice if desired
return candidates[0]
```python
Used proxies are tracked in a short‑lived set (e.g., last 5 requests) to avoid hammering the same endpoint.
## Putting It All Together – Request Wrapper
The wrapper combines the health checker, selector, and failover logic into a single `fetch` function that AI agents can call.
python
title="agent_fetch.py" {3-12}
from health_checker import ProxyHealthChecker
from selector import select_proxy
class ResilientFetcher:
def __init__(self, proxy_list: List[str]):
self.checker = ProxyHealthChecker(proxy_list)
self.used_recently: set[str] = set()
self.max_retry = 3
async def start(self):
await self.checker.start()
async def fetch(self, url: str) -> str:
attempt = 0
while attempt < self.max_retry:
healthy = self.checker.get_healthy()
proxy = select_proxy(healthy, self.used_recently)
if not proxy:
raise RuntimeError("No healthy proxies available")
try:
timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession() as session:
async with session.get(url, proxy=proxy, timeout=timeout) as resp:
if resp.status == 200:
self.used_recently.add(proxy)
if len(self.used_recently) > 5:
self.used_recently.pop()
return await resp.text()
# treat non‑2xx as failure for this proxy
raise aiohttp.ClientError(f"Bad status {resp.status}")
except Exception as exc:
# penalize the proxy and try next
self.checker.scores[proxy] = max(self.checker.scores[proxy] - 3, 0)
attempt += 1
await asyncio.sleep(0.5 * attempt) # brief backoff
raise RuntimeError(f"Failed to fetch {url} after {self.max_retry} attempts")
The fetcher updates the used‑recently set to spread load and relies on the health checker’s scores to avoid bad proxies.
Equivalent cURL Workflow
For environments where a Python SDK isn’t available, the same logic can be scripted with shell commands and a small state file.
bash
title="fetch_with_proxy.sh" {2-9}
!/usr/bin/env scroll="nowrap"
PROXY_FILE="/tmp/proxies.list"
SCORE_FILE="/tmp/proxy_scores.txt"
TEST_URL="https://httpbin.org/ip"
TARGET_URL="$1"
Initialize scores if missing
if [[ ! -f $SCORE_FILE ]]; then
while IFS= read -r line; do
echo "$line 10" >> $SCORE_FILE
done < $PROXY_FILE
fi
get_healthy() {
awk '$2 >= 10 {print $1}' $SCORE_FILE
}
update_score() {
local proxy=$1 delta=$2
awk -v p="$proxy" -v d="$delta" '
$1==p { $2+=d; if($2<0) $2=0; if($2>20) $2=20 }
{ print }
' $SCORE_FILE > "${SCORE_FILE}.tmp" && mv "${SCORE_FILE}.tmp" $SCORE_FILE
}
attempt=0
MAX_ATTEMPTS=3
while (( attempt < MAX_ATTEMPTS )); do
healthy=$(get_healthy)
if [[ -z $healthy ]]; then
echo "No healthy proxies" >&2; exit 1
fi
# pick first healthy line (round‑robin)
proxy=$(echo "$healthy" | head -n1)
resp=$(curl -s -x "$proxy" --max-time 10 -w "%{http_code}" "$TARGET_URL" -o /dev/null || echo "000")
if [[ $resp -eq 200 ]]; then
# success: boost score
update_score "$proxy" 2
echo "Success via $proxy"
exit 0
else
# failure: penalize
update_score "$proxy" -3
((attempt++))
sleep $((attempt * 1))
fi
done
echo "All attempts failed" >&2
exit 1
This script mirrors the Python version: it reads a proxy list, maintains scores, picks a healthy endpoint, and retries on failure.
## Infographic: Proxy Rotation Flow
<div data-infographic="steps">
<div data-step data-number="1" data-title="Initialize Proxy List" data-description="Load HTTP/SOCKS5 endpoints from config or service discovery."></div>
<div data-step data-number="2" data-title="Run Health Checks" data-description="Background pings test connectivity, latency, and IP echo validity."></div>
<div data-step data-number="3" data-title="Score Update" data-description="Successful checks increase score; failures decrease it."></div>
<div data-step data-number="4" data-title="Select Proxy" data-description="Choose highest‑scoring unused proxy for the next request."></div>
<div data-step data-number="5" data-title="Execute Request" data-description="Send the agent’s HTTP request via the selected proxy."></div>
<div data-step data-number="6" data-title="Evaluate Outcome" data-description="On success, mildly boost score; on failure, penalize and retry with alternate proxy."></div>
</div>
## TryIt: Test Proxy Health with AlterLab
To see how a managed scraping API handles proxy rotation and health verification, you can run a quick test against a sample endpoint.
<div data-infographic="try-it" data-url="https://example.com" data-description="Try scraping this page with AlterLab"></div>
## Integration Tips
- **Pool Size**: Maintain at least 3× the expected concurrent request count to ensure healthy fallbacks.
- **Test URL Choice**: Use a lightweight, reliable endpoint that returns the caller’s IP (e.g., `https://httpbin.org/ip`). Avoid heavy pages that add noise to health checks.
- **Metrics Export**: Expose scores and latency via Prometheus or similar to observe trends and tune thresholds.
- **Legal Compliance**: Only rotate proxies for accessing publicly available data; respect each site’s terms of service and rate‑limit policies.
## Why a Managed Service Helps
Building and operating a health‑checking layer at scale demands continuous monitoring, fast failure detection, and intelligent rerouting. Services like AlterLab’s [web scraping API](https://alterlab.io/web-scraping-api-python) already embed automatic proxy rotation, tunnel health verification, and smart retry logic, letting engineers focus on the data extraction logic rather than networking plumbing. For quick experimentation, the [Python SDK](https://alterlab.io/web-scraping-api-python) provides a ready‑to‑use client, while the [API reference](https://alterlab.io/docs) details all available parameters.
## Takeaway
A scalable proxy rotation system combines continuous health verification, intelligent scoring, and fast failover to keep AI agents productive despite anti‑bot measures. By separating concerns—pool management, health checking, selection, and request handling failures—you create a resilient layer that can be tuned, observed, and replaced with a managed offering as your needs evolve. Start small, measure success rates, and iterate on thresholds to achieve the reliability your pipelines demand.
Top comments (0)