The AI Research Analyst: 50 Websites Overnight, Report by Morning

# Create the orchestrator sandbox
export SANDBOX_API_KEY="ab0t_sk_live_YOUR_KEY_HERE"
export SANDBOX_URL="https://sandbox.dev.ab0t.com"

SANDBOX=$(curl -s -X POST "$SANDBOX_URL/api/sandboxes" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "research-orchestrator",
    "instance_type": "ab0t.medium",
    "auto_stop_minutes": 120
  }')

SANDBOX_ID=$(echo "$SANDBOX" | jq -r '.sandbox_id')
echo "Sandbox: $SANDBOX_ID"
echo "Status:  $(echo $SANDBOX | jq -r '.status')"
echo "IP:      $(echo $SANDBOX | jq -r '.instance_ip')"
echo "Cost:    $(echo $SANDBOX | jq -r '.hourly_cost')/hour"

# Poll until ready
while true; do
  STATUS=$(curl -s "$SANDBOX_URL/api/sandboxes/$SANDBOX_ID" \
    -H "Authorization: Bearer $SANDBOX_API_KEY" | jq -r '.status')
  echo "Status: $STATUS"
  [ "$STATUS" = "running" ] && break
  sleep 5
done

# Upload the research brief
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SANDBOX_ID/files" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "filename": "research_brief.md",
    "content": "# Competitive Analysis: Cloud IDE Market\n\nResearch the following companies. For each, extract:\n\n1. **Pricing** — tiers, per-seat cost, free tier limits\n2. **Key features** — what they emphasize on their homepage and product page\n3. **Target audience** — enterprise, startup, indie developer, student\n4. **Recent changes** — any pricing changes, new features, or pivots in the last 6 months\n5. **Integration ecosystem** — what tools, languages, and platforms they support\n\nFor each finding, record the exact URL where the information was found.\n\nOutput format: JSON with one object per company."
  }'

# Upload the URL list
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SANDBOX_ID/files" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "filename": "urls.json",
    "content": "[\n  {\"company\": \"Replit\", \"urls\": [\"https://replit.com\", \"https://replit.com/pricing\"]},\n  {\"company\": \"Gitpod\", \"urls\": [\"https://gitpod.io\", \"https://gitpod.io/pricing\"]},\n  {\"company\": \"GitHub Codespaces\", \"urls\": [\"https://github.com/features/codespaces\", \"https://docs.github.com/en/billing/managing-billing-for-github-codespaces\"]},\n  {\"company\": \"CodeSandbox\", \"urls\": [\"https://codesandbox.io\", \"https://codesandbox.io/pricing\"]},\n  {\"company\": \"StackBlitz\", \"urls\": [\"https://stackblitz.com\", \"https://stackblitz.com/pricing\"]}\n]"
  }'

import asyncio
import json
import os
import time
from datetime import datetime
import httpx

# Configuration
SANDBOX_URL = os.environ["SANDBOX_URL"]
API_KEY = os.environ["SANDBOX_API_KEY"]
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}


async def create_browser(client: httpx.AsyncClient) -> dict:
    """Create a Chrome browser container with CDP enabled."""
    resp = await client.post(
        f"{SANDBOX_URL}/api/browsers",
        headers=HEADERS,
        json={
            "browser_type": "chrome",
            "enable_debug_port": True,
            "idle_timeout_minutes": 15,
        },
    )
    resp.raise_for_status()
    return resp.json()


async def wait_for_container(client, container_id: str, timeout=120) -> dict:
    """Poll container status until assigned or timeout."""
    start = time.time()
    while time.time() - start < timeout:
        resp = await client.get(
            f"{SANDBOX_URL}/api/containers/{container_id}/status",
            headers=HEADERS,
        )
        data = resp.json()
        if data["state"] == "assigned":
            return data
        if data["state"] == "failed":
            raise RuntimeError(f"Container {container_id} failed")
        await asyncio.sleep(3)
    raise TimeoutError(f"Container {container_id} not ready after {timeout}s")


async def release_container(client, container_id: str):
    """Release container back to the warm pool."""
    try:
        await client.post(
            f"{SANDBOX_URL}/api/containers/{container_id}/release",
            headers=HEADERS,
        )
    except Exception:
        pass  # Best effort — cleanup loop handles stuck containers


