OpenAI Operator Proxy SetupGPT browser agents with mobile IPs

from openai import OpenAI

# Initialize OpenAI client
client = OpenAI()

# Create Operator session with proxy
# Note: API structure may vary based on final release
session = client.operator.sessions.create(
    model="gpt-4-operator",
    browser_config={
        "proxy": {
            "server": "socks5://proxy.proxies.sx:10001",
            "username": "your_username",
            "password": "your_password"
        },
        "viewport": {"width": 1920, "height": 1080},
        "user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)..."
    }
)

# Execute task
result = session.run(
    task="Search for flights from NYC to London on December 25th"
)

print(result.output)

import asyncio
import base64
from openai import OpenAI
from playwright.async_api import async_playwright

class OperatorAgent:
    def __init__(self, proxy_server: str, proxy_user: str, proxy_pass: str):
        self.client = OpenAI()
        self.proxy_config = {
            "server": proxy_server,
            "username": proxy_user,
            "password": proxy_pass
        }

    async def run(self, task: str):
        async with async_playwright() as p:
            browser = await p.chromium.launch(
                headless=True,
                proxy=self.proxy_config
            )

            context = await browser.new_context(
                viewport={"width": 1920, "height": 1080},
                user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)..."
            )

            page = await context.new_page()
            await page.goto("https://google.com")

            while True:
                # Take screenshot
                screenshot = await page.screenshot()
                screenshot_b64 = base64.b64encode(screenshot).decode()

                # Ask GPT-4V what to do
                response = self.client.chat.completions.create(
                    model="gpt-4-vision-preview",
                    messages=[{
                        "role": "user",
                        "content": [
                            {"type": "text", "text": f"Task: {task}\n\nWhat action should I take?"},
                            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{screenshot_b64}"}}
                        ]
                    }]
                )

                action = self.parse_action(response.choices[0].message.content)
                if action["type"] == "done":
                    break

                await self.execute_action(page, action)

            await browser.close()

    async def execute_action(self, page, action):
        if action["type"] == "click":
            await page.click(f"text={action['target']}")
        elif action["type"] == "type":
            await page.fill(action["selector"], action["text"])
        elif action["type"] == "scroll":
            await page.evaluate(f"window.scrollBy(0, {action['amount']})")

# Usage
agent = OperatorAgent(
    proxy_server="socks5://proxy.proxies.sx:10001",
    proxy_user="your_username",
    proxy_pass="your_password"
)
asyncio.run(agent.run("Book a table at a restaurant in Manhattan for 2 people"))

# .env configuration for Operator agents

# OpenAI
OPENAI_API_KEY=sk-your-api-key

# Proxy Configuration
PROXY_SERVER=socks5://proxy.proxies.sx:10001
PROXY_USERNAME=your_username
PROXY_PASSWORD=your_password

# Browser Settings
BROWSER_HEADLESS=true
BROWSER_VIEWPORT_WIDTH=1920
BROWSER_VIEWPORT_HEIGHT=1080

# Agent Settings
MAX_STEPS=50
SCREENSHOT_QUALITY=80
ACTION_DELAY_MS=500

class OperatorSessionManager:
    """Manage Operator sessions with proxy rotation"""

    def __init__(self, proxy_ports: list[int]):
        self.proxy_ports = proxy_ports
        self.current_index = 0
        self.sessions = {}

    def get_proxy_for_task(self, task_id: str, sticky: bool = True) -> dict:
        """Get proxy configuration for a task"""
        if sticky and task_id in self.sessions:
            return self.sessions[task_id]

        # Round-robin port selection
        port = self.proxy_ports[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxy_ports)

        proxy_config = {
            "server": f"socks5://proxy.proxies.sx:{port}",
            "username": "your_username",
            "password": "your_password"
        }

        if sticky:
            self.sessions[task_id] = proxy_config

        return proxy_config

    def release_session(self, task_id: str):
        """Release a sticky session"""
        if task_id in self.sessions:
            del self.sessions[task_id]

# Usage
manager = OperatorSessionManager(proxy_ports=[10001, 10002, 10003, 10004, 10005])

# Get sticky session for multi-step task
proxy = manager.get_proxy_for_task("flight_booking_123", sticky=True)

# Complete task...

# Release session when done
manager.release_session("flight_booking_123")