Parallel

{
  "ownerId": "kn7d7xy7794nh6aaabfga5wwzh7zptdm",
  "slug": "parallel",
  "version": "3.0.0",
  "publishedAt": 1769355659028
}

{
  "version": 1,
  "registry": "https://clawhub.ai",
  "slug": "parallel",
  "installedVersion": "1.0.1",
  "installedAt": 1770777480501
}

# Exclude binaries and dev artifacts from ClawHub bundle
node_modules/
*.png
*.jpg
*.jpeg
*.gif
*.mp3
*.mp4
*.zip
*.tar.gz
tests/
test/
__pycache__/
*.pyc
.DS_Store

.venv/
__pycache__/
*.pyc
.env

parallel-web>=1.0.0,<2.0.0
requests>=2.28.0,<3.0.0

#!/usr/bin/env python3
"""
Parallel.ai Extract API - Clean content extraction from any URL.

Usage:
  python3 extract.py https://stripe.com/docs/api  # Extract with excerpts
  python3 extract.py https://example.com/paper.pdf --full  # Full content
  python3 extract.py https://sec.gov/10-K.htm --objective "Extract risk factors"
"""

import os
import sys
import json
import argparse

from parallel import Parallel

API_KEY = os.environ.get("PARALLEL_API_KEY")
if not API_KEY:
    print("Error: PARALLEL_API_KEY environment variable is required", file=sys.stderr)
    sys.exit(1)


def extract(
    client: Parallel,
    urls: list,
    objective: str = None,
    full_content: bool = False,
) -> dict:
    """Extract content from URLs."""
    params = {
        "urls": urls,
    }
    
    if objective:
        params["objective"] = objective
    
    if full_content:
        params["full_content"] = {"enabled": True}
    
    result = client.beta.extract(**params)
    return result


def format_result(result) -> str:
    """Format extraction result for display."""
    output = []
    
    output.append(f"📄 Extract ID: {result.extract_id}")
    output.append("")
    
    for i, item in enumerate(result.results, 1):
        url = item.url
        title = getattr(item, 'title', 'No title')
        date = getattr(item, 'publish_date', None)
        
        date_str = f" ({date})" if date else ""
        output.append(f"**{i}. {title}**{date_str}")
        output.append(f"   URL: {url}")
        
        # Show excerpts or content
        excerpts = getattr(item, 'excerpts', None)
        content = getattr(item, 'content', None)
        
        if content:
            # Full content mode
            preview = content[:2000]
            if len(content) > 2000:
                preview += f"\n\n... [{len(content)} chars total]"
            output.append(f"\n{preview}")
        elif excerpts:
            # Excerpt mode
            output.append("")
            for excerpt in excerpts[:3]:
                excerpt_clean = excerpt.replace("\n", " ").strip()[:500]
                output.append(f"   > {excerpt_clean}")
        
        output.append("")
    
    return "\n".join(output)