async def research_company(client, company: dict, brief: str) -> dict:
    """
    Research a single company:
    1. Create a browser container
    2. Navigate to each URL, extract page content
    3. Return structured findings
    """
    company_name = company["company"]
    urls = company["urls"]
    print(f"[{company_name}] Starting research ({len(urls)} URLs)...")

    # Create browser
    browser = await create_browser(client)
    container_id = browser["container_id"]

    try:
        # Wait for browser to be ready
        status = await wait_for_container(client, container_id)
        cdp_url = status.get("cdp_url")
        print(f"[{company_name}] Browser ready: {container_id}")

        # Visit each URL and extract content
        page_data = []
        for url in urls:
            print(f"[{company_name}] Visiting: {url}")
            # Use CDP to navigate and extract
            # In production, use Playwright or Puppeteer connected to cdp_url
            # Here we show the extraction pattern:
            page_data.append({
                "url": url,
                "visited_at": datetime.utcnow().isoformat(),
                "content": "[extracted via CDP — see CDP section below]",
            })

        return {
            "company": company_name,
            "status": "success",
            "pages_visited": len(page_data),
            "pages": page_data,
            "container_id": container_id,
        }

    except Exception as e:
        print(f"[{company_name}] Error: {e}")
        return {
            "company": company_name,
            "status": "error",
            "error": str(e),
            "container_id": container_id,
        }

    finally:
        # Always release the browser back to the pool
        await release_container(client, container_id)
        print(f"[{company_name}] Browser released.")


async def main():
    # Load inputs
    with open("urls.json") as f:
        companies = json.load(f)
    with open("research_brief.md") as f:
        brief = f.read()

    print(f"Researching {len(companies)} companies...")
    start_time = time.time()

    # Fan out: research all companies in parallel
    async with httpx.AsyncClient(timeout=300) as client:
        tasks = [research_company(client, c, brief) for c in companies]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    # Fan in: collect results
    successes = [r for r in results if isinstance(r, dict) and r["status"] == "success"]
    failures = [r for r in results if isinstance(r, dict) and r["status"] == "error"]
    exceptions = [r for r in results if isinstance(r, Exception)]

    elapsed = time.time() - start_time

    # Save raw results
    with open("results_raw.json", "w") as f:
        json.dump(successes, f, indent=2)

    # Compile the report
    report = compile_report(successes, failures, brief, elapsed)
    with open("report.md", "w") as f:
        f.write(report)

    print(f"\n{'='*60}")
    print(f"Research complete in {elapsed:.0f}s")
    print(f"  Succeeded: {len(successes)}/{len(companies)}")
    print(f"  Failed:    {len(failures)}")
    print(f"  Errors:    {len(exceptions)}")
    print(f"  Report:    /workspace/report.md")
    print(f"{'='*60}")


def compile_report(results, failures, brief, elapsed) -> str:
    """Compile extraction results into a Markdown report."""
    lines = [
        f"# Competitive Analysis Report",
        f"",
        f"Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}",
        f"Companies analyzed: {len(results)}",
        f"Total time: {elapsed:.0f} seconds",
        f"",
        f"---",
        f"",
    ]

    for result in results:
        lines.append(f"## {result['company']}")
        lines.append(f"")
        lines.append(f"Pages visited: {result['pages_visited']}")
        lines.append(f"")
        for page in result["pages"]:
            lines.append(f"- [{page['url']}]({page['url']}) (visited {page['visited_at']})")
        lines.append(f"")
        # In production, include the structured extraction here:
        # pricing, features, target audience, recent changes, etc.
        lines.append(f"")

    if failures:
        lines.append(f"## Failures")
        lines.append(f"")
        for f in failures:
            lines.append(f"- **{f['company']}**: {f['error']}")
        lines.append(f"")

    return "\n".join(lines)


if __name__ == "__main__":
    asyncio.run(main())

# Upload the orchestrator script
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SANDBOX_ID/files" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d @- << 'EOF'
{
  "filename": "orchestrator.py",
  "content": "... (the script above)"
}
EOF

