aiohttp Proxy Guide: Async Requests, Authentication & High-Concurrency Patterns (2026)

Developer guide · Last reviewed 14 October 2026 · 25 min read

Adding a proxy to aiohttp takes one argument. Building an asynchronous proxy client that stays predictable when one request becomes 100, 1,000 or 100,000 takes a lot more: connection pooling, concurrency limits, proxy authentication, retries, timeouts, session persistence, IP rotation, memory, backpressure, DNS behavior and observability. This aiohttp proxy guide covers both, from the first proxied request to a production worker pool, with Python examples throughout.

Quick Start

A basic proxied aiohttp request looks like this:

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://example.com",
            proxy="http://proxy.example.com:8080",
        ) as response:
            print(await response.text())

asyncio.run(main())

For proxy authentication, add proxy_auth:

import aiohttp
import asyncio

async def main():
    proxy_auth = aiohttp.BasicAuth(
        "proxy_username",
        "proxy_password",
    )

    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://example.com",
            proxy="http://proxy.example.com:8080",
            proxy_auth=proxy_auth,
        ) as response:
            print(response.status)

asyncio.run(main())

For concurrency, do not simply create thousands of tasks without controls. Use a reusable ClientSession, configure connection-pool limits, and apply an application-level semaphore:

connector = aiohttp.TCPConnector(
    limit=200,
    limit_per_host=50,
)

semaphore = asyncio.Semaphore(100)

For large public-web workloads, a managed proxy network such as ProxyEmpire can provide rotating or sticky exit IPs while aiohttp handles asynchronous request execution. The architecture should look like this:

Jobs
 |
 v
Async workers
 |
 +--> concurrency control
 +--> retry policy
 +--> timeout policy
 +--> session policy
 |
 v
Proxy gateway
 |
 v
Target

Not like this, which calls create_task() in a loop until the process runs out of sockets or memory:

create_task()
create_task()
create_task()
create_task()
...

Why Use aiohttp with Proxies?

Python has several mature HTTP clients. aiohttp is attractive when the workload is dominated by network I/O and needs to handle many independent requests concurrently. If you are choosing between libraries, our guide to Python web scraping with residential proxies covers requests, httpx, Scrapy and Playwright.

A sequential workflow behaves roughly like this:

request 1 ---> wait ---> response
request 2 ---> wait ---> response
request 3 ---> wait ---> response

An asynchronous workflow can overlap those waits:

request 1 ----------->
request 2 ----------->
request 3 ----------->
request 4 ----------->
           <-------- responses

That makes asynchronous networking useful for:

• web data collection;
• price monitoring;
• search-result monitoring;
• localization QA;
• uptime checks;
• public API aggregation;
• ad verification;
• distributed research pipelines.

Adding proxies introduces another network layer between the aiohttp worker and the destination. At small scale, this is configuration. At high scale, it becomes infrastructure.

aiohttp vs requests for proxy workloads

Neither library is universally better. They serve different concurrency models.

For 10 requests in an administration script, requests may be simpler. For thousands of overlapping network operations, aiohttp can provide a much better execution model. But asynchronous code only improves efficiency if concurrency is deliberately bounded.

Installing aiohttp and Sending a First Proxied Request

Install aiohttp

pip install aiohttp

A minimal asynchronous request, without a proxy, is:

import asyncio
import aiohttp

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://example.com") as response:
            html = await response.text()
            print(response.status)
            print(html[:200])

asyncio.run(main())

The central object is ClientSession. It owns important state, including its connection pool and reusable connections. In production, creating a new ClientSession for every request is usually the wrong design.

Basic aiohttp proxy example

Pass the proxy endpoint through the proxy argument:

import asyncio
import aiohttp

PROXY = "http://proxy.example.com:8080"

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://example.com",
            proxy=PROXY,
        ) as response:
            print("Status:", response.status)
            print(await response.text())

asyncio.run(main())

The request path becomes:

Python application
      |
      v
aiohttp
      |
      v
HTTP proxy
      |
      v
https://example.com

The destination normally observes the proxy’s exit IP rather than the public IP of the application host.

HTTP proxy with an HTTPS destination

A common misconception is that an HTTPS destination requires an HTTPS proxy URL. It does not. This is perfectly normal:

async with session.get(
    "https://example.com",
    proxy="http://proxy.example.com:8080",
) as response:
    ...

The proxy can create a tunnel to the HTTPS destination using HTTP CONNECT:

aiohttp
 |
 | CONNECT example.com:443
 v
HTTP proxy
 |
 | TCP tunnel
 v
example.com
 |
 | TLS
 v