def main():
    parser = argparse.ArgumentParser(description="Parallel.ai Extract API")
    parser.add_argument("urls", nargs="+", help="URLs to extract content from")
    parser.add_argument("--objective", "-o", metavar="TEXT",
                       help="Focus extraction on specific content (e.g., 'Extract API endpoints')")
    parser.add_argument("--full", "-f", action="store_true",
                       help="Return full page content instead of excerpts")
    parser.add_argument("--json", "-j", action="store_true",
                       help="Output raw JSON")
    
    args = parser.parse_args()
    
    client = Parallel(api_key=API_KEY)
    
    try:
        result = extract(
            client,
            urls=args.urls,
            objective=args.objective,
            full_content=args.full,
        )
        
        if args.json:
            output = {
                "extract_id": result.extract_id,
                "results": [
                    {
                        "url": r.url,
                        "title": getattr(r, 'title', None),
                        "publish_date": getattr(r, 'publish_date', None),
                        "excerpts": getattr(r, 'excerpts', None),
                        "content": getattr(r, 'content', None),
                    }
                    for r in result.results
                ]
            }
            print(json.dumps(output, indent=2, default=str))
        else:
            print(format_result(result))
            
    except Exception as e:
        print(f"❌ Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Parallel.ai FindAll API - Natural language → structured datasets.

Usage:
  python3 findall.py "Find all AI startups that raised Series A in the last 6 months"
  python3 findall.py "dental practices in Ohio with 4+ star Google reviews" --limit 50
  python3 findall.py "portfolio companies of Khosla Ventures" --enrich "funding,employee_count"
  python3 findall.py --status findall_abc123  # Check status of running job
"""

import os
import sys
import json
import argparse
import time

from parallel import Parallel

API_KEY = os.environ.get("PARALLEL_API_KEY")
if not API_KEY:
    print("Error: PARALLEL_API_KEY environment variable is required", file=sys.stderr)
    sys.exit(1)


def ingest_query(client: Parallel, query: str) -> dict:
    """Convert natural language to structured schema."""
    result = client.beta.findall.ingest(objective=query)
    return result


def create_findall(
    client: Parallel,
    objective: str,
    entity_type: str,
    match_conditions: list,
    generator: str = "core",
    match_limit: int = 25,
    enrichments: list = None,
) -> str:
    """Start a FindAll run."""
    params = {
        "objective": objective,
        "entity_type": entity_type,
        "match_conditions": match_conditions,
        "generator": generator,
        "match_limit": match_limit,
    }
    
    if enrichments:
        params["enrichments"] = enrichments
    
    result = client.beta.findall.create(**params)
    return result.findall_id


def poll_findall(client: Parallel, findall_id: str, timeout: int = 600) -> dict:
    """Poll until FindAll completes."""
    start = time.time()
    while time.time() - start < timeout:
        result = client.beta.findall.retrieve(findall_id)
        status = result.status.status if hasattr(result.status, 'status') else result.status
        
        if status == "completed":
            return result
        elif status == "failed":
            raise Exception(f"FindAll failed: {result}")
        
        # Show progress
        if hasattr(result.status, 'metrics'):
            m = result.status.metrics
            gen = getattr(m, 'generated_candidates_count', 0)
            matched = getattr(m, 'matched_candidates_count', 0)
            print(f"⏳ Progress: {matched} matched / {gen} generated", file=sys.stderr)
        
        time.sleep(5)
    raise TimeoutError(f"FindAll {findall_id} did not complete within {timeout}s")


def format_result(result) -> str:
    """Format FindAll result for display."""
    output = []
    
    findall_id = result.findall_id
    status = result.status.status if hasattr(result.status, 'status') else result.status
    metrics = result.status.metrics if hasattr(result.status, 'metrics') else None
    
    output.append(f"🔍 FindAll: {findall_id}")
    output.append(f"   Status: {status}")
    
    if metrics:
        gen = getattr(metrics, 'generated_candidates_count', 0)
        matched = getattr(metrics, 'matched_candidates_count', 0)
        output.append(f"   Candidates: {matched} matched / {gen} generated")
    output.append("")
    
    if hasattr(result, 'candidates') and result.candidates:
        output.append("**Matched Entities:**")
        for i, candidate in enumerate(result.candidates, 1):
            name = getattr(candidate, 'name', 'Unknown')
            url = getattr(candidate, 'url', '')
            desc = getattr(candidate, 'description', '')[:150]
            
            output.append(f"\n**{i}. {name}**")
            if url:
                output.append(f"   URL: {url}")
            if desc:
                output.append(f"   {desc}")
            
            # Show enrichments if present
            if hasattr(candidate, 'enrichments') and candidate.enrichments:
                for key, val in candidate.enrichments.items():
                    val_str = str(val)[:100]
                    output.append(f"   • {key}: {val_str}")
    
    return "\n".join(output)


def main():
    parser = argparse.ArgumentParser(description="Parallel.ai FindAll API")
    parser.add_argument("query", nargs="*", help="Natural language query")
    parser.add_argument("--generator", "-g", default="core",
                       choices=["base", "core", "pro"],
                       help="Generator tier (base=budget, core=balanced, pro=comprehensive)")
    parser.add_argument("--limit", "-l", type=int, default=25,
                       help="Maximum matched entities to return")
    parser.add_argument("--enrich", "-e", metavar="FIELDS",
                       help="Comma-separated enrichment fields (e.g., 'funding,employee_count')")
    parser.add_argument("--status", "-s", metavar="ID",
                       help="Check status of existing FindAll job")
    parser.add_argument("--timeout", "-t", type=int, default=600,
                       help="Timeout in seconds (default: 600)")
    parser.add_argument("--json", "-j", action="store_true",
                       help="Output raw JSON")
    parser.add_argument("--no-wait", action="store_true",
                       help="Don't wait for completion, just return findall_id")
    
    args = parser.parse_args()
    
    client = Parallel(api_key=API_KEY)
    
    # Check status of existing job
    if args.status:
        result = client.beta.findall.retrieve(args.status)
        if args.json:
            print(json.dumps(result.__dict__, indent=2, default=str))
        else:
            print(format_result(result))
        return
    
    if not args.query:
        parser.print_help()
        sys.exit(1)
    
    query = " ".join(args.query)
    
    try:
        # Step 1: Ingest - convert natural language to schema
        print(f"📝 Analyzing query...", file=sys.stderr)
        schema = ingest_query(client, query)
        
        entity_type = schema.entity_type
        match_conditions = [
            {"name": c.name, "description": c.description}
            for c in schema.match_conditions
        ]
        
        print(f"   Entity type: {entity_type}", file=sys.stderr)
        print(f"   Match conditions: {len(match_conditions)}", file=sys.stderr)
        
        # Parse enrichments
        enrichments = None
        if args.enrich:
            enrichments = [
                {"name": f.strip(), "description": f"The {f.strip().replace('_', ' ')}"}
                for f in args.enrich.split(",")
            ]
        
        # Step 2: Create FindAll run
        print(f"🚀 Starting FindAll...", file=sys.stderr)
        findall_id = create_findall(
            client,
            objective=schema.objective,
            entity_type=entity_type,
            match_conditions=match_conditions,
            generator=args.generator,
            match_limit=args.limit,
            enrichments=enrichments,
        )
        
        if args.no_wait:
            print(f"FindAll created: {findall_id}")
            return
        
        # Step 3: Poll for completion
        result = poll_findall(client, findall_id, timeout=args.timeout)
        
        if args.json:
            output = {
                "findall_id": result.findall_id,
                "status": result.status.status if hasattr(result.status, 'status') else result.status,
                "candidates": [
                    {
                        "name": getattr(c, 'name', None),
                        "url": getattr(c, 'url', None),
                        "description": getattr(c, 'description', None),
                        "enrichments": getattr(c, 'enrichments', None),
                    }
                    for c in (result.candidates or [])
                ]
            }
            print(json.dumps(output, indent=2, default=str))
        else:
            print(format_result(result))
            
    except Exception as e:
        print(f"❌ Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Parallel.ai Monitor API - Continuous web tracking with alerts.

Usage:
  python3 monitor.py create "Track AI funding news" --cadence daily
  python3 monitor.py create "Alert when AirPods drop below $150" --cadence hourly --webhook https://...
  python3 monitor.py list  # List all monitors
  python3 monitor.py events monitor_abc123  # Get events for a monitor
  python3 monitor.py delete monitor_abc123  # Delete a monitor
"""

import os
import sys
import json
import argparse
import requests

API_KEY = os.environ.get("PARALLEL_API_KEY")
if not API_KEY:
    print("Error: PARALLEL_API_KEY environment variable is required", file=sys.stderr)
    sys.exit(1)
BASE_URL = "https://api.parallel.ai/v1alpha"


def api_request(method: str, endpoint: str, data: dict = None) -> dict:
    """Make API request to Parallel."""
    headers = {
        "x-api-key": API_KEY,
        "Content-Type": "application/json",
    }
    
    url = f"{BASE_URL}{endpoint}"
    
    if method == "GET":
        response = requests.get(url, headers=headers, params=data)
    elif method == "POST":
        response = requests.post(url, headers=headers, json=data)
    elif method == "DELETE":
        response = requests.delete(url, headers=headers)
    else:
        raise ValueError(f"Unsupported method: {method}")
    
    response.raise_for_status()
    return response.json() if response.text else {}


def create_monitor(
    query: str,
    cadence: str = "daily",
    webhook_url: str = None,
    metadata: dict = None,
) -> dict:
    """Create a new monitor."""
    data = {
        "query": query,
        "cadence": cadence,
    }
    
    if webhook_url:
        data["webhook"] = {
            "url": webhook_url,
            "event_types": ["monitor.event.detected", "monitor.run.completed"]
        }
    
    if metadata:
        data["metadata"] = metadata
    
    return api_request("POST", "/monitors", data)


def list_monitors() -> list:
    """List all monitors."""
    return api_request("GET", "/monitors")


def get_events(monitor_id: str, lookback: str = None) -> dict:
    """Get events for a monitor."""
    params = {}
    if lookback:
        params["lookback"] = lookback
    return api_request("GET", f"/monitors/{monitor_id}/events", params)


def delete_monitor(monitor_id: str) -> bool:
    """Delete a monitor."""
    api_request("DELETE", f"/monitors/{monitor_id}")
    return True


def format_monitor(monitor: dict) -> str:
    """Format a single monitor for display."""
    output = []
    
    monitor_id = monitor.get("monitor_id", "unknown")
    query = monitor.get("query", "")
    status = monitor.get("status", "unknown")
    cadence = monitor.get("cadence", "unknown")
    created = monitor.get("created_at")
    
    output.append(f"📡 {monitor_id}")
    output.append(f"   Query: {query[:100]}")
    output.append(f"   Status: {status} | Cadence: {cadence}")
    if created:
        output.append(f"   Created: {created}")
    
    return "\n".join(output)


def format_events(events_result: dict) -> str:
    """Format events for display."""
    output = []
    
    events = events_result.get("events", [])
    
    if not events:
        return "No events found."
    
    output.append(f"📋 Events ({len(events)} total)")
    output.append("")
    
    for i, event in enumerate(events[:20], 1):
        event_type = event.get("type", "event")
        event_date = event.get("event_date")
        event_output = event.get("output", "")
        sources = event.get("source_urls", [])
        
        date_str = f" ({event_date})" if event_date else ""
        output.append(f"**{i}. {event_type}**{date_str}")
        
        if event_output:
            output.append(f"   {event_output[:200]}")
        
        if sources:
            for src in sources[:2]:
                output.append(f"   🔗 {src}")
        output.append("")
    
    return "\n".join(output)


def main():
    parser = argparse.ArgumentParser(description="Parallel.ai Monitor API")
    subparsers = parser.add_subparsers(dest="command", help="Commands")
    
    # Create command
    create_parser = subparsers.add_parser("create", help="Create a new monitor")
    create_parser.add_argument("query", help="What to monitor")
    create_parser.add_argument("--cadence", "-c", default="daily",
                              choices=["hourly", "daily", "weekly"],
                              help="How often to check")
    create_parser.add_argument("--webhook", "-w", metavar="URL",
                              help="Webhook URL for notifications")
    create_parser.add_argument("--metadata", "-m", metavar="JSON",
                              help="JSON metadata to attach")
    
    # List command
    subparsers.add_parser("list", help="List all monitors")
    
    # Events command
    events_parser = subparsers.add_parser("events", help="Get events for a monitor")
    events_parser.add_argument("monitor_id", help="Monitor ID")
    events_parser.add_argument("--lookback", "-l", metavar="DURATION",
                              help="Lookback duration (e.g., '10d', '1w')")
    
    # Delete command
    delete_parser = subparsers.add_parser("delete", help="Delete a monitor")
    delete_parser.add_argument("monitor_id", help="Monitor ID to delete")
    
    # Global options
    parser.add_argument("--json", "-j", action="store_true",
                       help="Output raw JSON")
    
    args = parser.parse_args()
    
    if not args.command:
        parser.print_help()
        sys.exit(1)
    
    try:
        if args.command == "create":
            metadata = json.loads(args.metadata) if args.metadata else None
            result = create_monitor(
                query=args.query,
                cadence=args.cadence,
                webhook_url=args.webhook,
                metadata=metadata,
            )
            if args.json:
                print(json.dumps(result, indent=2))
            else:
                print(f"✅ Monitor created!")
                print(format_monitor(result))
                
        elif args.command == "list":
            result = list_monitors()
            monitors = result.get("monitors", result) if isinstance(result, dict) else result
            
            if args.json:
                print(json.dumps(monitors, indent=2))
            else:
                if not monitors:
                    print("No monitors found.")
                else:
                    print(f"📡 Monitors ({len(monitors)} total)\n")
                    for monitor in monitors:
                        print(format_monitor(monitor))
                        print()
                        
        elif args.command == "events":
            result = get_events(args.monitor_id, lookback=args.lookback)
            if args.json:
                print(json.dumps(result, indent=2))
            else:
                print(format_events(result))
                
        elif args.command == "delete":
            delete_monitor(args.monitor_id)
            print(f"✅ Monitor {args.monitor_id} deleted.")
            
    except requests.exceptions.HTTPError as e:
        print(f"❌ API Error: {e.response.status_code} - {e.response.text}", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"❌ Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/bin/bash
# Parallel.ai Task API wrapper
# Usage: ./parallel.sh <command> [args]

set -e

API_KEY="${PARALLEL_API_KEY:?Error: PARALLEL_API_KEY environment variable is required}"
BASE_URL="https://api.parallel.ai/v1"
MAX_WAIT="${PARALLEL_MAX_WAIT:-120}"

# Submit task and poll for result
run_task() {
  local input="$1"
  local processor="${2:-base}"
  
  # Submit
  local payload=$(jq -n --arg proc "$processor" --arg inp "$input" \
    '{"processor": $proc, "input": $inp}')
  local response=$(curl -s -X POST "$BASE_URL/tasks/runs" \
    -H "x-api-key: $API_KEY" \
    -H "Content-Type: application/json" \
    -d "$payload")
  
  local run_id=$(echo "$response" | jq -r '.run_id // empty')
  
  if [ -z "$run_id" ]; then
    echo "$response" | jq '.'
    return 1
  fi
  
  echo "⏳ Task: $run_id" >&2
  
  # Poll for completion
  local elapsed=0
  while [ $elapsed -lt $MAX_WAIT ]; do
    sleep 3
    elapsed=$((elapsed + 3))
    
    local status_response=$(curl -s -X GET "$BASE_URL/tasks/runs/$run_id" \
      -H "x-api-key: $API_KEY")
    
    local status=$(echo "$status_response" | jq -r '.status')
    local output=$(echo "$status_response" | jq -r '.output // empty')
    
    if [ "$status" = "completed" ]; then
      if [ -n "$output" ] && [ "$output" != "null" ]; then
        echo "$output"
      else
        echo "✅ Completed (no output in response)" >&2
        echo "$status_response" | jq '.'
      fi
      return 0
    elif [ "$status" = "failed" ] || [ "$status" = "error" ]; then
      echo "❌ Failed" >&2
      echo "$status_response" | jq '.'
      return 1
    fi
    
    printf "." >&2
  done
  
  echo "" >&2
  echo "⏰ Timeout. Run ID: $run_id" >&2
  return 1
}

command="${1:-help}"
shift || true

case "$command" in
  research)
    QUERY="$*"
    [ -z "$QUERY" ] && { echo "Usage: parallel.sh research <query>" >&2; exit 1; }
    run_task "$QUERY" "base"
    ;;
    
  company)
    COMPANY="$*"
    [ -z "$COMPANY" ] && { echo "Usage: parallel.sh company <name>" >&2; exit 1; }
    run_task "Research this company comprehensively: $COMPANY. Include: description, leadership, products, recent news, funding, competitors." "base"
    ;;
    
  person)
    PERSON="$*"
    [ -z "$PERSON" ] && { echo "Usage: parallel.sh person <name>" >&2; exit 1; }
    run_task "Research this person: $PERSON. Include: background, current role, achievements, recent news." "base"
    ;;
    
  status)
    RUN_ID="$1"
    [ -z "$RUN_ID" ] && { echo "Usage: parallel.sh status <run_id>" >&2; exit 1; }
    curl -s -X GET "$BASE_URL/tasks/runs/$RUN_ID" \
      -H "x-api-key: $API_KEY" | jq '.'
    ;;
    
  help|*)
    cat << 'EOF'
