How to use a proxy with aiohttp

Both scripts below were run through local HTTP and SOCKS5 test proxies, with and without authentication, and printed the accepted target and the observed address.

Audience: Python developers who can run an asynchronous script. The commands use Bash or Zsh on macOS or Linux.

Warning: Proxy requests can consume paid traffic, and credentials can grant account access. Start with a sandbox allocation and an authorized test target. Keep credentials out of commands, source, and logs; stop the worker and rotate credentials if they appear in output.

Choose the route type

A local test showed what the last row looks like: session.get(url, proxy="socks5://...") ended with ServerDisconnectedError, not a clear configuration error. That error usually means the scheme, not the network, is wrong.

With an http:// proxy, Basic proxy credentials cross the client-to-proxy hop without encryption. HTTPS on the target does not protect that hop, so use a trusted network between the client and the proxy.

Install aiohttp

Tested with Python 3.12.0 from the official downloads page and aiohttp 3.14.3:

python3 -m venv .venv
.venv/bin/python -m pip install 'aiohttp==3.14.3'

Run the HTTP proxy check

Set a credential-free HTTP proxy URL and, when required, the proxy username. Replace <PROXY_HOST>, <PROXY_PORT>, and <PROXY_USERNAME> with values from the proxy operator:

export PROXY_URL='http://<PROXY_HOST>:<PROXY_PORT>'
export PROXY_USER='<PROXY_USERNAME>'

Load PROXY_PASSWORD from an approved secret manager before starting the worker. For an endpoint without authentication, leave both PROXY_USER and PROXY_PASSWORD unset. Save this program as aiohttp-proxy-check.py:

import asyncio
import ipaddress
import json
import math
import os
from urllib.parse import urlsplit

import aiohttp
from aiohttp import encode_basic_auth


TARGET_URL = "https://example.com/"
EXPECTED_MARKER = "Example Domain"
IP_CHECK_URL = "https://api.ipify.org?format=json"
MAX_ATTEMPTS = 3
MAX_BACKOFF_SECONDS = 2.0
RETRY_STATUSES = {502, 503, 504}
RETRY_ERRORS = (aiohttp.ClientConnectorError, aiohttp.ServerDisconnectedError, asyncio.TimeoutError)


def proxy_settings() -> tuple[str, dict[str, str] | None]:
    proxy_url = os.environ["PROXY_URL"]
    endpoint = urlsplit(proxy_url)
    if (
        endpoint.scheme != "http"
        or not endpoint.hostname
        or not endpoint.port
        or endpoint.username is not None
        or endpoint.password is not None
        or endpoint.path not in {"", "/"}
        or endpoint.query
        or endpoint.fragment
    ):
        raise ValueError("PROXY_URL must be a credential-free http:// URL")
    username = os.environ.get("PROXY_USER")
    password = os.environ.get("PROXY_PASSWORD")
    if bool(username) != bool(password):
        raise ValueError("Set both proxy username and password, or neither")
    headers = None
    if username and password:
        headers = {"Proxy-Authorization": encode_basic_auth(username, password)}
    return proxy_url, headers


async def get_with_retries(
    session: aiohttp.ClientSession,
    url: str,
    proxy_url: str,
    proxy_headers: dict[str, str] | None,
) -> tuple[int, str]:
    for attempt in range(MAX_ATTEMPTS):
        try:
            async with session.get(
                url,
                proxy=proxy_url,
                proxy_headers=proxy_headers,
                allow_redirects=False,
            ) as response:
                body = await response.text()
                if response.status in RETRY_STATUSES and attempt + 1 < MAX_ATTEMPTS:
                    value = response.headers.get("Retry-After")
                    try:
                        server_delay = float(value) if value is not None else 0.0
                    except ValueError:
                        return response.status, body
                    if (
                        not math.isfinite(server_delay)
                        or server_delay < 0
                        or server_delay > MAX_BACKOFF_SECONDS
                    ):
                        return response.status, body
                    delay = min(max(server_delay, 0.5 * (2**attempt)), MAX_BACKOFF_SECONDS)
                    await asyncio.sleep(delay)
                    continue
                return response.status, body
        except RETRY_ERRORS:
            if attempt + 1 == MAX_ATTEMPTS:
                raise RuntimeError("request exhausted its bounded connection retries") from None
            await asyncio.sleep(min(0.5 * (2**attempt), MAX_BACKOFF_SECONDS))
    raise RuntimeError("request ended without a response")


async def main() -> None:
    proxy_url, proxy_headers = proxy_settings()
    timeout = aiohttp.ClientTimeout(total=15, connect=5, sock_read=10)
    connector = aiohttp.TCPConnector(limit=8, limit_per_host=4)
    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        trust_env=False,
    ) as session:
        status, body = await get_with_retries(session, TARGET_URL, proxy_url, proxy_headers)
        if status != 200 or EXPECTED_MARKER not in body:
            raise RuntimeError(f"target check failed with HTTP {status}")
        print(f"Target accepted: HTTP {status}; marker found")

        status, body = await get_with_retries(session, IP_CHECK_URL, proxy_url, proxy_headers)
        if status != 200:
            raise RuntimeError(f"IP check failed with HTTP {status}")
        ip_data = json.loads(body)
        ip_value = ip_data.get("ip") if isinstance(ip_data, dict) else None
        if not isinstance(ip_value, str):
            raise ValueError("IP endpoint must return a JSON object with a string ip")
        address = ipaddress.ip_address(ip_value)
        print(f"Observed exit address: {address}")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except (aiohttp.ClientError, asyncio.TimeoutError, KeyError, ValueError, RuntimeError):
        print("Check failed; inspect the redacted failure phase and proxy configuration")
        raise SystemExit(1) from None