HTTPS response

The protocol used to connect to the proxy and the protocol used by the target are separate questions. That distinction becomes important when debugging TLS problems.

Proxy Authentication with aiohttp

Authenticated proxy gateways commonly use username and password credentials. aiohttp provides BasicAuth for this purpose:

import asyncio
import aiohttp

PROXY = "http://proxy.example.com:8080"

async def main():
    auth = aiohttp.BasicAuth(
        "proxy_username",
        "proxy_password",
    )

    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://example.com",
            proxy=PROXY,
            proxy_auth=auth,
        ) as response:
            print(response.status)
            print(await response.text())

asyncio.run(main())

Credentials in the proxy URL

You may also see proxy credentials embedded in the URL:

proxy = "http://username:[email protected]:8080"

async with session.get(
    "https://example.com",
    proxy=proxy,
) as response:
    ...

It works, but embedding credentials directly in application strings increases the risk of exposing them through logs, tracebacks, source control, debugging output, configuration dumps and APM platforms. Separate authentication objects or secrets-backed configuration are preferable.

Proxy authentication vs target authentication

Proxy credentials and destination credentials are two independent authentication layers. Suppose your destination API also requires Basic Authentication:

proxy_auth = aiohttp.BasicAuth(
    "proxy_user",
    "proxy_password",
)

target_auth = aiohttp.BasicAuth(
    "api_user",
    "api_password",
)

async with session.get(
    "https://api.example.com/private",
    proxy="http://proxy.example.com:8080",
    proxy_auth=proxy_auth,
    auth=target_auth,
) as response:
    ...

application
  |
  | proxy credentials
  v
proxy gateway
  |
  | API credentials
  v
destination

Do not merge the two credential systems in your secrets architecture. The status code usually tells you which layer failed:

That distinction can save significant debugging time. Our reference on proxy error codes covers the other common ones.

Using Proxy-Authorization headers directly

Proxy authentication can also be supplied through proxy-specific headers with proxy_headers:

import aiohttp

proxy_headers = {
    "Proxy-Authorization": aiohttp.BasicAuth(
        "proxy_username",
        "proxy_password",
    ).encode()
}

async with aiohttp.ClientSession() as session:
    async with session.get(
        "https://example.com",
        proxy="http://proxy.example.com:8080",
        proxy_headers=proxy_headers,
    ) as response:
        print(response.status)

For standard Basic proxy authentication, proxy_auth=BasicAuth(...) is easier to read. Proxy-specific headers become more useful when an enterprise proxy infrastructure requires additional metadata or custom controls.

Don’t Create a New ClientSession for Every Request

This is a common asynchronous anti-pattern:

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Called thousands of times, it creates and destroys thousands of sessions. A better design passes one shared session in:

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            fetch(session, "https://example.com/1"),
            fetch(session, "https://example.com/2"),
            fetch(session, "https://example.com/3"),
        )

One reusable session gives aiohttp access to connection pooling and keep-alive behavior. For high-volume applications, that difference is significant.

Controlling Concurrency

Basic concurrent requests through a proxy

Here is the simplest concurrent implementation:

import asyncio
import aiohttp

PROXY = "http://proxy.example.com:8080"

async def fetch(session, url):
    async with session.get(
        url,
        proxy=PROXY,
    ) as response:
        return response.status, await response.text()

async def main():
    urls = [
        "https://example.com",
        "https://example.org",
        "https://example.net",
    ]

    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            *(fetch(session, url) for url in urls)
        )

    for status, body in results:
        print(status, len(body))

asyncio.run(main())

This is acceptable for a small list. It becomes dangerous when urls contains hundreds of thousands of entries.

Why asyncio.gather() is not a concurrency policy

If len(urls) == 100_000, asyncio.gather(*(fetch(session, url) for url in urls)) can create an enormous number of tasks almost immediately. Every task consumes resources. Connections consume sockets. Responses consume buffers. Retries consume additional capacity. The proxy gateway itself can also have connection or request limits.

Limit concurrency with asyncio.Semaphore

A semaphore bounds how many coroutines enter the network-intensive section at once:

import asyncio
import aiohttp

PROXY = "http://proxy.example.com:8080"
CONCURRENCY = 100

semaphore = asyncio.Semaphore(CONCURRENCY)

async def fetch(session, url):
    async with semaphore:
        async with session.get(
            url,
            proxy=PROXY,
        ) as response:
            return {
                "url": url,
                "status": response.status,
                "body": await response.text(),
            }

Now asyncio.gather(*(fetch(session, url) for url in urls)) can still create many tasks, but no more than 100 enter the guarded request section at the same time. This is better. For extremely large job sets, though, a worker queue is usually more memory-efficient.