Parallel.ai Task API - Deep web research

Commands:
  research <query>     General research query
  company <name>       Company research  
  person <name>        Person research
  status <run_id>      Check task status

Examples:
  parallel.sh research "What are the latest developments in AI safety?"
  parallel.sh company "Anthropic"
  parallel.sh person "Dario Amodei"

Note: Search API requires separate product activation at platform.parallel.ai
EOF
    ;;
esac

#!/usr/bin/env python3
"""
Parallel.ai Search API
Usage: python3 search.py <query> [--max-results N] [--mode one-shot|agentic]
"""

import os
import sys
import json
import argparse

from parallel import Parallel

API_KEY = os.environ.get("PARALLEL_API_KEY")
if not API_KEY:
    print("Error: PARALLEL_API_KEY environment variable is required", file=sys.stderr)
    sys.exit(1)

def search(objective: str, max_results: int = 10, mode: str = "one-shot"):
    """Search using Parallel SDK."""
    client = Parallel(api_key=API_KEY)
    return client.beta.search(
        mode=mode,
        max_results=max_results,
        objective=objective
    )

def format_results(response) -> str:
    """Format search results for display."""
    output = []
    output.append(f"🔍 Search ID: {response.search_id}\n")
    
    for i, result in enumerate(response.results, 1):
        title = result.title or "No title"
        url = result.url
        excerpts = result.excerpts or []
        date = f" ({result.publish_date})" if result.publish_date else ""
        
        output.append(f"**{i}. [{title}]({url})**{date}")
        
        if excerpts:
            # Clean and truncate excerpt
            excerpt = excerpts[0].replace("\n", " ").strip()[:400]
            output.append(f"   {excerpt}...")
        output.append("")
    
    if response.usage:
        usage = ", ".join(f"{u.name}: {u.count}" for u in response.usage)
        output.append(f"📊 Usage: {usage}")
    
    return "\n".join(output)

