Mobile Proxies for AI Agents (LangChain, CrewAI, AutoGPT)

• →
  Agents scale fast. A single LangChain or CrewAI loop can fire hundreds of requests per minute. Datacenter IPs (AWS, GCP, Hetzner) are flagged by Cloudflare and DataDome the moment the throughput signature looks non-human.
• →
  Shared ranges are noisy. Every other agent operator is renting from the same cloud ASNs. Sites rate-limit those ranges collectively — your agent pays for other people's abuse.
• →
  CGNAT carrier IPs are shared with real users. Rate limits still apply per IP, but outright blocks are commercially painful — the site would block millions of real phone customers to reach you.

LangChain's ChatOpenAI, ChatAnthropic, and tool clients accept a custom http_client. You can pass an httpx.Client configured with your mobile proxy. There's also the env var path:

Option A — environment variables

export HTTPS_PROXY="http://USER:PASS@hostname:http_port"
export HTTP_PROXY="http://USER:PASS@hostname:http_port"

Option B — explicit httpx client (recommended)

import httpx
from langchain_openai import ChatOpenAI

proxy_url = "http://USER:PASS@hostname:http_port"
http_client = httpx.Client(proxy=proxy_url)

llm = ChatOpenAI(
    model="gpt-4o",
    http_client=http_client,
)

response = llm.invoke("What's the capital of France?")
print(response.content)

For agents that actually browse pages, use the Playwright browser toolkit. Launch the browser yourself with proxy credentials, then hand it to PlayWrightBrowserToolkit.from_browser(). The toolkit exposes tools like navigate_browser, click_element, and extract_text.

import asyncio
from playwright.async_api import async_playwright
from langchain_community.agent_toolkits import PlayWrightBrowserToolkit
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate

async def setup():
    p = await async_playwright().start()
    browser = await p.chromium.launch(
        headless=True,
        proxy={
            "server": "http://hostname:http_port",
            "username": "USER",
            "password": "PASS",
        },
    )
    toolkit = PlayWrightBrowserToolkit.from_browser(async_browser=browser)
    tools = toolkit.get_tools()

    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a web research agent."),
        ("user", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    result = await executor.ainvoke(
        {"input": "Visit example.com and summarize the page."}
    )
    print(result["output"])
    await browser.close()

asyncio.run(setup())

Every page the agent opens egresses from the mobile IP you supplied. Chromium authenticates using the username/password fields in the proxy object — no CDP hacks needed.

CrewAI ships a collection of scraping and browser tools under crewai_tools. The proxy story depends on which tool you use:

Selenium + CrewAI example

from crewai import Agent, Task, Crew
from crewai_tools import SeleniumScrapingTool
from selenium.webdriver.chrome.options import Options

PROXY_HOST = "hostname"
PROXY_PORT = "http_port"
PROXY_USER = "USER"
PROXY_PASS = "PASS"

class MobileProxySeleniumTool(SeleniumScrapingTool):
    """Selenium tool that routes traffic through a mobile proxy."""

    def _build_options(self) -> Options:
        options = Options()
        options.add_argument("--headless=new")
        options.add_argument("--no-sandbox")
        options.add_argument(
            f"--proxy-server=http://{PROXY_HOST}:{PROXY_PORT}"
        )
        # For authenticated proxies with Selenium, use selenium-wire
        # or a Chrome extension (Selenium Manager does not auth prompt).
        return options

scraper = MobileProxySeleniumTool()

researcher = Agent(
    role="Market Researcher",
    goal="Collect pricing data from public product pages",
    backstory="Expert at distilling public web data into insights.",
    tools=[scraper],
    allow_delegation=False,
)

task = Task(
    description="Scrape https://example.com and summarize visible pricing.",
    expected_output="A bulleted summary of prices found on the page.",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[task], verbose=True)
print(crew.kickoff())

For Basic-Auth proxies with Selenium, the clean path is selenium-wire or a tiny Chrome extension generated at runtime — see our dedicated Selenium mobile proxy setup guide.

Long-running agents benefit from a fresh IP between subtasks. mobileproxies.org exposes a Bearer-authenticated endpoint that triggers the modem to reconnect and request a new carrier IP:

import time
import requests

API_BASE = "https://buy.mobileproxies.org"

def list_proxies(api_key: str):
    r = requests.get(
        f"{API_BASE}/api/v1/proxies",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=15,
    )
    r.raise_for_status()
    return r.json()

def rotate_proxy_ip(slot_id: str, api_key: str):
    r = requests.post(
        f"{API_BASE}/api/v1/proxies/{slot_id}/switch",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=30,
    )
    r.raise_for_status()
    # Modem needs ~10s to reconnect and pick up a new carrier IP
    time.sleep(10)
    return r.json() if r.content else None

Wire it into your agent's task loop:

API_KEY = os.environ["MP_API_KEY"]
SLOT_ID = os.environ["MP_SLOT_ID"]

for subtask in agent_subtasks:
    rotate_proxy_ip(SLOT_ID, API_KEY)   # fresh IP
    result = executor.invoke({"input": subtask})
    results.append(result)

Rotate between logical units (a new account, a new target domain) rather than mid-session. Mid-session rotation breaks sticky cookies and TCP connections.

OpenAI and Anthropic both return a retry-after header (in seconds) on 429 responses. Websites your agent scrapes also return it per RFC 6585. The correct pattern is full jitter exponential backoff: sleep a random amount in [0, min(cap, base * 2**attempt)] so retries from many agents don't synchronize.

import random
import time
import httpx

BASE = 1.0       # starting delay in seconds
CAP  = 60.0      # maximum single sleep
MAX_ATTEMPTS = 6

def fetch_with_backoff(client: httpx.Client, url: str) -> httpx.Response:
    for attempt in range(MAX_ATTEMPTS):
        resp = client.get(url)

        if resp.status_code != 429 and resp.status_code < 500:
            return resp

        # Honor Retry-After first (seconds), fall back to full jitter.
        retry_after = resp.headers.get("retry-after")
        if retry_after and retry_after.isdigit():
            sleep_s = float(retry_after)
        else:
            sleep_s = random.uniform(0, min(CAP, BASE * (2 ** attempt)))

        time.sleep(sleep_s)

    resp.raise_for_status()
    return resp

Combine this with IP rotation: if a single IP keeps 429-ing on a target, rotate the slot and reset the backoff counter. The carrier IP pool resets rate-limit state site-side.