Run it in the environment that holds the proxy values:

.venv/bin/python aiohttp-proxy-check.py

Expected output. The address is a documentation example, not a captured exit:

Target accepted: HTTP 200; marker found
Observed exit address: 203.0.113.10

The address shows what the IP service saw on this route. It does not establish location, anonymity, or acceptance by the target.

Run the SOCKS5 proxy check

Install the SOCKS connector. Tested with aiohttp-socks 0.12.0, which installed python-socks 3.1.1:

.venv/bin/python -m pip install 'aiohttp-socks==0.12.0'

Set the SOCKS5 host and port. Add PROXY_USER and PROXY_PASSWORD through the secret manager when the endpoint requires them:

export SOCKS_HOST='<PROXY_HOST>'
export SOCKS_PORT='<PROXY_PORT>'

Save this program as aiohttp-socks-check.py. rdns=True sends the target hostname to the proxy for resolution, the same behavior as a socks5h:// URL; the SOCKS5 vs SOCKS5h guide explains when that matters.

import asyncio
import ipaddress
import os

import aiohttp
from aiohttp_socks import (
    ProxyConnectionError,
    ProxyConnector,
    ProxyError,
    ProxyTimeoutError,
    ProxyType,
)


TARGET_URL = "https://example.com/"
EXPECTED_MARKER = "Example Domain"
IP_CHECK_URL = "https://api.ipify.org?format=json"


def socks_connector() -> ProxyConnector:
    # rdns=True sends the target hostname to the proxy (SOCKS5h behavior).
    return ProxyConnector(
        proxy_type=ProxyType.SOCKS5,
        host=os.environ["SOCKS_HOST"],
        port=int(os.environ["SOCKS_PORT"]),
        username=os.environ.get("PROXY_USER"),
        password=os.environ.get("PROXY_PASSWORD"),
        rdns=True,
    )


async def main() -> None:
    timeout = aiohttp.ClientTimeout(total=15, connect=5, sock_read=10)
    async with aiohttp.ClientSession(
        connector=socks_connector(), timeout=timeout, trust_env=False
    ) as session:
        async with session.get(TARGET_URL, allow_redirects=False) as response:
            body = await response.text()
            if response.status != 200 or EXPECTED_MARKER not in body:
                raise RuntimeError(f"target check failed with HTTP {response.status}")
            print(f"Target accepted: HTTP {response.status}; marker found")
        async with session.get(IP_CHECK_URL, allow_redirects=False) as response:
            if response.status != 200:
                raise RuntimeError(f"IP check failed with HTTP {response.status}")
            data = await response.json()
            address = ipaddress.ip_address(data["ip"])
            print(f"Observed exit address: {address}")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except (
        aiohttp.ClientError,
        ProxyError,
        ProxyConnectionError,
        ProxyTimeoutError,
        asyncio.TimeoutError,
        asyncio.IncompleteReadError,
        KeyError,
        ValueError,
        RuntimeError,
    ):
        print("Check failed; inspect the SOCKS settings and target")
        raise SystemExit(1) from None

.venv/bin/python aiohttp-socks-check.py

A successful run prints the same two lines as the HTTP check. A closed port or rejected password prints the short failure line and exits with status 1. The connector applies to the whole session, so create a separate session for each route or rotation lifecycle.

Understand retries and limits

The HTTP helper makes at most 3 attempts per URL for connector errors, server disconnects, timeouts, and HTTP 502, 503, or 504. It caps backoff at 2 seconds and honors numeric Retry-After values only within that cap. It never retries HTTP 407, authorization failures, redirects, or rate-limit responses.

The two checks can make up to 6 proxy requests. Use the helper only for idempotent GET requests. A session timeout applies to each request, not the whole job, so add an outer task deadline for a wall-clock limit. The connector caps the pool at 8 simultaneous connections and 4 per host.

Both programs buffer each body in memory without a byte cap. Use a small authorized test target; for larger responses, switch to streamed reads with an explicit size limit.

trust_env=False keeps environment proxy settings and .netrc from changing the route or credentials. In aiohttp 3.14, send HTTP proxy credentials through proxy_headers with encode_basic_auth(); the older proxy_auth parameter is deprecated.

Troubleshoot the failed stage

Keep TLS certificate verification enabled. For a trusted intercepting proxy, configure its approved CA certificate instead of disabling verification.

The aiohttp proxy reference documents route and authentication options, and the client reference covers timeouts and connectors. The aiohttp-socks 0.12.0 source documents the connector. For a synchronous client, compare the HTTPX proxy guide and the Python Requests proxy guide.

ProxyLaneTraffic never expires

Premium residential proxies
From $2/GB

Scale your workflows with the tools you already use. Rotating or sticky sessions up to 72 hours.

Try 350 MB for $1.95 View plans Proxies for web scraping