Python mobile proxy setup

export PROXY_URL="http://USERNAME-mbl-us-rot-auto10:[email protected]:7000"

# Same string works for http:// and https:// targets — the client
# issues a CONNECT tunnel for TLS automatically.

import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

PROXY = os.environ["PROXY_URL"]

session = requests.Session()
session.proxies = {"http": PROXY, "https": PROXY}

retry = Retry(
    total=4,
    backoff_factor=1.5,                    # 1.5s, 3s, 6s, 12s
    status_forcelist=(429, 500, 502, 503, 504),
    respect_retry_after_header=True,
    allowed_methods=frozenset(["GET", "HEAD"]),
)
# pool_maxsize keeps tunnels alive instead of reconnecting per request.
session.mount("https://", HTTPAdapter(max_retries=retry, pool_maxsize=20))
session.mount("http://", HTTPAdapter(max_retries=retry, pool_maxsize=20))

# Assert the exit before the job commits to anything.
who = session.get("https://ipinfo.io/json", timeout=30).json()
print(f"exit {who['ip']} — {who['country']} — {who['org']}")

r = session.get("https://example.com", timeout=30)
print(r.status_code, len(r.content))

Session plus pool_maxsize is the difference between one tunnel reused and one tunnel per request.

import os, asyncio, httpx

PROXY = os.environ["PROXY_URL"]

# Sync
with httpx.Client(proxy=PROXY, timeout=30.0) as client:
    print(client.get("https://api.ipify.org?format=json").json())

# Async — bound concurrency, or you will open a tunnel per task.
async def main(urls):
    limits = httpx.Limits(max_connections=10, max_keepalive_connections=10)
    sem = asyncio.Semaphore(10)
    async with httpx.AsyncClient(proxy=PROXY, limits=limits, timeout=30.0) as c:
        async def one(u):
            async with sem:
                r = await c.get(u)
                return u, r.status_code, len(r.content)
        return await asyncio.gather(*(one(u) for u in urls))

print(asyncio.run(main(["https://example.com"] * 5)))

httpx 0.26+ takes proxy=; older versions used proxies=. Bound concurrency explicitly — unbounded async is what gets a client rate-limited.

import os, asyncio, aiohttp

PROXY = os.environ["PROXY_URL"]

async def main():
    conn = aiohttp.TCPConnector(limit=10)
    async with aiohttp.ClientSession(connector=conn) as s:
        # aiohttp takes proxy= on the call, not on the session.
        async with s.get("https://api.ipify.org?format=json", proxy=PROXY) as r:
            print(r.status, await r.json())

asyncio.run(main())

A common gotcha: setting proxy on ClientSession does nothing — it belongs on each request.

import os, re

def with_country(url: str, cc: str) -> str:
    return re.sub(r"-(us|gb|de|fr|es|nl|pl|ch|pa|am|ge)-", f"-{cc}-", url, count=1)

def with_session(url: str, sid: str) -> str:
    if "-sid-" in url:
        return re.sub(r"-sid-[a-z0-9_]+", f"-sid-{sid}", url)
    return url.replace("-rot-", f"-sid-{sid}-rot-")

base = os.environ["PROXY_URL"]
print(with_country(base, "de"))
print(with_session(with_country(base, "fr"), "worker07"))

Routing lives in the username string, so switching market is a substitution — not a redeploy.