Code Examples

Real-world automation scenarios with full working code.

Quick Data Extraction Beginner

Navigate to Hacker News and extract the top story titles and point counts.

import asyncio, httpx, os

API_KEY = os.environ["BABELWRAP_API_KEY"]
BASE = "https://api.babelwrap.com/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

async def main():
    async with httpx.AsyncClient(timeout=30.0) as c:
        sid = (
            await c.post(f"{BASE}/sessions", headers=HEADERS, json={})
        ).json()["session_id"]

        try:
            await c.post(f"{BASE}/sessions/{sid}/navigate", headers=HEADERS,
                json={"url": "https://news.ycombinator.com"})

            resp = await c.post(f"{BASE}/sessions/{sid}/extract", headers=HEADERS,
                json={"query": "all story titles and their point counts"})

            for s in resp.json()["data"]:
                print(f"  [{s.get('points', '?')}] {s['title']}")
        finally:
            await c.delete(f"{BASE}/sessions/{sid}", headers=HEADERS)

asyncio.run(main())
const API_KEY = process.env.BABELWRAP_API_KEY;
const BASE = "https://api.babelwrap.com/v1";
const h = { "Authorization": `Bearer ${API_KEY}`, "Content-Type": "application/json" };

const { session_id: sid } = await fetch(`${BASE}/sessions`, {
  method: "POST", headers: h, body: "{}"
}).then(r => r.json());

await fetch(`${BASE}/sessions/${sid}/navigate`, {
  method: "POST", headers: h,
  body: JSON.stringify({ url: "https://news.ycombinator.com" })
});

const { data } = await fetch(`${BASE}/sessions/${sid}/extract`, {
  method: "POST", headers: h,
  body: JSON.stringify({ query: "all story titles and their point counts" })
}).then(r => r.json());

data.forEach(s => console.log(`  [${s.points}] ${s.title}`));
await fetch(`${BASE}/sessions/${sid}`, { method: "DELETE", headers: h });

Login to a Website Beginner

Automate a complete login flow: navigate, fill credentials, submit, and verify.

async def login_and_extract(client, email, password):
    # Create session
    sid = (await client.post(f"{BASE}/sessions", headers=HEADERS, json={})).json()["session_id"]

    try:
        # Navigate to login
        await client.post(f"{BASE}/sessions/{sid}/navigate", headers=HEADERS,
            json={"url": "https://app.example.com/login"})

        # Fill credentials
        await client.post(f"{BASE}/sessions/{sid}/fill", headers=HEADERS,
            json={"target": "Email address", "value": email})
        await client.post(f"{BASE}/sessions/{sid}/fill", headers=HEADERS,
            json={"target": "Password", "value": password})

        # Submit login form
        resp = await client.post(f"{BASE}/sessions/{sid}/click", headers=HEADERS,
            json={"target": "Sign In button"})
        snap = resp.json()["snapshot"]

        # Verify login succeeded
        if snap["alerts"]:
            raise Exception(f"Login failed: {snap['alerts'][0]['text']}")

        # Now extract data from the authenticated page
        resp = await client.post(f"{BASE}/sessions/{sid}/extract", headers=HEADERS,
            json={"query": "user name and account status"})
        return resp.json()["data"]
    finally:
        await client.delete(f"{BASE}/sessions/{sid}", headers=HEADERS)

Product Price Monitor Intermediate

Monitor prices across multiple product pages and track changes.

import asyncio, httpx, json, os
from datetime import datetime

PRODUCTS = [
    "https://shop.example.com/product/wireless-headphones",
    "https://shop.example.com/product/mechanical-keyboard",
    "https://shop.example.com/product/usb-c-hub",
]

async def check_prices():
    async with httpx.AsyncClient(timeout=30.0) as client:
        sid = (await client.post(
            f"{BASE}/sessions", headers=HEADERS, json={}
        )).json()["session_id"]

        prices = []
        try:
            for url in PRODUCTS:
                await client.post(f"{BASE}/sessions/{sid}/navigate",
                    headers=HEADERS, json={"url": url})

                resp = await client.post(f"{BASE}/sessions/{sid}/extract",
                    headers=HEADERS,
                    json={"query": "product name, current price, and availability"})

                data = resp.json()["data"]
                prices.append({
                    "url": url,
                    "product": data.get("name", "Unknown"),
                    "price": data.get("price"),
                    "available": data.get("available", True),
                    "checked_at": datetime.utcnow().isoformat(),
                })
        finally:
            await client.delete(f"{BASE}/sessions/{sid}", headers=HEADERS)

        return prices

# Run and save results
prices = asyncio.run(check_prices())
with open("prices.json", "w") as f:
    json.dump(prices, f, indent=2)
