import asyncio
import aiohttp
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]

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}
import aiohttp
import asyncio
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")

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

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.