Crawl4AI proxy configuration and rotation

import asyncio, os
from crawl4ai import AsyncWebCrawler, BrowserConfig, CacheMode, CrawlerRunConfig, ProxyConfig

proxy = ProxyConfig(
    server=f"http://{os.environ['PROXY_HOST']}:8080",
    username=os.environ["PROXY_USER"],        # rotates per request: no session on the username
    password=os.environ["PROXY_PASS"],
)
run = CrawlerRunConfig(proxy_config=proxy, cache_mode=CacheMode.BYPASS, page_timeout=30_000)

async def main():
    async with AsyncWebCrawler(config=BrowserConfig(headless=True)) as crawler:
        probe = await crawler.arun(url="https://api.ipify.org?format=json", config=run)
        print("exit:", probe.html)                 # confirm the residential exit before real pages
        result = await crawler.arun(url="https://example.com/", config=run)
        print(result.success, result.status_code, len(result.markdown))

asyncio.run(main())

import asyncio, os
from crawl4ai import AsyncWebCrawler, BrowserConfig, CacheMode, CrawlerRunConfig, ProxyConfig
from crawl4ai.proxy_strategy import RoundRobinProxyStrategy

HOST, USER, PASS = os.environ["PROXY_HOST"], os.environ["PROXY_USER"], os.environ["PROXY_PASS"]

def session(session_id: str, lifetime: int = 600, country: str | None = None) -> ProxyConfig:
    suffix = (f"-country-{country}" if country else "") + f"-session-{session_id}-lifetime-{lifetime}"
    return ProxyConfig(server=f"http://{HOST}:8080", username=f"{USER}{suffix}", password=PASS)

# one exit per crawl: every page of this crawl leaves through the same IP for 10 minutes
one_crawl = CrawlerRunConfig(proxy_config=session("crawl01", country="us"), cache_mode=CacheMode.BYPASS)

# several independent crawls spread over four exits, round robin
pool = RoundRobinProxyStrategy([session(f"crawl{i:02d}") for i in range(1, 5)])
spread = CrawlerRunConfig(proxy_rotation_strategy=pool, cache_mode=CacheMode.BYPASS)

async def main():
    async with AsyncWebCrawler(config=BrowserConfig(headless=True)) as crawler:
        pages = await crawler.arun_many(urls=["https://example.com/a", "https://example.com/b"], config=one_crawl)
        print([(p.url, p.status_code) for p in pages])

asyncio.run(main())