Configure aiohttp’s TCPConnector

aiohttp also has connection-level concurrency controls:

connector = aiohttp.TCPConnector(
    limit=200,
    limit_per_host=50,
)

async with aiohttp.ClientSession(
    connector=connector,
) as session:
    ...

These controls complement application-level concurrency limits rather than replacing them. Think of them as separate layers:

job concurrency
    |
    v
asyncio.Semaphore
    |
    v
connection concurrency
    |
    v
aiohttp.TCPConnector
    |
    v
proxy / target

A useful starting pattern might be:

CONCURRENCY = 100

semaphore = asyncio.Semaphore(CONCURRENCY)

connector = aiohttp.TCPConnector(
    limit=150,
    limit_per_host=50,
)

Those numbers are not universal recommendations. Tune them using real latency, success-rate, memory, proxy-capacity and target-behavior data.

Per-target concurrency limits

A global limit is sometimes not enough. Suppose global concurrency is 200 and your queue contains 180 requests for target A and 10 each for targets B and C. Target A could consume nearly the entire budget. You may want a limit per target as well:

One simple pattern uses per-domain semaphores:

from collections import defaultdict
from urllib.parse import urlparse
import asyncio

domain_limits = defaultdict(
    lambda: asyncio.Semaphore(20)
)

async def fetch(session, url):
    hostname = urlparse(url).hostname

    semaphore = domain_limits[hostname]

    async with semaphore:
        async with session.get(url) as response:
            return await response.text()

Production implementations should also bound the number of semaphore objects created if the set of hosts is effectively unbounded. The principle remains: concurrency should often be controlled per destination, not only globally.

Rate limiting and concurrency are not the same

A concurrency of 20 does not mean 20 requests per second. If each request completes in 100 ms, those 20 workers could theoretically generate around 200 completions per second.

You may need both, especially when a destination publishes explicit rate limits or your own infrastructure budget requires stable throughput.

Async architecture does not remove proxy limits

A frequent mistake is reasoning that because aiohttp can run 10,000 coroutines, the proxy gateway should receive 10,000 simultaneous connections. That conclusion does not follow. Your effective limit is constrained by every system on the path:

The safe concurrency is approximately the lowest sustainable limit across the whole path. Optimize the system, not a single benchmark.

Timeouts and Retries

Set timeouts explicitly

Network calls can stall, and a proxy adds another network hop, so relying on default behavior is poor production hygiene. Use ClientTimeout:

timeout = aiohttp.ClientTimeout(
    total=30,
    connect=10,
    sock_connect=10,
    sock_read=20,
)

async with aiohttp.ClientSession(
    timeout=timeout,
) as session:
    ...

Do not copy timeout values blindly between systems. A monitoring endpoint expected to respond in 500 ms needs different limits from a cross-region data-collection job retrieving multi-megabyte pages.

A retry pattern with exponential backoff and jitter

Retries should be deliberate. A simple asynchronous retry helper could look like this:

import asyncio
import random
import aiohttp

RETRYABLE_STATUS = {
    429,
    500,
    502,
    503,
    504,
}

async def fetch_with_retry(
    session,
    url,
    proxy,
    proxy_auth,
    attempts=4,
):
    for attempt in range(attempts):
        try:
            async with session.get(
                url,
                proxy=proxy,
                proxy_auth=proxy_auth,
            ) as response:

                if response.status not in RETRYABLE_STATUS:
                    return response.status, await response.text()

                if attempt == attempts - 1:
                    return response.status, await response.text()

        except (
            asyncio.TimeoutError,
            aiohttp.ClientConnectionError,
        ):
            if attempt == attempts - 1:
                raise

        delay = min(
            2 ** attempt + random.random(),
            10,
        )

        await asyncio.sleep(delay)

The jitter matters. Without it, hundreds of workers that fail together retry together. With it, retry traffic spreads out:

This reduces synchronization during upstream incidents.

Do not retry every status code

A proxy architecture should classify failures, because they mean different things:

Before sending another request, your retry logic should answer:

1. Is the failure transient?
2. Is the request safe to retry?
3. Should the same proxy be reused?
4. Should a new session be selected?
5. Should the job be delayed?
6. Should the failure stop the workflow?

Worker Queues and Backpressure

Use asyncio.Queue for large job sets

Suppose you have 5,000,000 URLs. Creating five million asyncio tasks is a poor queueing system. Use asyncio.Queue with a fixed number of workers:

import asyncio
import aiohttp

