Python aiohttp Tunnel Proxy: Async Four IP Control Scenarios

Why aiohttp

aiohttp is Python's most mature async HTTP framework. For proxy scenarios, its proxy_headers parameter injects custom headers at the CONNECT stage — similar to httpx, simpler than requests.

Setup

export PROXY_HOST=t.16yun.cn
export PROXY_PORT=31111
export PROXY_USERNAME=your-username
export PROXY_PASSWORD=your-password
export PROXY_TUNNEL=aiohttp-demo-12345

Scenario A: Force New Connection

New ClientSession per request, no connection reuse.

import asyncio, os, aiohttp
host, port = os.getenv("PROXY_HOST","t.16yun.cn"), os.getenv("PROXY_PORT","31111")
user, pwd = os.getenv("PROXY_USERNAME","user"), os.getenv("PROXY_PASSWORD","password")
proxy = f"http://{host}:{port}"
auth = aiohttp.BasicAuth(user, pwd)

async def scenario_a():
    for i in range(1, 4):
        connector = aiohttp.TCPConnector(limit=1, limit_per_host=1)
        async with aiohttp.ClientSession(connector=connector) as s:
            async with s.get("https://httpbin.org/ip", proxy=proxy,
                proxy_auth=auth, headers={"Connection":"close","Proxy-Connection":"close"},
                timeout=aiohttp.ClientTimeout(total=15)) as r:
                d = await r.json(); print(f"Request {i}: {d['origin']}")
            await asyncio.sleep(0.5)
asyncio.run(scenario_a())

Scenario B: Keep-Alive Same IP

Reuse one ClientSession with persistent connector.

async def scenario_b():
    connector = aiohttp.TCPConnector(limit=10, keepalive_timeout=30)
    async with aiohttp.ClientSession(connector=connector) as s:
        for i in range(1, 4):
            async with s.get("https://httpbin.org/ip", proxy=proxy,
                proxy_auth=auth, headers={"Connection":"keep-alive","Proxy-Connection":"keep-alive"},
                timeout=aiohttp.ClientTimeout(total=15)) as r:
                d = await r.json(); print(f"Request {i}: {d['origin']}")
            await asyncio.sleep(0.2)

Scenario C-HTTP: Proxy-Tunnel for HTTP

async def scenario_c_http():
    tunnel = os.getenv("PROXY_TUNNEL", "aiohttp-demo")
    for i in range(1, 4):
        async with aiohttp.ClientSession() as s:
            async with s.get("http://httpbin.org/ip", proxy=proxy,
                proxy_auth=auth, headers={"Proxy-Tunnel":tunnel,"Connection":"keep-alive"},
                timeout=aiohttp.ClientTimeout(total=15)) as r:
                d = await r.json(); print(f"Request {i}: {d['origin']}")
            await asyncio.sleep(0.2)

Scenario C-HTTPS: Proxy-Tunnel for HTTPS

Use proxy_headers to inject Proxy-Tunnel at the CONNECT stage:

async def scenario_c_https():
    tunnel = os.getenv("PROXY_TUNNEL", "aiohttp-demo")
    for i in range(1, 4):
        async with aiohttp.ClientSession() as s:
            async with s.get("https://httpbin.org/ip", proxy=proxy,
                proxy_auth=auth, proxy_headers={"Proxy-Tunnel":tunnel},
                headers={"Connection":"keep-alive"},
                timeout=aiohttp.ClientTimeout(total=15)) as r:
                d = await r.json(); print(f"Request {i}: {d['origin']} (HTTPS)")
            await asyncio.sleep(0.2)

proxy_headers is aiohttp-specific. These headers are sent during the HTTPS CONNECT handshake. requests requires a custom HTTPAdapter for the same effect.

Scenario Comparison