Cookbook

from browsa import Client

c = Client(api_key="agt_live_...")

# 1. Provision a persistent identity (cookies + fingerprint persist 30d)
agent = c.create_agent(mode="persistent", country="us", persona="us-shopper")
print(f"trust_score: {agent['trust_score']}/1000")
# → trust_score: 893/1000

# 2. Why is it 893? See the breakdown.
tb = c.trust_breakdown(agent["id"])
print(tb["band"], "—", len(tb["factors"]), "factors")
for f in tb["factors"]:
    if f["delta"] != 0:
        print(f"  {f['label']}: {f['delta']:+}")

# 3. What fingerprint will it present?
fp = c.fingerprint(agent["id"])
print(fp["navigator"]["user_agent"])
print(fp["webgl"]["unmasked_renderer"])

# 4. Run a task against it.
job = c.run_task(
    task="Go to news.ycombinator.com and return the top 3 titles as JSON.",
    llm="claude-sonnet-4-6",
    llm_api_key="sk-ant-...",
    agent_id=agent["id"],   # reuse the persistent identity
)
print(job.final_result)

import { Client } from "browsa";

const c = new Client({ apiKey: "agt_live_..." });

// 1. Provision a persistent identity (cookies + fingerprint persist 30d)
const agent = await c.createAgent({
  mode: "persistent", country: "us", persona: "us-shopper",
});
console.log(`trust_score: ${agent.trust_score}/1000`);

// 2. Why is it that score? See the breakdown.
const tb = await c.trustBreakdown(agent.id);
console.log(tb.band, "—", tb.factors.length, "factors");
for (const f of tb.factors) {
  if (f.delta !== 0) console.log(`  ${f.label}: ${f.delta > 0 ? "+" : ""}${f.delta}`);
}

// 3. What fingerprint will it present?
const fp = await c.fingerprint(agent.id);
console.log(fp.navigator.user_agent);
console.log(fp.webgl.unmasked_renderer);

// 4. Run a task against it.
const job = await c.runTask({
  task: "Go to news.ycombinator.com and return the top 3 titles as JSON.",
  llm: "claude-sonnet-4-6",
  llmApiKey: "sk-ant-...",
  agentId: agent.id,    // reuse the persistent identity
});
console.log(job.final_result);

from browsa import Client
c = Client(api_key="agt_live_...")

agent = c.create_agent(country="us", ttl="10m")

# framework = playwright_py | playwright_ts | puppeteer_js | curl
out = c.connect(agent["id"], framework="playwright_py")
print(out["snippet"])
# → import asyncio
#   from playwright.async_api import async_playwright
#   CDP_URL = "https://browsa.io/api/v1/cdp/<profile>"
#   ...

# Or just grab the URL directly:
print(out["cdp_http"])   # https://browsa.io/api/v1/cdp/<profile>
print(out["cdp_url"])    # wss://... (first page)

import uuid
from browsa import Client

c = Client(api_key="agt_live_...")

key = str(uuid.uuid4())  # generate once per logical request

# Wrap your retry logic in the same idempotency_key — safe to retry as
# many times as your network layer needs.
try:
    res = c.scrape("https://news.ycombinator.com", idempotency_key=key)
except Exception:
    # Even on transient failure, retry with the SAME key:
    res = c.scrape("https://news.ycombinator.com", idempotency_key=key)

# Response includes a `http_status` field — the upstream HTTP code.
# Detect target-side failures without HTML scraping:
if res["http_status"] >= 400:
    raise RuntimeError(f"target returned {res['http_status']}")

job = c.run_task(
    task="Find the cheapest flight from SFO to JFK next Tuesday.",
    llm="claude-opus-4-7",
    llm_api_key="sk-ant-...",
    max_cost_usd=0.50,   # platform stops the task at $0.50 spend
    max_steps=20,
)
# The job's terminal response carries the full cost split:
print(job.total_cost_usd, "=", job.browser_cost_usd, "+", job.llm_cost_usd or 0)

# Register a webhook (returns whk_…)
hook = c.create_webhook(
    url="https://your-app.com/hooks/browsa",
    events=["job.completed", "job.failed"],
)

# Get the HMAC signing secret (one per workspace, deterministic)
secret = c.webhook_signing_secret()["secret"]  # hex string

# Fire a test ping to verify your handler works:
c.ping_webhook(hook["id"])

# See what was delivered (audit log):
deliveries = c.webhook_deliveries(hook["id"])
for d in deliveries["deliveries"][:5]:
    print(d["event_type"], d["status_code"], d["duration_ms"], "ms")

Verify the signature in your handler (FastAPI example):

import hashlib, hmac, os, time
from fastapi import FastAPI, Request, HTTPException

SECRET = bytes.fromhex(os.environ["BROWSA_WEBHOOK_SECRET"])
app = FastAPI()

@app.post("/hooks/browsa")
async def hook(req: Request):
    sig_header = req.headers["X-Agents-Signature"]
    # Format: "t=1700000000, v1=abc123..."
    parts = dict(p.strip().split("=", 1) for p in sig_header.split(","))
    ts = int(parts["t"])
    if abs(time.time() - ts) > 300:
        raise HTTPException(400, "stale signature")
    body = await req.body()
    expected = hmac.new(SECRET, f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(parts["v1"], expected):
        raise HTTPException(400, "bad signature")
    payload = await req.json()
    print("job", payload["data"]["job_id"], payload["type"])
    return {"ok": True}

cursor = None
all_jobs = []
while True:
    page = c.list_jobs(limit=100, cursor=cursor)
    all_jobs.extend(page["jobs"])
    if not page.get("has_more"):
        break
    cursor = page["next_cursor"]
print(f"loaded {len(all_jobs)} jobs total")

agent = c.create_agent(mode="persistent", country="de", os_type="OS_TYPE_MACOS", warmup_recipe="amazon")

tb = c.trust_breakdown(agent["id"])
assert tb["trust_score"] >= 850, f"identity below undetected band: {tb['trust_score']}"

fp = c.fingerprint(agent["id"])
assert fp["navigator"]["webdriver"] is False
assert fp["timezone"]["name"] == "Europe/Berlin"   # coherent with country=de
assert "Apple" in fp["webgl"]["unmasked_vendor"]    # macOS shape

print("✓ identity passes pre-production checks")

from browsa import Client, errors

c = Client(api_key="agt_live_...")
try:
    c.scrape("https://example.com", country="ZZ")  # bad country code
except errors.BadRequestError as e:
    print(e.code)        # BAD_REQUEST
    print(e.request_id)  # uuid — quote in support ticket
    print(e.body)        # raw envelope from the platform

from browsa import Client
from playwright.async_api import async_playwright
import asyncio

c = Client(api_key="agt_live_...")
agent = c.create_agent(country="us", ttl="5m")

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(agent["cdp_http"])
        page = browser.contexts[0].pages[0]
        await page.goto("https://news.ycombinator.com")
        titles = await page.eval_on_selector_all(
            ".titleline > a", "els => els.slice(0, 5).map(e => e.textContent)"
        )
        print(titles)
        await browser.close()
    c.destroy_agent(agent["id"])

asyncio.run(main())

job = c.create_task(
    task="Scrape product reviews from the URL list.",
    llm="claude-sonnet-4-6",
    llm_api_key="sk-ant-...",
)
# Block until terminal (or timeout). Reuses HTTP long-poll under the hood
# so we don't burn through your client's API budget polling.
final = c.wait_for_job(job.id, timeout=360, poll=10,
                       on_progress=lambda j: print(j.status, j.step_count))
if final.succeeded:
    print(final.final_result)
else:
    print(final.error_code, "—", final.error_message)