Proxy Integration With AIOHTTP

This tutorial will show you how to use Python’s AIOHTTP and asyncio with Anonymous Proxies’ HTTP and SOCKS5 proxies, then move on to simple proxy rotation examples for larger scraping runs.

Recommended proxy services

HTTP Proxies

HTTP Proxies are handling HTTP requests towards the internet on behalf of a client. They are fast and very popular when it comes to any kind of anonymous web browsing.

SOCKSv5

SOCKSv5 is an internet protocol that is more versatile than a regular HTTP proxy since it can run on any port and traffic can flow both on TCP and UDP. Useful in games and other applications that do not use the http protocol.

What Is AIOHTTP?

AIOHTTP is a popular, open-source Python framework for working with HTTP asynchronously, built on top of asyncio. Think of it as a toolkit that speaks “web” fluently in both directions. You can use it to build a server that receives requests, adds middleware, and keeps routes tidy and easy to manage. You can also use it as a client to make HTTP requests without freezing your code while the network does its thing. And when you need real-time updates, AIOHTTP includes WebSocket support, so two way communication feels like a natural part of the same workflow.

How to Set Up a Proxy in AIOHTTP?

Prerequisites

Before you start, make sure you have these ready:

Python 3.9 or newer
Anonymous Proxies access details, plus your username and password if your plan uses authentication

Installing AIOHTTP

Install aiohttp with pip.

On macOS or Linux:

pip install aiohttp

On Windows, or whenever you want to be 100% sure you are using the right Python interpreter:

python -m pip install aiohttp

If you plan to use SOCKS5 proxies, install the SOCKS connector package as well:

pip install aiohttp-socks

asyncio comes bundled with Python, so you do not need to install it separately.

Once installed, import what you need in your Python file:

import aiohttp
import asyncio

# Only needed for SOCKS5
from aiohttp_socks import ProxyConnector

Set Up a Static HTTP Proxy in AIOHTTP

A static HTTP proxy setup is the quickest way to confirm your Anonymous Proxies endpoint is working. Every request goes through the same proxy, so if something fails you can troubleshoot without any extra moving parts like rotation or pooling.

In the script below, we do two small checks:

Confirm the request is routed through the proxy by calling an IP endpoint
Fetch a normal HTML page and extract a clean piece of text from it

import asyncio
import aiohttp
import re

PROXY_URL = "http://YOUR_PROXY_HOST:YOUR_PROXY_PORT"
PROXY_AUTH = aiohttp.BasicAuth("YOUR_USERNAME", "YOUR_PASSWORD")

IP_CHECK_URL = "https://httpbin.org/ip"
HTML_TEST_URL = "https://httpbin.org/html"

def extract_heading(html: str) -> str:
    match = re.search(r"<h1>\s*(.*?)\s*</h1>", html, re.IGNORECASE | re.DOTALL)
    return match.group(1).strip() if match else "No heading found"

async def main():
    timeout = aiohttp.ClientTimeout(total=20)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(IP_CHECK_URL, proxy=PROXY_URL, proxy_auth=PROXY_AUTH) as resp:
            print("IP check status", resp.status)
            print("IP check body", await resp.text())

        async with session.get(HTML_TEST_URL, proxy=PROXY_URL, proxy_auth=PROXY_AUTH) as resp:
            print("HTML page status", resp.status)
            html = await resp.text()

    heading = extract_heading(html)
    print("\nExtracted heading")
    print(heading)

asyncio.run(main())

You should see 200 responses, your proxy exit IP, and the extracted result printed.

AIOHTTP HTTP proxy test output showing exit IP and status 200

Set Up a Static SOCKS5 Proxy in AIOHTTP

With SOCKS5, the clean approach is to route traffic through a connector. That way, your session is born “proxy-aware” and your request calls stay simple.

This script follows the same two-check flow as the HTTP example, but uses a SOCKS connector instead.

import asyncio
import aiohttp
import re
from aiohttp_socks import ProxyConnector

SOCKS_PROXY = "socks5://YOUR_USERNAME:YOUR_PASSWORD@YOUR_PROXY_HOST:YOUR_PROXY_PORT"