WORKERS = 100
PROXY = "http://proxy.example.com:8080"

async def worker(name, queue, session):
    while True:
        url = await queue.get()

        if url is None:
            queue.task_done()
            break

        try:
            async with session.get(
                url,
                proxy=PROXY,
            ) as response:
                body = await response.text()

                print(
                    name,
                    response.status,
                    url,
                    len(body),
                )

        except Exception as exc:
            print(name, url, exc)

        finally:
            queue.task_done()

async def main(urls):
    queue = asyncio.Queue(maxsize=1000)

    connector = aiohttp.TCPConnector(
        limit=150,
        limit_per_host=50,
    )

    async with aiohttp.ClientSession(
        connector=connector,
    ) as session:

        workers = [
            asyncio.create_task(
                worker(i, queue, session)
            )
            for i in range(WORKERS)
        ]

        for url in urls:
            await queue.put(url)

        await queue.join()

        for _ in workers:
            await queue.put(None)

        await asyncio.gather(*workers)

This gives you natural backpressure. The producer cannot outrun the consumers forever, because asyncio.Queue(maxsize=1000) eventually makes await queue.put(url) wait. That is desirable.

Why backpressure matters

Imagine an input rate of 10,000 URLs per second and a worker rate of 1,000. Without a bounded queue, the application accumulates work faster than it can process it, and the cost eventually appears somewhere: RAM, file descriptors, task objects, timeouts, queue latency or the retry backlog. A bounded queue turns that resource pressure into controlled waiting instead. The same principle applies at crawler scale, as covered in our guide to designing distributed crawlers for billion-page datasets.

Protect downstream systems too

High-speed fetching can create a bottleneck after the request. If your fetch layer produces 10,000 responses per second but your database can insert 2,000 records per second, the HTTP client is no longer the slow component. Put bounded queues between every stage:

URL queue
 |
 v
fetchers
 |
 v
response queue
 |
 v
parsers
 |
 v
storage queue
 |
 v
database writers

Otherwise asynchronous network performance simply moves the outage downstream.

Stream large responses instead of reading everything at once

body = await response.read() loads the entire response into memory. That may be fine for small HTML pages. For large files or responses, stream the body:

async with session.get(url) as response:
    with open("output.bin", "wb") as file:
        async for chunk in response.content.iter_chunked(
            64 * 1024
        ):
            file.write(chunk)

Memory consumption is now bounded by the chunk size rather than the full response size. Under high concurrency this is significant: with 10 MB responses and a concurrency of 500, buffering every response could mean gigabytes of transient memory. Async code still obeys arithmetic.

Handle response bodies carefully

Always consume or close response objects correctly. The recommended pattern is async with session.get(url) as response:. When the context exits, the response lifecycle is managed properly. Failing to release resources reduces connection reuse and can eventually exhaust the pool. Under low traffic, lifecycle mistakes may be invisible. Under high traffic, they become incidents.

Rotating Proxies and Sticky Sessions

Client-side rotation vs a managed gateway

There are two common proxy rotation architectures. The first is client-side rotation over a list of endpoints:

import itertools

proxies = itertools.cycle([
    "http://proxy1.example.com:8000",
    "http://proxy2.example.com:8000",
    "http://proxy3.example.com:8000",
])

async def fetch(session, url):
    proxy = next(proxies)

    async with session.get(
        url,
        proxy=proxy,
    ) as response:
        return await response.text()

The second is a managed rotating gateway: the application connects to one logical proxy endpoint, and the provider manages the exit pool behind it.

application
   |
   v
one proxy gateway
   |
   +--> exit IP A
   +--> exit IP B
   +--> exit IP C
   +--> exit IP D

The managed gateway is often easier operationally. For a deeper look at the first model, see what a proxy pool is and how rotation strategies work.

Using ProxyEmpire with aiohttp

ProxyEmpire fits the gateway model. Its products include rotating residential, mobile and datacenter networks, with protocol and session options varying by product. Rotating residential and mobile proxies support HTTP, HTTPS and SOCKS5, with geographic and session controls.

The aiohttp integration stays straightforward. In the dashboard, the Connection Builder under Proxy Manager shows the gateway host (v2.proxyempire.io), the port (5000) and your username and password. Load them from the environment:

import asyncio
import aiohttp
import os

PROXY_HOST = os.environ["PROXY_HOST"]
PROXY_PORT = os.environ["PROXY_PORT"]
PROXY_USER = os.environ["PROXY_USER"]
PROXY_PASSWORD = os.environ["PROXY_PASSWORD"]

PROXY = f"http://{PROXY_HOST}:{PROXY_PORT}"

