Browser Use Proxy Setup for AI Agents

import asyncio
import os
from browser_use import Agent, BrowserProfile, BrowserSession
from langchain_anthropic import ChatAnthropic

async def run_agent_task(task: str) -> str:
    # 1. Configure BrowserProfile with Coronium mobile proxy
    profile = BrowserProfile(
        proxy={
            "server": "http://mobile.coronium.io:10001",
            "username": os.environ["CORONIUM_USER"],
            "password": os.environ["CORONIUM_PASS"],
        },
        user_agent=(
            "Mozilla/5.0 (Linux; Android 14; Pixel 8) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/132.0.0.0 Mobile Safari/537.36"
        ),
        viewport={"width": 412, "height": 915},
        device_scale_factor=2.625,
        is_mobile=True,
        has_touch=True,
        user_data_dir="./browser-profile",  # persist cookies
    )

    # 2. Start BrowserSession
    session = BrowserSession(browser_profile=profile)
    await session.start()

    try:
        # 3. Configure Claude 3.5 Sonnet (2026 top choice)
        llm = ChatAnthropic(
            model="claude-3-5-sonnet-20241022",
            temperature=0.0,
            anthropic_api_key=os.environ["ANTHROPIC_API_KEY"],
        )

        # 4. Run the agent
        agent = Agent(
            task=task,
            llm=llm,
            browser_session=session,
            max_steps=30,
            max_failures=3,
        )
        result = await agent.run()
        return result.final_result()
    finally:
        # 5. Clean up browser + proxy connection
        await session.close()

if __name__ == "__main__":
    task = (
        "Go to coronium.io, find the mobile proxy pricing, "
        "and return the starting monthly price."
    )
    result = asyncio.run(run_agent_task(task))
    print(f"Agent result: {result}")

import redis
import json
from contextlib import asynccontextmanager
from browser_use import BrowserProfile

class CoroniumProxyPool:
    def __init__(self, proxies: list[dict], redis_url: str):
        self.proxies = {p["id"]: p for p in proxies}
        self.redis = redis.from_url(redis_url)

    @asynccontextmanager
    async def acquire(self, agent_id: str):
        # Atomically grab an idle proxy with best success rate
        proxy_id = self._select_best_idle_proxy()
        if not proxy_id:
            raise RuntimeError("No idle proxies available")

        self.redis.setex(f"proxy:{proxy_id}:owner", 3600, agent_id)
        proxy = self.proxies[proxy_id]

        profile = BrowserProfile(
            proxy={
                "server": f"http://{proxy['host']}:{proxy['port']}",
                "username": proxy["user"],
                "password": proxy["pass"],
            },
            is_mobile=True,
            user_data_dir=f"./profiles/{proxy_id}",
        )
        try:
            yield profile, proxy_id
        finally:
            self.redis.delete(f"proxy:{proxy_id}:owner")
            self._update_success_rate(proxy_id, success=True)

    def _select_best_idle_proxy(self) -> str | None:
        candidates = [
            pid for pid in self.proxies
            if not self.redis.exists(f"proxy:{pid}:owner")
        ]
        # Rank by recent success rate stored in Redis
        return max(
            candidates,
            key=lambda p: float(
                self.redis.get(f"proxy:{p}:success_rate") or 0.95
            ),
            default=None,
        )