print(f"Checked {len(prices)} products")

Contact Form Submission Intermediate

Fill and submit a multi-field contact form, then verify the submission.

async def submit_contact_form(client, sid, contact_info):
    # Navigate to contact page
    await client.post(f"{BASE}/sessions/{sid}/navigate", headers=HEADERS,
        json={"url": "https://company.example.com/contact"})

    # Fill all form fields
    fields = [
        ("First name", contact_info["first_name"]),
        ("Last name", contact_info["last_name"]),
        ("Email address", contact_info["email"]),
        ("Phone number", contact_info["phone"]),
        ("Company name", contact_info["company"]),
        ("Message", contact_info["message"]),
    ]

    for target, value in fields:
        await client.post(f"{BASE}/sessions/{sid}/fill", headers=HEADERS,
            json={"target": target, "value": value})

    # Select a dropdown value
    await client.post(f"{BASE}/sessions/{sid}/fill", headers=HEADERS,
        json={"target": "Inquiry type dropdown", "value": "Partnership"})

    # Submit the form
    resp = await client.post(f"{BASE}/sessions/{sid}/submit", headers=HEADERS, json={})
    snap = resp.json()["snapshot"]

    # Check for success
    success_alerts = [a for a in snap["alerts"] if a["type"] == "success"]
    if success_alerts:
        print(f"Submitted: {success_alerts[0]['text']}")
    else:
        # Check for validation errors
        errors = [a for a in snap["alerts"] if a["type"] == "error"]
        print(f"Errors: {[e['text'] for e in errors]}")
    return snap

E-Commerce Checkout Advanced

Complete a multi-step e-commerce flow: product selection, cart, shipping, and order confirmation.

async def full_checkout(client, product_url, shipping, payment):
    sid = (await client.post(
        f"{BASE}/sessions", headers=HEADERS, json={}
    )).json()["session_id"]

    try:
        # Step 1: Go to product page
        await client.post(f"{BASE}/sessions/{sid}/navigate",
            headers=HEADERS, json={"url": product_url})

        # Step 2: Add to cart
        resp = await client.post(f"{BASE}/sessions/{sid}/click",
            headers=HEADERS, json={"target": "Add to Cart"})
        snap = resp.json()["snapshot"]

        # Check for out-of-stock
        if any(a["type"] == "error" for a in snap["alerts"]):
            return {"success": False, "error": "Product unavailable"}

        # Step 3: Go to checkout
        await client.post(f"{BASE}/sessions/{sid}/click",
            headers=HEADERS, json={"target": "Proceed to Checkout"})

        # Step 4: Fill shipping info
        for field, value in shipping.items():
            await client.post(f"{BASE}/sessions/{sid}/fill",
                headers=HEADERS, json={"target": field, "value": value})

        await client.post(f"{BASE}/sessions/{sid}/click",
            headers=HEADERS, json={"target": "Continue to Payment"})

        # Step 5: Extract order summary
        resp = await client.post(f"{BASE}/sessions/{sid}/extract",
            headers=HEADERS,
            json={"query": "order total, subtotal, shipping cost, and items"})

        return {"success": True, "order": resp.json()["data"]}
    finally:
        await client.delete(f"{BASE}/sessions/{sid}", headers=HEADERS)

# Usage
result = await full_checkout(
    client,
    product_url="https://shop.example.com/product/headphones",
    shipping={
        "First name": "Jane", "Last name": "Doe",
        "Address": "123 Main St", "City": "San Francisco",
        "State dropdown": "California", "ZIP code": "94102",
    },
    payment={},  # payment handled separately
)

MCP Agent Integration Intermediate

Use BabelWrap tools from a custom MCP client (not Claude Desktop).

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

server_params = StdioServerParameters(
    command="babelwrap-mcp",
    env={"BABELWRAP_API_KEY": "bw_your_api_key"},
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()

        # List available tools
        tools = await session.list_tools()
        print(f"Tools: {[t.name for t in tools.tools]}")

        # Create a browser session
        result = await session.call_tool(
            "babelwrap_new_session",
            arguments={"metadata": {"purpose": "research"}},
        )
        sid = result.content[0].text  # parse session_id from response

        # Navigate and extract
        await session.call_tool(
            "babelwrap_navigate",
            arguments={"session_id": sid, "url": "https://example.com"},
        )

        result = await session.call_tool(
            "babelwrap_extract",
            arguments={"session_id": sid, "query": "main heading and description"},
        )
        print(result.content[0].text)

        # Clean up
        await session.call_tool(
            "babelwrap_close_session",
            arguments={"session_id": sid},
        )
Need help? See the Quickstart Guide for initial setup, or the API Reference for complete endpoint documentation.