async def main():
    auth = aiohttp.BasicAuth(
        PROXY_USER,
        PROXY_PASSWORD,
    )

    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://ifconfig.me",
            proxy=PROXY,
            proxy_auth=auth,
        ) as response:
            print(await response.text())

asyncio.run(main())

Rotation and async concurrency are different controls

Async concurrency is not IP rotation. You can send 100 asynchronous requests through one IP, or through 100 different exit IPs. With aiohttp concurrency at 100 and the proxy set to a new exit per request, the result looks like this:

worker 1   -> gateway -> IP A
worker 2   -> gateway -> IP B
worker 3   -> gateway -> IP C
...
worker 100 -> gateway -> IP n

A sticky-session workload might instead group workers by session:

worker group A -> session A -> IP A
worker group B -> session B -> IP B
worker group C -> session C -> IP C

ProxyEmpire offers both rotating and sticky-session behavior across the relevant products, so your Python application keeps the asynchronous scheduling logic while the proxy layer manages exit nodes. Our guide to sticky vs rotating proxies explains when to use each.

Sticky sessions for stateful workflows

Random IP rotation is not always desirable. Consider this sequence:

GET /search
GET /search?page=2
GET /product/123
GET /availability/123

If all four requests belong to the same logical browser or application session, changing IP between steps may create inconsistent behavior. Instead, keep the logical session on one proxy session and one exit IP:

logical session ABC
      |
      +--> request 1
      +--> request 2
      +--> request 3
      +--> request 4
      |
      v
same proxy session
      |
      v
same exit IP

Map application sessions to proxy sessions

Suppose each worker processes one product journey. Create a stable session identifier:

import secrets

def new_session_id():
    return secrets.token_hex(8)

Then incorporate the ID using the provider’s supported session mechanism. Conceptually:

proxy_username = build_proxy_username(
    account="my_account",
    session=session_id,
    country="us",
)

The exact credential format is provider-specific, so generate it from your ProxyEmpire configuration rather than hard-coding assumptions into the client. ProxyEmpire’s documentation on how to use proxy sticky sessions covers the details. The broader architecture is what matters:

business task
   |
   v
application session ID
   |
   v
proxy session ID
   |
   v
stable exit identity

This also makes debugging substantially easier.

Environment Variables, SOCKS5, HTTPS Proxies and Secrets

Environment variables and trust_env

Unlike some Python HTTP libraries, aiohttp does not use proxy environment variables by default. If your infrastructure exposes them:

export HTTP_PROXY=http://proxy.example.com:8080
export HTTPS_PROXY=http://proxy.example.com:8080

you need trust_env=True:

async with aiohttp.ClientSession(
    trust_env=True,
) as session:
    ...

aiohttp then reads proxy configuration from the environment. That is convenient in containers, enterprise egress environments, CI runners and standardized developer environments. Implicit configuration is not always ideal, though. For application-specific data pipelines, explicit configuration such as session.get(url, proxy=settings.proxy_url) makes routing easier to audit than relying on ambient process state.

Respect NO_PROXY rules

With trust_env=True, environment-level bypass rules also matter:

export NO_PROXY="localhost,127.0.0.1,.internal.example.com"

This lets internal traffic bypass the external proxy, which is important when an application talks to both the public internet and internal microservices. Sending internal service traffic through an external proxy can create latency, outages, information exposure, routing loops and unexpected cost.

SOCKS5 and aiohttp

aiohttp’s built-in proxy support primarily targets HTTP proxying. If your architecture specifically requires SOCKS, teams commonly add a SOCKS-capable connector, such as the aiohttp-socks package, rather than assuming the standard proxy= argument behaves like SOCKS in tools such as cURL.

aiohttp
 |
 v
SOCKS-capable connector
 |
 v
SOCKS5 proxy
 |
 v
destination

For a production platform, don’t treat HTTP and SOCKS proxies as identical until you have tested DNS, TLS, timeout and connection-pooling behavior for the connector you actually deploy. Our comparison of SOCKS vs HTTP proxies covers the protocol differences.

HTTPS proxies need extra attention

There are two HTTPS concepts in a proxied request: an HTTP proxy to an HTTPS destination, or an HTTPS proxy to an HTTPS destination. The second involves TLS both to the proxy and inside the tunnel to the destination, so certificate trust can involve both systems. If you configure a custom SSL context, check that it trusts every certificate chain on the actual network path.

Secrets management

Do not do this in production:

PROXY_USER = "company-prod"
PROXY_PASSWORD = "super-secret-password"

Use environment variables at minimum:

import os