def main():
    parser = argparse.ArgumentParser(description="Parallel.ai Search")
    parser.add_argument("query", nargs="*", help="Search query")
    parser.add_argument("--max-results", "-n", type=int, default=10)
    parser.add_argument("--mode", "-m", default="one-shot", choices=["one-shot", "agentic", "fast"])
    parser.add_argument("--json", "-j", action="store_true", help="Output raw JSON")
    
    args = parser.parse_args()
    
    if not args.query:
        parser.print_help()
        sys.exit(1)
    
    query = " ".join(args.query)
    response = search(query, max_results=args.max_results, mode=args.mode)
    
    if args.json:
        # Convert to dict for JSON output
        print(json.dumps({
            "search_id": response.search_id,
            "results": [
                {
                    "url": r.url,
                    "title": r.title,
                    "publish_date": r.publish_date,
                    "excerpts": r.excerpts
                }
                for r in response.results
            ]
        }, indent=2))
    else:
        print(format_results(response))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Parallel.ai Task API - Deep research, enrichment, and authenticated sources.

Usage:
  python3 task.py "What was France's GDP in 2023?"
  python3 task.py --enrich "company_name=Stripe" --output "founding_year,funding"
  python3 task.py --report "Market analysis of HVAC industry"
  
  # Authenticated page access (requires browser-use.com API key)
  export BROWSERUSE_API_KEY="your-key"
  python3 task.py "Extract migration docs from https://nxp.com/products/K66_180"