IP_CHECK_URL = "https://httpbin.org/ip"
HTML_TEST_URL = "https://httpbin.org/html"

def extract_heading(html: str) -> str:
    match = re.search(r"<h1>\s*(.*?)\s*</h1>", html, re.IGNORECASE | re.DOTALL)
    return match.group(1).strip() if match else "No heading found"

async def main():
    connector = ProxyConnector.from_url(SOCKS_PROXY)
    timeout = aiohttp.ClientTimeout(total=20)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        async with session.get(IP_CHECK_URL) as resp:
            print("IP check status", resp.status)
            print("IP check body", await resp.text())

        async with session.get(HTML_TEST_URL) as resp:
            print("HTML page status", resp.status)
            html = await resp.text()

    heading = extract_heading(html)
    print("\nExtracted heading")
    print(heading)

asyncio.run(main())

You should see 200 responses, your SOCKS5 exit IP, and the extracted result printed.

AIOHTTP SOCKS5 proxy test output showing exit IP and status 200

Important note

If you enabled authentication in your Anonymous Proxies dashboard, you must fill in the YOUR_USERNAME and YOUR_PASSWORD fields (and use the correct proxy host and port) exactly as shown in your dashboard credentials.

If you are using IP whitelisting, you can skip username and password entirely. In that case, keep only the proxy host and port for HTTP proxies, and use a SOCKS URL without credentials for SOCKS5.

How to Use Rotating Proxies with AIOHTTP

If you plan to scale your scraper, rotating your proxies is one of the simplest ways to keep it stable. Instead of sending every request from one IP, you cycle through a small pool. That helps reduce slowdowns and lowers the chance of getting blocked when you scrape more pages or run jobs more often.

Choose the right proxy type

Now, for scraping, residential proxies are usually a must because many sites flag and block datacenter IPs quickly. Residential IPs blend in more naturally with real user traffic, which often means fewer captchas and fewer surprise bans. Datacenter proxies can still work well for lighter targets, public APIs, or speed-focused tasks, but if you are dealing with stricter websites, a residential pool plus rotation is typically the more reliable setup.

Rotating HTTP Proxies in AIOHTTP

Here’s a quick script that rotates your HTTP proxies as it runs, so each page goes out through a different proxy from your list.

import asyncio
import aiohttp
import re
from itertools import cycle

BASE_URL = "https://quotes.toscrape.com/page/{}/"

PROXIES = [
    "PROXY_HOST_1:PROXY_PORT_1:YOUR_USERNAME:YOUR_PASSWORD",
    "PROXY_HOST_2:PROXY_PORT_2:YOUR_USERNAME:YOUR_PASSWORD",
    "PROXY_HOST_3:PROXY_PORT_3:YOUR_USERNAME:YOUR_PASSWORD",
]

QUOTE_RE = re.compile(r'<span class="text" itemprop="text">([^<]+)</span>')

def parse_quotes(html: str) -> list[str]:
    return QUOTE_RE.findall(html)

def build_http_proxy(proxy_line: str):
    host, port, user, pwd = proxy_line.split(":", 3)
    proxy_url = f"http://{host}:{port}"
    proxy_auth = aiohttp.BasicAuth(user, pwd)
    return proxy_url, proxy_auth

async def fetch_page(session: aiohttp.ClientSession, sem: asyncio.Semaphore, page_no: int, proxy_line: str):
    url = BASE_URL.format(page_no)
    proxy_url, proxy_auth = build_http_proxy(proxy_line)

    async with sem:
        last_error = None
        for attempt in range(3):
            try:
                async with session.get(url, proxy=proxy_url, proxy_auth=proxy_auth) as resp:
                    if resp.status in (429, 502, 503, 504):
                        await asyncio.sleep(0.4 * (attempt + 1))
                        continue

                    html = await resp.text()
                    quotes = parse_quotes(html)
                    return page_no, quotes, proxy_url
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                last_error = e
                await asyncio.sleep(0.4 * (attempt + 1))

        return page_no, [], proxy_url