PROXY_USER = os.environ["PROXY_USER"]
PROXY_PASSWORD = os.environ["PROXY_PASSWORD"]

A production environment should preferably inject them from AWS Secrets Manager, Google Secret Manager, Azure Key Vault, HashiCorp Vault, Kubernetes Secrets backed by appropriate controls, or your existing enterprise secret-management platform. Proxy credentials are infrastructure credentials. Treat them accordingly.

Use a proxy configuration object

As the application grows, avoid scattering proxy strings across the codebase. Use a configuration object:

import os
from dataclasses import dataclass

import aiohttp

@dataclass(frozen=True)
class ProxyConfig:
    url: str
    username: str
    password: str

config = ProxyConfig(
    url="http://proxy.example.com:8080",
    username=os.environ["PROXY_USER"],
    password=os.environ["PROXY_PASSWORD"],
)

proxy_auth = aiohttp.BasicAuth(
    config.username,
    config.password,
)

Building authentication at the network boundary makes future changes easier, from a single proxy to a rotating gateway, a region-aware gateway, multiple providers or failover pools, without rewriting business logic.

Multiple Proxy Pools, Failover and Circuit Breakers

Multiple proxy pools

Larger systems may need several types of egress:

A routing layer can choose the proxy profile per job:

def choose_proxy(job):
    if job.requires_mobile:
        return mobile_proxy

    if job.requires_residential:
        return residential_proxy

    return datacenter_proxy

ProxyEmpire offers network categories that fit this kind of tiered architecture, including rotating residential, mobile and datacenter products. The provider should not determine application policy, though. Your application should decide which class of network fits a given workload.

Failover between proxy pools

Plan for what happens when a proxy pool becomes unhealthy:

primary residential pool
       |
       | high connection failure rate
       v
circuit breaker
       |
       v
secondary pool

Don’t fail over instantly because of one request. Use measurable thresholds, for example proxy connection errors above 15% over a rolling five-minute window, then open the circuit temporarily. This avoids flapping between providers or pools because of random noise.

The circuit-breaker pattern

if proxy_health.error_rate > threshold:
    disable_pool(proxy_pool)

if proxy_health.cooldown_expired:
    test_pool(proxy_pool)

if proxy_health.recovered:
    enable_pool(proxy_pool)

The breaker has to operate at the right level. A target-specific 403 does not necessarily mean the proxy network is unhealthy globally. Segment health by provider, pool, region, target and failure class before making routing decisions.

A Complete High-Concurrency Example

A production-oriented client with a semaphore

Putting the earlier pieces together:

import asyncio
import aiohttp

PROXY = "http://proxy.example.com:8080"

PROXY_USER = "proxy_username"
PROXY_PASSWORD = "proxy_password"

CONCURRENCY = 100

timeout = aiohttp.ClientTimeout(
    total=30,
    connect=10,
    sock_read=20,
)

semaphore = asyncio.Semaphore(CONCURRENCY)

async def fetch(session, url):
    async with semaphore:
        try:
            async with session.get(
                url,
                proxy=PROXY,
                proxy_auth=aiohttp.BasicAuth(
                    PROXY_USER,
                    PROXY_PASSWORD,
                ),
            ) as response:
                body = await response.text()

                return {
                    "url": url,
                    "status": response.status,
                    "body": body,
                    "error": None,
                }

        except asyncio.TimeoutError:
            return {
                "url": url,
                "status": None,
                "body": None,
                "error": "timeout",
            }

        except aiohttp.ClientError as exc:
            return {
                "url": url,
                "status": None,
                "body": None,
                "error": str(exc),
            }

async def main(urls):
    connector = aiohttp.TCPConnector(
        limit=150,
        limit_per_host=50,
    )

    async with aiohttp.ClientSession(
        timeout=timeout,
        connector=connector,
    ) as session:
        return await asyncio.gather(
            *(fetch(session, url) for url in urls)
        )

urls = [
    "https://example.com",
    "https://example.org",
]

results = asyncio.run(main(urls))

for result in results:
    print(result)

This is much closer to production behavior. It has session reuse, proxy authentication, explicit timeouts, connection limits, concurrency limits and basic error classification. It still needs proper retries and backpressure for serious workloads.

The full pattern: queue, workers, retries and proxy auth

import asyncio
import os
import random
import aiohttp

WORKERS = 100

PROXY = os.environ["PROXY_URL"]
PROXY_USER = os.environ["PROXY_USER"]
PROXY_PASSWORD = os.environ["PROXY_PASSWORD"]

RETRYABLE = {
    429,
    500,
    502,
    503,
    504,
}

