Claude Computer Use Proxy Setup Guide

import os
import sys
import base64
import subprocess
from anthropic import Anthropic

# SDK automatically reads HTTP_PROXY / HTTPS_PROXY from env.
# Since the Dockerfile sets them to the Coronium endpoint,
# EVERY outbound call (including anthropic.com) goes through the mobile proxy.
# To force Anthropic API calls to go direct, set no_proxy=api.anthropic.com
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

MODEL = "claude-3-5-sonnet-20241022"
TOOL_VERSION = "computer_20250124"
WIDTH, HEIGHT = 1280, 800
MAX_ITERATIONS = 40

def screenshot() -> str:
    """Capture the Xvfb display and return base64-encoded PNG."""
    subprocess.run(["scrot", "-z", "/tmp/shot.png"], check=True)
    with open("/tmp/shot.png", "rb") as f:
        return base64.standard_b64encode(f.read()).decode()

def execute_action(action: dict) -> dict:
    """Translate a Claude tool_use input into xdotool commands."""
    a = action["action"]
    if a == "screenshot":
        pass  # fall through to screenshot below
    elif a == "left_click":
        x, y = action["coordinate"]
        subprocess.run(["xdotool", "mousemove", str(x), str(y), "click", "1"], check=True)
    elif a == "type":
        subprocess.run(["xdotool", "type", "--delay", "25", action["text"]], check=True)
    elif a == "key":
        subprocess.run(["xdotool", "key", action["text"]], check=True)
    elif a == "scroll":
        direction = action.get("scroll_direction", "down")
        amount = action.get("scroll_amount", 3)
        button = "5" if direction == "down" else "4"
        for _ in range(amount):
            subprocess.run(["xdotool", "click", button], check=True)
    # return the new screenshot
    return {
        "type": "image",
        "source": {"type": "base64", "media_type": "image/png", "data": screenshot()},
    }

def run_agent(goal: str):
    messages = [{"role": "user", "content": goal}]
    tools = [{
        "type": TOOL_VERSION,
        "name": "computer",
        "display_width_px": WIDTH,
        "display_height_px": HEIGHT,
        "display_number": 1,
    }]

    for iteration in range(MAX_ITERATIONS):
        resp = client.beta.messages.create(
            model=MODEL,
            max_tokens=4096,
            tools=tools,
            messages=messages,
            betas=["computer-use-2025-01-24"],
        )

        # Save assistant response to transcript
        messages.append({"role": "assistant", "content": resp.content})

        if resp.stop_reason == "end_turn":
            print(f"[agent] task complete after {iteration + 1} iterations")
            return

        # Execute each tool_use block and collect tool_results
        tool_results = []
        for block in resp.content:
            if block.type == "tool_use":
                print(f"[agent] action: {block.input.get('action')} {block.input}")
                result = execute_action(block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": [result],
                })

        if not tool_results:
            print("[agent] no tool_use emitted but stop_reason != end_turn; stopping")
            return

        messages.append({"role": "user", "content": tool_results})

if __name__ == "__main__":
    run_agent(" ".join(sys.argv[1:]))

import os
import httpx
from anthropic import Anthropic

CORONIUM_PROXY = os.environ["CORONIUM_PROXY_URL"]
# e.g. http://user:pass@proxy.coronium.io:30000

# Anthropic API calls: direct (low latency, billing dashboard requires real IP)
api_client = httpx.Client(timeout=60)
anthropic = Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    http_client=api_client,
)

# Target website calls: through Coronium mobile proxy
# Used by tools spawned inside the sandbox (curl, requests, Firefox, etc.)
# Enforced via env vars set in the Dockerfile
# NOTE: Xvfb's Firefox automatically picks up HTTP_PROXY / HTTPS_PROXY

# For Python scripts that scrape DURING the agent task, use httpx explicitly:
proxy_client = httpx.Client(
    proxies={"http://": CORONIUM_PROXY, "https://": CORONIUM_PROXY},
    timeout=60,
    follow_redirects=True,
)

# Verify the mobile IP
r = proxy_client.get("https://ifconfig.me")
print(f"Mobile egress IP: {r.text.strip()}")
# -> should print a T-Mobile / Verizon / Vodafone IP, not your host IP

while True:
    resp = client.beta.messages.create(
        model=MODEL,
        max_tokens=4096,
        tools=tools,
        messages=messages,
        betas=["computer-use-2025-01-24"],
    )
    messages.append({"role": "assistant", "content": resp.content})

    if resp.stop_reason == "end_turn":
        break

    if resp.stop_reason == "pause_turn":
        # Claude is thinking. Re-send the conversation unchanged.
        # No new user message is added.
        continue

    if resp.stop_reason == "tool_use":
        tool_results = execute_all_tool_uses(resp.content)
        messages.append({"role": "user", "content": tool_results})
        continue

    if resp.stop_reason == "max_tokens":
        # Output truncated. Raise max_tokens or summarize and continue.
        raise RuntimeError("Agent hit max_tokens mid-response")

    if resp.stop_reason == "refusal":
        # Model refused (safety classifier). Log and abort.
        raise RuntimeError(f"Claude refused: {resp.content}")