async def main():
    timeout = aiohttp.ClientTimeout(total=25)
    sem = asyncio.Semaphore(10)

    proxy_ring = cycle(PROXIES)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        tasks = []
        for page_no in range(1, 4):
            proxy_line = next(proxy_ring)
            tasks.append(fetch_page(session, sem, page_no, proxy_line))

        results = await asyncio.gather(*tasks)

    for page_no, quotes, proxy_used in sorted(results, key=lambda x: x[0]):
        print(f"\nPage {page_no} via {proxy_used}")
        for q in quotes[:5]:
            print(q)

asyncio.run(main())

Once you run it, you will see each page number printed with the proxy endpoint used, followed by a few quotes pulled from that page.

AIOHTTP HTTP proxy rotation output showing different proxy IPs per page

Rotating SOCKS5 Proxies in AIOHTTP

Here’s the SOCKS5 version, which cycles through multiple SOCKS5 proxies by rotating sessions, so each page is requested through a different proxy IP.

import asyncio
import aiohttp
import re
from itertools import cycle
from urllib.parse import urlparse
from aiohttp_socks import ProxyConnector

BASE_URL = "https://quotes.toscrape.com/page/{}/"

SOCKS_PROXIES = [
    "socks5://YOUR_USERNAME:YOUR_PASSWORD@PROXY_HOST_1:PROXY_PORT_1",
    "socks5://YOUR_USERNAME:YOUR_PASSWORD@PROXY_HOST_2:PROXY_PORT_2",
    "socks5://YOUR_USERNAME:YOUR_PASSWORD@PROXY_HOST_3:PROXY_PORT_3",
]

QUOTE_RE = re.compile(r'<span class="text" itemprop="text">([^<]+)</span>')

def parse_quotes(html: str) -> list[str]:
    return QUOTE_RE.findall(html)

async def fetch_page(
    session: aiohttp.ClientSession,
    sem: asyncio.Semaphore,
    page_no: int,
    proxy_ip: str,
):
    url = BASE_URL.format(page_no)

    async with sem:
        async with session.get(url) as resp:
            html = await resp.text()
            return page_no, parse_quotes(html), proxy_ip

async def main():
    timeout = aiohttp.ClientTimeout(total=25)
    sem = asyncio.Semaphore(10)

    sessions = []
    for proxy_url in SOCKS_PROXIES:
        proxy_ip = urlparse(proxy_url).hostname or "unknown"
        connector = ProxyConnector.from_url(proxy_url)
        session = aiohttp.ClientSession(connector=connector, timeout=timeout)
        sessions.append((session, proxy_ip))

    try:
        ring = cycle(sessions)
        tasks = []
        for page_no in range(1, 4):
            session, proxy_ip = next(ring)
            tasks.append(fetch_page(session, sem, page_no, proxy_ip))

        results = await asyncio.gather(*tasks)

        for page_no, quotes, proxy_ip in sorted(results, key=lambda x: x[0]):
            print(f"\nPage {page_no} via {proxy_ip}")
            for q in quotes[:5]:
                print(q)
    finally:
        await asyncio.gather(*(session.close() for session, _ in sessions))

asyncio.run(main())

After you run it, you will see each page printed with the proxy IP used and a few quotes from the page.

AIOHTTP SOCKS5 proxy rotation output showing different proxy IPs per page

Quick setup reminder

If your Anonymous Proxies plan uses authentication, fill in YOUR_USERNAME and YOUR_PASSWORD using the dashboard details. If you use IP whitelisting, skip credentials entirely and use only the proxy host and port.

Conclusion

AIOHTTP and proxies are a great combination when you need to scrape data on a large scale. AIOHTTP ensures that you are always fast and responsive, with concurrent requests, while proxies help you bypass the common issues of scraping, such as IP blocking, rate limiting, and geo-restrictions. In this tutorial, you have learned how to integrate Anonymous Proxies with AIOHTTP, from a simple one-proxy configuration to rotating proxy lists for a more stable and scalable process.

If you get stuck with the setup, you can contact our support team and they will be happy to help you. And if you want to see more setups like this for other 3rd party tools, you can take a look at our integrations page.