async def fetch(
    session,
    proxy_auth,
    url,
    attempts=3,
):
    for attempt in range(attempts):
        try:
            async with session.get(
                url,
                proxy=PROXY,
                proxy_auth=proxy_auth,
            ) as response:

                body = await response.text()

                if (
                    response.status not in RETRYABLE
                    or attempt == attempts - 1
                ):
                    return {
                        "url": url,
                        "status": response.status,
                        "body": body,
                        "attempt": attempt + 1,
                    }

        except (
            asyncio.TimeoutError,
            aiohttp.ClientConnectionError,
        ) as exc:

            if attempt == attempts - 1:
                return {
                    "url": url,
                    "status": None,
                    "body": None,
                    "attempt": attempt + 1,
                    "error": str(exc),
                }

        delay = min(
            (2 ** attempt) + random.random(),
            10,
        )

        await asyncio.sleep(delay)

async def worker(
    name,
    queue,
    session,
    proxy_auth,
):
    while True:
        url = await queue.get()

        try:
            if url is None:
                return

            result = await fetch(
                session,
                proxy_auth,
                url,
            )

            print(name, result)

        finally:
            queue.task_done()

async def main(urls):
    queue = asyncio.Queue(maxsize=1000)

    timeout = aiohttp.ClientTimeout(
        total=30,
        connect=10,
        sock_read=20,
    )

    connector = aiohttp.TCPConnector(
        limit=150,
        limit_per_host=50,
    )

    proxy_auth = aiohttp.BasicAuth(
        PROXY_USER,
        PROXY_PASSWORD,
    )

    async with aiohttp.ClientSession(
        timeout=timeout,
        connector=connector,
    ) as session:

        workers = [
            asyncio.create_task(
                worker(
                    worker_id,
                    queue,
                    session,
                    proxy_auth,
                )
            )
            for worker_id in range(WORKERS)
        ]

        for url in urls:
            await queue.put(url)

        for _ in workers:
            await queue.put(None)

        await queue.join()

        await asyncio.gather(*workers)

asyncio.run(
    main([
        "https://example.com",
        "https://example.org",
    ])
)

This provides a solid foundation:

A real production system would add structured logging, metrics, persistence, graceful shutdown, proxy-session selection and destination-specific policies.

Graceful Shutdown and Idempotency

Graceful shutdown matters

Containers terminate, deployments restart and autoscaling removes workers. A production asynchronous service needs to stop without corrupting state:

SIGTERM
 |
 v
stop accepting new jobs
 |
 v
finish or checkpoint active jobs
 |
 v
close ClientSession
 |
 v
exit

Don’t simply kill thousands of in-flight operations if jobs cannot safely be replayed. If your queue system provides acknowledgements, acknowledge work only when the request and its downstream processing have reached the state your delivery semantics require.

Exactly-once is usually the wrong assumption

Distributed network systems retry. Workers crash. Connections reset. Queues redeliver. Assume some operations will happen more than once, and design around idempotency where possible by storing identifiers such as job_id, target_url, logical_request_id and attempt with every result. Duplicate execution then becomes observable and manageable rather than mysterious.

Observability and Cost

Separate network errors from application errors

Do not collapse every failure into except Exception: return None. That makes production debugging almost impossible. At minimum, distinguish the main classes:

try:
    ...
except asyncio.TimeoutError:
    failure = "timeout"

except aiohttp.ClientProxyConnectionError:
    failure = "proxy_connection_error"

except aiohttp.ClientConnectorError:
    failure = "connect_error"

except aiohttp.ClientResponseError:
    failure = "http_error"

except aiohttp.ClientError:
    failure = "aiohttp_error"

Then record structured fields:

{
  "target": "example.com",
  "proxy_pool": "residential_us",
  "status": 502,
  "failure_type": "http_error",
  "attempt": 2,
  "latency_ms": 1843
}

That is far more actionable than “Request failed.”

Measure latency

You can measure request duration at the application layer:

import time

start = time.perf_counter()

async with session.get(
    url,
    proxy=proxy,
) as response:
    body = await response.text()

elapsed = time.perf_counter() - start

One request is not useful infrastructure data, though. Track distributions such as p50, p95 and p99 latency, timeout rate, connection failure rate, HTTP failure rate, successful requests per second, bytes per second and retry rate, and segment them by proxy pool, country, destination, worker, status and session mode.

What to log

Observability should improve debugging without becoming a credential leak.

Metrics to put on the dashboard

A high-concurrency proxy platform deserves first-class infrastructure metrics. Track at least:

