How to Use Proxies With aiohttp (async Python)

aiohttp routes through a proxy differently from every synchronous client, and the difference is the first thing that trips people coming from requests. There is no proxies dictionary and no session-level proxy setting that sticks. You pass the proxy on the request itself, aiohttp ignores your environment by default, and SOCKS needs a separate library. This guide covers proxies with aiohttp end to end: the proxy argument, authentication, trust_env, SOCKS5 through aiohttp-socks, rotating a pool across concurrent tasks without melting your file-descriptor limit, and the timeout and retry handling that keeps an async crawler alive.

We run a proxy network and a live proxy checker, so the failure we see most with aiohttp proxy code is a run that silently uses your real IP because the proxy was set the requests way. Every example below is runnable on aiohttp 3.9 or newer. Where one needs a live proxy, pull a fresh one from our free proxy API, which returns real, recently checked endpoints with no key.

How do you use a proxy with aiohttp?

Pass the proxy URL as the proxy argument to the request call, not to the session: session.get(url, proxy="http://host:port"). aiohttp has no proxies dict, so the proxy lives on each get or post. Always attach an explicit ClientTimeout so a dead proxy cannot stall a task forever.

import aiohttp
import asyncio

async def main():
    timeout = aiohttp.ClientTimeout(total=20, connect=5)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(
            "https://httpbin.org/ip",
            proxy="http://203.0.113.7:8080",
        ) as resp:
            print(await resp.json())

asyncio.run(main())

The proxy argument, and why the session will not hold it

Coming from requests, the instinct is to set the proxy once on the session and forget it. aiohttp does not work that way. There is no session.proxies, and assigning one does nothing. The proxy is a property of the request, so it goes on every call:

async with session.get(url, proxy="http://203.0.113.7:8080") as resp:
    ...

This looks like more typing than the requests proxies dict, but it buys you something: in a scraper that rotates exits, the proxy is supposed to change per request, and aiohttp puts it exactly where that decision belongs. The pattern that trips people is reaching for a session-wide setting that does not exist and then wondering why every request went out on the real IP.

Environment variables are off by default. requests reads HTTP_PROXY and HTTPS_PROXY unless you stop it. aiohttp does the opposite: it ignores them unless you opt in with trust_env=True on the session. That default is a feature, because a proxy hiding in your shell cannot silently reroute you, but it surprises anyone expecting the requests behaviour.

# Read HTTP_PROXY / HTTPS_PROXY / NO_PROXY from the environment
async with aiohttp.ClientSession(trust_env=True) as session:
    async with session.get("https://example.com") as resp:
        ...

Proxy authentication

Paid proxies need a username and password. The direct way is the userinfo part of the URL, user:pass@host:port:

proxy = "http://user:pass@203.0.113.7:8080"
async with session.get(url, proxy=proxy) as resp:
    ...

aiohttp also accepts credentials as a separate proxy_auth, which is cleaner when the password carries characters that are awkward inside a URL:

auth = aiohttp.BasicAuth("myuser", "p@ss:word/!")
async with session.get(url, proxy="http://203.0.113.7:8080", proxy_auth=auth) as resp:
    ...

Wrong or missing credentials come back as HTTP 407 Proxy Authentication Required. As with any client, a 407 is oddly good news: it proves the proxy answered and only the login was wrong.

SOCKS5 with aiohttp-socks

aiohttp speaks HTTP proxies only. For SOCKS4 or SOCKS5 you install one extra package:

pip install aiohttp_socks

Then route the whole session through a ProxyConnector instead of passing proxy=:

import aiohttp
from aiohttp_socks import ProxyConnector

async def fetch():
    connector = ProxyConnector.from_url("socks5://user:pass@203.0.113.7:1080")
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://httpbin.org/ip") as resp:
            return await resp.json()

There is a DNS subtlety here that mirrors the socks5 versus socks5h split in curl and requests. By default the connector can resolve hostnames locally, which leaks every target name to your own resolver. Pass rdns=True to make the proxy do the lookup instead, which is the private choice for scraping:

connector = ProxyConnector.from_url("socks5://203.0.113.7:1080", rdns=True)

One structural point matters for rotation: a ProxyConnector binds the session to one proxy. You do not swap it mid-session, so a pool of SOCKS proxies means one connector, and usually one session, per exit. That is the opposite of the HTTP case, where proxy= changes freely per request.

Rotating a pool across concurrent tasks

The reason to use aiohttp at all is concurrency, and the mistake that follows is firing ten thousand requests at once until the process runs out of sockets. Bound it with an asyncio.Semaphore, hand each task a proxy, and retry on a different exit when one fails:

import aiohttp
import asyncio
import random

POOL = [
    "203.0.113.7:8080",
    "203.0.113.24:3128",
    "198.51.100.14:8080",
    "198.51.100.66:8000",
]