# Install dependencies and run
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SANDBOX_ID/execute" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "command": "cd /workspace && pip install httpx && SANDBOX_URL=https://sandbox.dev.ab0t.com SANDBOX_API_KEY=ab0t_sk_live_YOUR_KEY python orchestrator.py"
  }' | jq .

# Response:
# {
#   "stdout": "Researching 50 companies...\n[Replit] Starting research (2 URLs)...\n[Gitpod] Starting research (2 URLs)...\n...\n============================================================\nResearch complete in 847s\n  Succeeded: 48/50\n  Failed:    2\n  Report:    /workspace/report.md\n============================================================",
#   "stderr": "",
#   "exit_code": 0,
#   "execution_time_ms": 854230
# }

# Download the entire workspace (report + raw data)
curl -s "$SANDBOX_URL/api/sandboxes/$SANDBOX_ID/download" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -o research-output.tar.gz

tar xzf research-output.tar.gz
cat workspace/report.md

from playwright.async_api import async_playwright

async def extract_page(cdp_url: str, url: str) -> dict:
    """Navigate to a URL and extract structured data."""
    async with async_playwright() as p:
        # Connect to the remote browser via CDP
        browser = await p.chromium.connect_over_cdp(cdp_url)
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else await context.new_page()

        # Navigate to the target URL
        await page.goto(url, wait_until="networkidle", timeout=30000)

        # Wait for dynamic content to render
        await page.wait_for_timeout(2000)

        # Extract the full page text (rendered, not source HTML)
        title = await page.title()
        text_content = await page.evaluate("() => document.body.innerText")
        meta_desc = await page.evaluate("""() => {
            const el = document.querySelector('meta[name=\"description\"]');
            return el ? el.content : '';
        }""")

        # Extract all links for follow-up
        links = await page.evaluate("""() => {
            return Array.from(document.querySelectorAll('a[href]'))
                .map(a => ({text: a.innerText.trim(), href: a.href}))
                .filter(l => l.text && l.href.startsWith('http'))
                .slice(0, 50);
        }""")

        # Take a screenshot for visual reference
        screenshot = await page.screenshot(full_page=False)

        return {
            "url": url,
            "title": title,
            "meta_description": meta_desc,
            "text_content": text_content[:10000],  # Cap at 10k chars
            "links": links,
            "screenshot_bytes": len(screenshot),
        }

# Dismiss cookie banners — try common selectors
cookie_selectors = [
    "button:has-text('Accept')",
    "button:has-text('Accept all')",
    "button:has-text('I agree')",
    "button:has-text('Got it')",
    "[id*='cookie'] button",
    "[class*='cookie'] button",
    "[class*='consent'] button",
]
for selector in cookie_selectors:
    try:
        btn = page.locator(selector).first
        if await btn.is_visible(timeout=2000):
            await btn.click()
            break
    except:
        continue

# Scroll to load all content (pricing tables, feature lists)
previous_height = 0
for _ in range(10):  # Max 10 scroll iterations
    current_height = await page.evaluate("document.body.scrollHeight")
    if current_height == previous_height:
        break
    await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
    await page.wait_for_timeout(1500)
    previous_height = current_height

# Extract structured pricing data from pricing pages
pricing_data = await page.evaluate("""() => {
    // Look for common pricing card patterns
    const cards = document.querySelectorAll(
        '[class*="pricing"], [class*="plan"], [data-plan]'
    );
    return Array.from(cards).map(card => ({
        name: card.querySelector('h2, h3, [class*="name"]')?.innerText || '',
        price: card.querySelector('[class*="price"], [class*="amount"]')?.innerText || '',
        features: Array.from(card.querySelectorAll('li, [class*="feature"]'))
            .map(f => f.innerText.trim())
            .filter(Boolean),
    }));
}""")

import anthropic
import base64

async def extract_with_vision(page, extraction_prompt: str) -> str:
    """Screenshot the page and use Claude's vision to extract data."""
    screenshot = await page.screenshot(full_page=True)
    b64 = base64.b64encode(screenshot).decode()

    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": b64,
                    },
                },
                {
                    "type": "text",
                    "text": extraction_prompt,
                },
            ],
        }],
    )
    return response.content[0].text