Segment them where useful by proxy provider, proxy network, country, destination and application. The dashboard should answer one question: is throughput down because the application is failing, the proxy is failing, or the destination is responding differently? If it can’t, the observability model is incomplete.

The key metric: cost per successful outcome

Proxy networks are often priced by bandwidth or another consumption metric, but price per GB is not necessarily the right procurement metric. A better one is total network cost divided by successful business outcomes:

      total network cost
-----------------------------
successful business outcomes

For a data platform, that might be cost per successful page, per valid record or per completed monitoring job. A cheaper network with more failed responses can produce more expensive usable data. Engineering and procurement should evaluate the same denominator.

Common aiohttp Proxy Mistakes

1. Creating one ClientSession per request. Opening async with aiohttp.ClientSession() inside the loop over URLs throws away pooling. Reuse the session.
2. Treating gather() as concurrency control. asyncio.gather(*(fetch(url) for url in millions_of_urls)) is not a limit. Use bounded workers or semaphores.
3. Setting TCPConnector(limit=0) without a reason. An unlimited connector is not automatically faster. It can shift failures into socket exhaustion, proxy overload, target overload and memory pressure. Unlimited should be an explicit architecture decision, not a performance shortcut.
4. Retrying authentication failures. A 407 caused by invalid credentials won’t disappear because you retry it five times. Classify first, retry second.
5. Treating every 403 as a proxy failure. 403 Forbidden is an application response whose meaning depends on the destination. Don’t rotate network identity indefinitely.
6. Ignoring downstream capacity. If parsing or storage can’t keep up, more HTTP concurrency makes the system worse.
7. Hard-coding proxy credentials. Keep secrets outside application code.
8. Confusing concurrency with requests per second. Twenty concurrent requests can generate far more than 20 requests per second.
9. Rotating IPs during workflows that need continuity. Use sticky sessions where application state depends on a stable network identity.
10. Disabling SSL verification to fix TLS problems. Fix certificate trust instead.

aiohttp Proxy Cheat Sheet

async with session.get(
    "https://example.com",
    proxy="http://proxy.example.com:8080",
) as response:
    ...

proxy_auth = aiohttp.BasicAuth(
    "username",
    "password",
)

async with session.get(
    "https://example.com",
    proxy="http://proxy.example.com:8080",
    proxy_auth=proxy_auth,
) as response:
    ...

proxy = (
    "http://username:password"
    "@proxy.example.com:8080"
)

async with aiohttp.ClientSession(
    trust_env=True,
) as session:
    ...

timeout = aiohttp.ClientTimeout(
    total=30,
    connect=10,
    sock_read=20,
)

connector = aiohttp.TCPConnector(
    limit=200,
    limit_per_host=50,
)

semaphore = asyncio.Semaphore(100)

queue = asyncio.Queue(
    maxsize=1000,
)

When the Proxy Layer Becomes an Infrastructure Decision

At modest scale, proxy configuration can remain a single line such as proxy = "http://proxy.example.com:8080". At larger scale, the engineering questions change:

• Which geography?
• Which proxy type?
• How long should the session persist?
• What happens when an exit fails?
• How many requests can run concurrently?
• What is the retry policy?
• How much bandwidth are we consuming?
• What does a successful result cost?

This is where a provider such as ProxyEmpire becomes part of the network architecture rather than another string in a Python script. A reasonable separation of concerns is:

This division keeps application logic independent from individual proxy nodes and makes scaling easier to reason about.

Security and compliance

Proxy infrastructure changes the network route. It does not change the organization’s obligations. Teams should still establish controls for applicable law, target terms, robots policies where relevant, data minimization, personal-data handling, access controls, credential rotation, rate limits, retention policies and auditability.

If a destination provides an appropriate official API, evaluate it before building an extraction system around its web interface. Proxy routing should be part of a documented data-access policy, especially when several engineering teams share the same infrastructure.

Final Thoughts

Using a proxy with aiohttp takes one request parameter. Building a reliable asynchronous proxy system takes substantially more engineering:

• Reuse ClientSession objects so connections can be pooled.
• Authenticate to the proxy separately from the destination.
• Set explicit timeouts.
• Bound both request concurrency and connection concurrency.
• Use queues for backpressure.
• Retry only failures that are actually retryable, and add jitter.
• Preserve proxy sessions when workflows require continuity.
• Record enough metrics to tell network failures from target failures and application failures.

For high-volume workloads, managed infrastructure such as ProxyEmpire can handle the proxy-network side, including rotating and sticky exit behavior, while your Python application stays responsible for scheduling, concurrency, request semantics, validation and observability. That separation is the key architectural point.

Frequently Asked Questions