async def fetch_one(session, url, sem, tries=4):
    async with sem:                      # cap how many run at once
        for _ in range(tries):
            proxy = f"http://{random.choice(POOL)}"
            try:
                async with session.get(url, proxy=proxy) as resp:
                    resp.raise_for_status()
                    return await resp.text()
            except aiohttp.ClientError:
                continue                 # dead or blocked exit, try the next
        return None                      # all tries failed for this url

async def crawl(urls, concurrency=20):
    sem = asyncio.Semaphore(concurrency)
    timeout = aiohttp.ClientTimeout(total=30, connect=5)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        tasks = [fetch_one(session, u, sem) for u in urls]
        return await asyncio.gather(*tasks)

urls = ["https://example.com/"] * 100
results = asyncio.run(crawl(urls))

Two things carry this. The semaphore caps concurrency so you open twenty sockets, not a hundred, no matter how long the URL list is. And the retry picks a different proxy each attempt, because on any pool some exits are always down and retrying the same dead one just wastes the attempt. You do not have to maintain the list by hand; our free proxy API returns a fresh pool you can load at startup:

async def load_pool(session):
    async with session.get(
        "https://hproxy.com/api/proxy-list",
        params={"format": "txt", "protocol": "http", "recent": "true", "limit": 50},
    ) as resp:
        return (await resp.text()).split()

Timeouts and error handling

An unattended async run lives or dies on its timeout. aiohttp's default is a five-minute total, which is far too patient for a proxy that accepted your connection and went silent. Set a ClientTimeout with the parts that matter:

timeout = aiohttp.ClientTimeout(
    total=30,          # whole request budget
    connect=5,         # time to get a connection from the pool or proxy
    sock_connect=5,    # time to open the socket to the proxy
    sock_read=20,      # time between reads once connected
)
async with aiohttp.ClientSession(timeout=timeout) as session:
    ...

The exceptions aiohttp raises for proxy trouble all descend from aiohttp.ClientError, so catch the specific ones you can act on and sweep the rest:

from aiohttp import ClientProxyConnectionError, ClientHttpProxyError
from asyncio import TimeoutError

try:
    async with session.get(url, proxy=proxy) as resp:
        resp.raise_for_status()
        body = await resp.text()
except ClientProxyConnectionError:
    ...   # could not reach the proxy at all: dead exit or wrong port
except ClientHttpProxyError:
    ...   # the proxy answered with an error, e.g. 407 auth or 502
except TimeoutError:
    ...   # proxy connected then stalled past the timeout
except aiohttp.ClientError:
    ...   # everything else: connection reset, TLS failure, and so on

ClientProxyConnectionError means the proxy never answered, ClientHttpProxyError means it answered with a proxy-level HTTP error, and a bare asyncio.TimeoutError means it went quiet after connecting. Catching ClientError last is the safety net that stops one odd exit from killing the whole gather.

Verify the exit IP

Never assume the proxy took effect. Confirm the target sees the proxy's address and not yours, and check the forwarded headers too, because a transparent proxy can leak your real IP even when the visible origin looks changed:

async def check(session, proxy):
    async with session.get("https://httpbin.org/ip") as r:
        real = (await r.json())["origin"]
    async with session.get("https://httpbin.org/ip", proxy=proxy) as r:
        via = (await r.json())["origin"]
    assert real != via, "proxy is not changing your IP"

    async with session.get("http://httpbin.org/headers", proxy=proxy) as r:
        fwd = (await r.json())["headers"].get("X-Forwarded-For")
    print("leaks real IP" if fwd else "no X-Forwarded-For leak")

If the two addresses match, the proxy is not routing you, which in aiohttp almost always means the proxy= argument was left off a call somewhere.

Where to go from here

aiohttp gives you concurrency for free and takes back the convenience of a session-wide proxy in return. Once the per-request proxy argument, trust_env, and the aiohttp-socks connector are muscle memory, the rest is the same discipline every proxied crawler needs: verify each exit before you trust it, bound your concurrency, and pull from a pool that is actually alive.

For verifying exits, our proxy checker runs exit IP, anonymity grade, geolocation and latency in one paste. For the pool, the free proxy API hands you recently checked endpoints with no key, which is ideal for testing this code before you wire in a paid gateway. The synchronous requests guide and the httpx guide are the sibling references if part of your stack is not async, proxies for web scraping covers picking the right proxy type, and how to avoid IP bans while scraping is the prevention checklist for when rotation alone is not enough. When a hand-rolled pool outgrows a list, a rotating gateway that returns a fresh residential IP per request at $0.44/GB removes the list management entirely.

Sources

• aiohttp: Client Usage, Proxy support: the per-request proxy argument, proxy_auth, and trust_env.
• aiohttp-socks on PyPI: the ProxyConnector, the socks5 scheme, and the rdns flag for proxy-side DNS.
• aiohttp: ClientTimeout: the total, connect, sock_connect and sock_read fields.