"""

import os
import sys
import json
import argparse
import time

from parallel import Parallel

API_KEY = os.environ.get("PARALLEL_API_KEY")
if not API_KEY:
    print("Error: PARALLEL_API_KEY environment variable is required", file=sys.stderr)
    sys.exit(1)


def create_task(
    client: Parallel,
    input_data,
    processor: str = "core",
    task_spec: dict = None,
    include_domains: list = None,
    exclude_domains: list = None,
    mcp_servers: list = None,
) -> dict:
    """Create and run a task."""
    params = {
        "input": input_data,
        "processor": processor,
    }
    
    if task_spec:
        params["task_spec"] = task_spec
    
    # Source policy
    if include_domains or exclude_domains:
        source_policy = {}
        if include_domains:
            source_policy["include_domains"] = include_domains
        if exclude_domains:
            source_policy["exclude_domains"] = exclude_domains
        params["source_policy"] = source_policy
    
    # MCP servers for authenticated browsing (Jan 2026 feature)
    if mcp_servers:
        params["mcp_servers"] = mcp_servers
        params["betas"] = ["mcp-server-2025-07-17"]
    
    # Create task run
    task_run = client.beta.task_run.create(**params)
    return task_run


def poll_task(client: Parallel, run_id: str, timeout: int = 300) -> dict:
    """Poll until task completes."""
    start = time.time()
    while time.time() - start < timeout:
        result = client.beta.task_run.retrieve(run_id)
        if result.run.status == "completed":
            return result
        elif result.run.status == "failed":
            raise Exception(f"Task failed: {result.run}")
        time.sleep(2)
    raise TimeoutError(f"Task {run_id} did not complete within {timeout}s")


def build_enrichment_spec(input_fields: str, output_fields: str) -> tuple:
    """Build input/output schemas for enrichment."""
    # Parse input: "company_name=Stripe,website=stripe.com"
    input_data = {}
    input_props = {}
    for pair in input_fields.split(","):
        if "=" in pair:
            key, val = pair.split("=", 1)
            input_data[key.strip()] = val.strip()
            input_props[key.strip()] = {"type": "string"}
    
    # Parse output: "founding_year,employee_count,funding"
    output_props = {}
    for field in output_fields.split(","):
        field = field.strip()
        if field:
            output_props[field] = {
                "type": "string",
                "description": f"The {field.replace('_', ' ')} of the entity"
            }
    
    task_spec = {
        "input_schema": {
            "type": "json",
            "json_schema": {
                "type": "object",
                "properties": input_props,
                "required": list(input_props.keys()),
                "additionalProperties": False
            }
        },
        "output_schema": {
            "type": "json",
            "json_schema": {
                "type": "object",
                "properties": output_props,
                "required": list(output_props.keys()),
                "additionalProperties": False
            }
        }
    }
    
    return input_data, task_spec


def format_result(result) -> str:
    """Format task result for display."""
    output = []
    
    run = result.run
    output.append(f"🔬 Task: {run.run_id}")
    output.append(f"   Status: {run.status} | Processor: {run.processor}")
    output.append("")
    
    if hasattr(result, 'output') and result.output:
        content = result.output.content
        output_type = result.output.type
        
        if output_type == "json" and isinstance(content, dict):
            output.append("**Results:**")
            for key, val in content.items():
                # Truncate long values
                val_str = str(val)
                if len(val_str) > 200:
                    val_str = val_str[:200] + "..."
                output.append(f"  • {key}: {val_str}")
        elif output_type == "text":
            output.append("**Report:**")
            output.append(content[:2000] + "..." if len(content) > 2000 else content)
        else:
            output.append(f"**Output ({output_type}):**")
            output.append(str(content)[:2000])
        
        # Show basis/citations if available
        if hasattr(result.output, 'basis') and result.output.basis:
            output.append("")
            output.append("**Citations:**")
            for basis in result.output.basis[:5]:  # Limit to 5
                field = basis.field if hasattr(basis, 'field') else 'result'
                confidence = basis.confidence if hasattr(basis, 'confidence') else 'unknown'
                output.append(f"  [{field}] confidence: {confidence}")
                if hasattr(basis, 'citations'):
                    for cite in basis.citations[:2]:
                        output.append(f"    - {cite.title}: {cite.url}")
    
    return "\n".join(output)


def main():
    parser = argparse.ArgumentParser(description="Parallel.ai Task API")
    parser.add_argument("query", nargs="*", help="Research query or question")
    parser.add_argument("--processor", "-p", default="core", 
                       choices=["base", "core", "ultra"],
                       help="Processor tier (base=fast, core=standard, ultra=deep)")
    parser.add_argument("--enrich", "-e", metavar="FIELDS",
                       help="Enrichment mode: key=value pairs (e.g., 'company_name=Stripe,website=stripe.com')")
    parser.add_argument("--output", "-o", metavar="FIELDS",
                       help="Output fields for enrichment (e.g., 'founding_year,employee_count')")
    parser.add_argument("--report", "-r", action="store_true",
                       help="Generate markdown report with citations")
    parser.add_argument("--include-domains", metavar="DOMAINS",
                       help="Comma-separated domains to include")
    parser.add_argument("--exclude-domains", metavar="DOMAINS",
                       help="Comma-separated domains to exclude")
    parser.add_argument("--browseruse-key", metavar="KEY",
                       help="browser-use.com API key for authenticated page access")
    parser.add_argument("--timeout", "-t", type=int, default=300,
                       help="Timeout in seconds (default: 300)")
    parser.add_argument("--json", "-j", action="store_true",
                       help="Output raw JSON")
    parser.add_argument("--no-wait", action="store_true",
                       help="Don't wait for completion, just return run_id")
    
    args = parser.parse_args()
    
    client = Parallel(api_key=API_KEY)
    
    # Determine input and task spec
    input_data = None
    task_spec = None
    processor = args.processor
    
    if args.enrich:
        if not args.output:
            print("Error: --enrich requires --output fields", file=sys.stderr)
            sys.exit(1)
        input_data, task_spec = build_enrichment_spec(args.enrich, args.output)
    elif args.report:
        query = " ".join(args.query) if args.query else None
        if not query:
            print("Error: --report requires a query", file=sys.stderr)
            sys.exit(1)
        input_data = query
        task_spec = {"output_schema": {"type": "text"}}
        processor = "ultra"  # Reports need deep processing
    else:
        query = " ".join(args.query) if args.query else None
        if not query:
            parser.print_help()
            sys.exit(1)
        input_data = query
    
    # Parse domain filters
    include_domains = None
    exclude_domains = None
    if args.include_domains:
        include_domains = [d.strip() for d in args.include_domains.split(",")]
    if args.exclude_domains:
        exclude_domains = [d.strip() for d in args.exclude_domains.split(",")]
    
    # Build MCP servers for authenticated browsing
    mcp_servers = None
    browseruse_key = args.browseruse_key or os.environ.get("BROWSERUSE_API_KEY")
    if browseruse_key:
        mcp_servers = [{
            "type": "url",
            "url": "https://api.browser-use.com/mcp",
            "name": "browseruse",
            "headers": {"Authorization": f"Bearer {browseruse_key}"}
        }]
    
    # Create task
    try:
        task_run = create_task(
            client,
            input_data,
            processor=processor,
            task_spec=task_spec,
            include_domains=include_domains,
            exclude_domains=exclude_domains,
            mcp_servers=mcp_servers,
        )
        
        run_id = task_run.run.run_id if hasattr(task_run, 'run') else task_run.run_id
        
        if args.no_wait:
            print(f"Task created: {run_id}")
            return
        
        # Poll for completion
        print(f"⏳ Running task {run_id}...", file=sys.stderr)
        result = poll_task(client, run_id, timeout=args.timeout)
        
        if args.json:
            # Convert to dict for JSON output
            output = {
                "run_id": result.run.run_id,
                "status": result.run.status,
                "processor": result.run.processor,
            }
            if hasattr(result, 'output') and result.output:
                output["output"] = {
                    "type": result.output.type,
                    "content": result.output.content,
                }
                if hasattr(result.output, 'basis'):
                    output["basis"] = [
                        {
                            "field": b.field if hasattr(b, 'field') else None,
                            "confidence": b.confidence if hasattr(b, 'confidence') else None,
                            "citations": [
                                {"title": c.title, "url": c.url}
                                for c in (b.citations if hasattr(b, 'citations') else [])
                            ]
                        }
                        for b in result.output.basis
                    ]
            print(json.dumps(output, indent=2, default=str))
        else:
            print(format_result(result))
            
    except Exception as e:
        print(f"❌ Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()