import asyncio

# Limit to 50 concurrent browser containers
CONCURRENCY = 50
semaphore = asyncio.Semaphore(CONCURRENCY)

async def research_with_limit(client, company, brief):
    async with semaphore:
        return await research_company(client, company, brief)

# Fan out with concurrency limit
tasks = [research_with_limit(client, c, brief) for c in companies]
results = await asyncio.gather(*tasks, return_exceptions=True)

# asyncio.gather with return_exceptions=True
# This ensures one failure doesn't cancel the other 49 tasks
results = await asyncio.gather(*tasks, return_exceptions=True)

# Separate successes from failures
successes = []
failures = []
for r in results:
    if isinstance(r, Exception):
        failures.append({"error": str(r), "type": type(r).__name__})
    elif r.get("status") == "error":
        failures.append(r)
    else:
        successes.append(r)

print(f"Succeeded: {len(successes)}, Failed: {len(failures)}")
# Continue with successes — don't throw away good data

async def research_with_retry(client, company, brief, max_retries=2):
    """Retry transient failures with exponential backoff."""
    for attempt in range(max_retries + 1):
        result = await research_company(client, company, brief)
        if result["status"] == "success":
            return result
        if attempt < max_retries:
            wait = 2 ** attempt * 5  # 5s, 10s
            print(f"[{company['company']}] Retry {attempt+1} in {wait}s...")
            await asyncio.sleep(wait)
    return result  # Return last attempt even if failed

except Exception as e:
    # Take a failure screenshot before releasing the browser
    try:
        screenshot = await page.screenshot()
        with open(f"failures/{company_name}_error.png", "wb") as f:
            f.write(screenshot)
    except:
        pass
    return {"company": company_name, "status": "error", "error": str(e)}

# Competitive Analysis: Project Management SaaS

## Objective
Map the competitive landscape of project management tools targeting
teams of 10-200 people. Focus on pricing leverage points, feature
gaps, and positioning opportunities.

## Per-company extraction schema

For each company, extract the following into structured JSON:

### Pricing
- Free tier: exists? user limit? feature limits?
- Lowest paid tier: name, price/user/month, key features
- Enterprise tier: listed price or "contact sales"?
- Annual vs monthly pricing delta
- Any usage-based pricing components

### Product
- Primary value proposition (first heading on homepage)
- Key features listed on product/features page
- Integrations page: which tools do they integrate with?
- AI features: do they mention AI? what specifically?
- Mobile apps: iOS? Android? Both?

### Positioning
- Target company size (from messaging)
- Industry focus (vertical or horizontal?)
- Competitor comparisons on their site (vs pages)
- Customer logos displayed
- Testimonial quotes (first 3)

### Recent changes
- Blog: any product announcements in last 90 days?
- Pricing page: any "new" or "coming soon" badges?
- Changelog: last entry date?

### Metadata
- Exact URL for each data point
- Page title
- Timestamp of visit

{
  "company": "Linear",
  "research_date": "2026-03-27T06:15:00Z",
  "pages_visited": 4,
  "pricing": {
    "free_tier": {
      "exists": true,
      "user_limit": "Up to 250 issues",
      "feature_limits": "No roadmaps, limited integrations"
    },
    "lowest_paid": {
      "name": "Standard",
      "price_per_user_month": 8,
      "billing": "monthly",
      "key_features": ["Unlimited issues", "Roadmaps", "Custom fields"]
    },
    "enterprise": {
      "price": "Contact sales",
      "extras": ["SAML SSO", "SCIM", "Audit log"]
    },
    "annual_discount_pct": 17,
    "source_url": "https://linear.app/pricing"
  },
  "product": {
    "value_proposition": "Linear is a purpose-built tool for planning and building products",
    "key_features": [
      "Issue tracking", "Project management", "Roadmaps",
      "Cycles", "Triage", "Git integration"
    ],
    "ai_features": [
      "AI-powered issue creation from Slack",
      "Auto-categorization of bug reports",
      "Natural language project search"
    ],
    "integrations": ["GitHub", "GitLab", "Slack", "Figma", "Sentry", "Zendesk"],
    "mobile": {"ios": true, "android": true},
    "source_url": "https://linear.app/features"
  },
  "positioning": {
    "target_size": "Startups and scaling teams",
    "vertical": "Horizontal (software companies)",
    "competitor_comparisons": ["vs Jira", "vs Asana", "vs Shortcut"],
    "customer_logos": ["Vercel", "Ramp", "Loom", "Retool", "PostHog"],
    "source_url": "https://linear.app"
  },
  "recent_changes": {
    "latest_blog_post": {
      "title": "Introducing Linear Asks",
      "date": "2026-03-15",
      "source_url": "https://linear.app/blog/introducing-linear-asks"
    },
    "source_url": "https://linear.app/changelog"
  }
}

