Python 3 Examples

The following examples demonstrate how to handle session creation, long-polling, command execution and gracefully stopping the session.

1. Minimal API Client

This helper class handles API authentication, long-polling for session creation and command execution.

python
import time
import requests
import json

API_URL = "https://api.getbro.ws"
API_KEY = "your_api_key_here"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

class BroClient:
    def __init__(self):
        self.session_id = None

    def create_session(self, **kwargs):
        print("Creating session...")
        response = requests.post(f"{API_URL}/v1/sessions", json=kwargs, headers=HEADERS)
        print(f"Create Session Init Response: {response.text}")
        response.raise_for_status()
        self.session_id = response.json()["session_id"]
        
        # Poll until the session is ready
        while True:
            resp = requests.get(f"{API_URL}/v1/sessions/{self.session_id}", headers=HEADERS)
            print(f"Session Poll Response: {resp.text}")
            status = resp.json().get("status")
            if status == "idle":
                break
            elif status in ["failed", "cancelled"]:
                raise Exception(f"Failed to create session. Status: {status}")
            time.sleep(1)
            
        print(f"Session {self.session_id} is ready!")
        return self.session_id

    def execute(self, commands):
        print("Executing command batch...")
        payload = {"commands": commands}
        resp = requests.post(
            f"{API_URL}/v1/sessions/{self.session_id}/execute", 
            json=payload, 
            headers=HEADERS
        )
        print(f"Execute Init Response: {resp.text}")
        resp.raise_for_status()
        command_id = resp.json()["command_id"]
        
        # Poll command execution status
        while True:
            c_resp = requests.get(
                f"{API_URL}/v1/sessions/{self.session_id}/commands/{command_id}", 
                headers=HEADERS
            )
            print(f"Command Poll Response: {c_resp.text}")
            c_data = c_resp.json()
            status = c_data.get("status")
            if status in ["done", "failed"]:
                return c_data
            time.sleep(2)

    def stop_session(self):
        if self.session_id:
            resp = requests.delete(f"{API_URL}/v1/sessions/{self.session_id}", headers=HEADERS)
            print(f"Delete Session Response: {resp.text}")
            print("Session stopped.")
            
            # Wait for backend to finalize costs
            time.sleep(2)
            stats_resp = requests.get(f"{API_URL}/v1/sessions/{self.session_id}", headers=HEADERS)
            print(f"Final Session Stats:\n{json.dumps(stats_resp.json(), indent=2)}")
            
            self.session_id = None

2. Manual Commands & Scraping

This example shows how to open a URL, interact with the page manually and retrieve HTML or take a screenshot.

python
client = BroClient()

try:
    # 1. Create a session with residential proxy enabled
    client.create_session(
        enable_proxy=True,
        proxy_tier="lite",
        proxy_policy="extended",
        country="US"
    )
    
    # 2. Execute a batch of manual commands
    result = client.execute([
        {"command": "open_url", "params": {"url": "https://news.ycombinator.com/"}},
        {"command": "get_html"}
    ])
    
    if result["status"] == "done":
        # Check if the payload was offloaded
        html_step = result["response"]["commands"][2] # get_html is at index 2
        
        if html_step.get("offloaded_data_url"):
            print(f"HTML is large, offloaded to: {html_step['offloaded_data_url']}")
        else:
            print("HTML retrieved inline successfully.")
            
finally:
    client.stop_session()

3. Autopilot act (Autonomous Actions + Extraction)

Use the act command to let the AI navigate complex pages and perform goals on your behalf. You can also extract data at the end of the action.

python
import json

client = BroClient()

try:
    client.create_session()
    
    # Search for laptops and extract info
    result = client.execute([
        {"command": "open_url", "params": {"url": "https://www.amazon.com/"}},
        {
            "command": "act",
            "params": {
                "instruction": "Search for 'gaming laptops' and click on the first organic result.",
                "max_steps": 10,
                "model_size": "medium",
                "extract_data": True,
                "data_instruction": "Extract specifications of the laptop",
                "json_schema": {
                    "title": "<product title>",
                    "price": "<price>",
                    "cpu": "<processor type>",
                    "gpu": "<graphics card>"
                }
            }
        }
    ])
    
    if result["status"] == "done":
        act_step = result["response"]["commands"][1]
        extracted_data = act_step["data"].get("extracted_json")
        print("Extracted Data:", json.dumps(extracted_data, indent=2))

finally:
    client.stop_session()

4. Autopilot extract (Static Data Extraction & Pagination)

If you already have a target URL, you can extract structured data directly, including automatically clicking "Next" to traverse multiple pages.

python
import json

client = BroClient()

try:
    client.create_session()
    
    result = client.execute([
        {"command": "open_url", "params": {"url": "https://news.ycombinator.com/"}},
        {
            "command": "extract",
            "params": {
                "data_instruction": "Extract a list of top publications on this page",
                "json_schema": [{
                    "title": "<post title>",
                    "link": "<post URL>",
                    "points": "<number of points>"
                }],
                "model_size": "small",
                "paginate": True,
                "pages_to_paginate": 2
            }
        }
    ])
    
    if result["status"] == "done":
        extract_step = result["response"]["commands"][1]
        data = extract_step["data"].get("extracted_json")
        print("Extracted Data:", json.dumps(data, indent=2))
        
finally:
    client.stop_session()