import anthropic

def compile_with_claude(results_json: str, brief: str) -> str:
    """Use Claude to synthesize raw data into a narrative report."""
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=8192,
        messages=[{
            "role": "user",
            "content": f"""You are a senior market analyst. Given the research brief
and the raw extraction data from {len(results)} companies, produce a
comprehensive competitive analysis report in Markdown.

Include:
1. Executive summary (3 paragraphs)
2. Pricing comparison table (all companies, key tiers)
3. Feature matrix (key features vs companies)
4. AI capabilities comparison
5. Positioning map analysis
6. Actionable recommendations (5 bullet points)
7. Appendix: per-company detail sheets

Research brief:
{brief}

Raw data:
{results_json}""",
        }],
    )

    return response.content[0].text

#!/bin/bash
# Weekly competitive research — runs every Monday at 6am UTC
set -euo pipefail

SANDBOX_URL="https://sandbox.dev.ab0t.com"
SANDBOX_API_KEY="$RESEARCH_AGENT_API_KEY"

# 1. Create sandbox
SANDBOX=$(curl -s -X POST "$SANDBOX_URL/api/sandboxes" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"name":"weekly-research","instance_type":"ab0t.medium","auto_stop_minutes":60}')
SID=$(echo "$SANDBOX" | jq -r '.sandbox_id')

# 2. Wait for boot
while [ "$(curl -s "$SANDBOX_URL/api/sandboxes/$SID" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" | jq -r '.status')" != "running" ]; do
  sleep 5
done

# 3. Upload scripts and run
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SID/execute" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"command\": \"cd /workspace && git clone https://github.com/your-org/research-agent.git . && pip install -r requirements.txt && python orchestrator.py\"}"

# 4. Download report and post to Slack
curl -s "$SANDBOX_URL/api/sandboxes/$SID/download" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -o /tmp/research.tar.gz
tar xzf /tmp/research.tar.gz -C /tmp/research
REPORT=$(cat /tmp/research/workspace/report.md)

curl -X POST "$SLACK_WEBHOOK_URL" \
  -H "Content-Type: application/json" \
  -d "{\"text\": \"Weekly competitive research is ready.\n\n$(echo $REPORT | head -20)\"}"

# 5. Delete sandbox
curl -s -X DELETE "$SANDBOX_URL/api/sandboxes/$SID" \
  -H "Authorization: Bearer $SANDBOX_API_KEY"

echo "Done. Sandbox $SID deleted."

> I need a competitive analysis of these 10 companies: [list].
>
> For each one, use the Sandbox Platform API at $SANDBOX_URL to
> create a Chrome browser container, navigate to their website
> and pricing page, extract pricing tiers and key features, then
> release the browser.
>
> After all extractions are done, compile a Markdown report comparing
> all 10 companies in a pricing table and feature matrix. Save it
> to /workspace/competitive-analysis.md.
>
> Use my API key in $SANDBOX_API_KEY. Run all browsers in parallel
> using asyncio. Handle failures gracefully — if a site is down,
> note it and continue with the others.

import uuid

run_id = str(uuid.uuid4())[:8]

# Every container gets the run_id in metadata
browser = await client.post(f"{SANDBOX_URL}/api/browsers", json={
    "browser_type": "chrome",
    "enable_debug_port": True,
    "metadata": {
        "run_id": run_id,
        "task": "competitive-research",
        "company": company_name,
    },
})