Give Every CrewAI Agent Its Own Machine

# Install CrewAI and httpx (our HTTP client for sandbox API calls)
pip install crewai httpx

# Verify versions
python -c "import crewai; print(f'CrewAI {crewai.__version__}')"
python -c "import httpx; print(f'httpx {httpx.__version__}')"

# Set your API keys
export SANDBOX_API_KEY="ab0t_sk_live_YOUR_KEY_HERE"
export SANDBOX_URL="https://sandbox.dev.ab0t.com"
export OPENAI_API_KEY="sk-YOUR_KEY_HERE"  # CrewAI uses OpenAI by default

import os
import time
import httpx

class SandboxClient:
    """Thin wrapper around the Sandbox Platform API."""

    def __init__(self):
        self.base_url = os.environ["SANDBOX_URL"]
        self.api_key = os.environ["SANDBOX_API_KEY"]
        self.client = httpx.Client(
            base_url=self.base_url,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            timeout=120.0,
        )

    def create_sandbox(self, name: str, sandbox_type: str = "terminal",
                        instance_type: str = "ab0t.small") -> dict:
        """Create a sandbox and wait for it to be ready."""
        resp = self.client.post("/api/sandboxes", json={
            "name": name,
            "type": sandbox_type,  # "terminal", "browser", or "desktop"
            "instance_type": instance_type,
            "auto_stop_minutes": 60,
        })
        resp.raise_for_status()
        sandbox = resp.json()
        sandbox_id = sandbox["sandbox_id"]

        # Poll until ready (typically 20-40 seconds)
        for _ in range(60):
            status = self.get_sandbox(sandbox_id)
            if status["status"] == "running":
                return status
            time.sleep(2)
        raise TimeoutError(f"Sandbox {sandbox_id} did not start in time")

    def get_sandbox(self, sandbox_id: str) -> dict:
        resp = self.client.get(f"/api/sandboxes/{sandbox_id}")
        resp.raise_for_status()
        return resp.json()

    def execute(self, sandbox_id: str, command: str,
                timeout_seconds: int = 300) -> dict:
        """Execute a shell command in a sandbox."""
        resp = self.client.post(
            f"/api/sandboxes/{sandbox_id}/execute",
            json={"command": command, "timeout": timeout_seconds},
            timeout=timeout_seconds + 10,
        )
        resp.raise_for_status()
        return resp.json()

    def browse(self, sandbox_id: str, url: str) -> dict:
        """Open a URL in the sandbox's browser and get page content."""
        resp = self.client.post(
            f"/api/sandboxes/{sandbox_id}/browse",
            json={"url": url, "wait_for": "networkidle"},
        )
        resp.raise_for_status()
        return resp.json()

    def download_file(self, sandbox_id: str, path: str) -> bytes:
        """Download a file from a sandbox."""
        resp = self.client.get(
            f"/api/sandboxes/{sandbox_id}/files",
            params={"path": path},
        )
        resp.raise_for_status()
        return resp.content

    def upload_file(self, sandbox_id: str, path: str,
                    content: str) -> dict:
        """Upload a file to a sandbox."""
        resp = self.client.post(
            f"/api/sandboxes/{sandbox_id}/files",
            json={"path": path, "content": content},
        )
        resp.raise_for_status()
        return resp.json()

    def take_screenshot(self, sandbox_id: str) -> bytes:
        """Take a screenshot of a desktop sandbox."""
        resp = self.client.get(
            f"/api/sandboxes/{sandbox_id}/screenshot",
        )
        resp.raise_for_status()
        return resp.content

    def stop_sandbox(self, sandbox_id: str) -> dict:
        resp = self.client.post(f"/api/sandboxes/{sandbox_id}/stop")
        resp.raise_for_status()
        return resp.json()


# Singleton instance used by all tools
sandbox_client = SandboxClient()

from crewai.tools import tool
from sandbox_client import sandbox_client
import json

# ── Browser tools (for the Researcher) ──────────────────────────

@tool
def browse_url(sandbox_id: str, url: str) -> str:
    """Browse a URL in the sandbox's Chrome browser and return the page
    content as text. Use this to read documentation, READMEs, blog posts,
    and any other web page. Returns the visible text content of the page."""
    result = sandbox_client.browse(sandbox_id, url)
    return result.get("text_content", result.get("html", "No content returned"))

@tool
def search_web(sandbox_id: str, query: str) -> str:
    """Search the web using the sandbox's browser. Returns search result
    snippets and URLs. Use this to find documentation, tutorials, and
    relevant resources before browsing specific pages."""
    url = f"https://www.google.com/search?q={query}"
    result = sandbox_client.browse(sandbox_id, url)
    return result.get("text_content", "No results returned")

# ── Terminal tools (for the Developer) ──────────────────────────

@tool
def run_command(sandbox_id: str, command: str) -> str:
    """Execute a shell command in the sandbox terminal. Use this to run
    git commands, install packages, run tests, compile code, or any other
    CLI operation. Returns stdout and stderr."""
    result = sandbox_client.execute(sandbox_id, command)
    output = result.get("stdout", "")
    errors = result.get("stderr", "")
    exit_code = result.get("exit_code", -1)
    return f"Exit code: {exit_code}\n\nSTDOUT:\n{output}\n\nSTDERR:\n{errors}"

@tool
def write_file(sandbox_id: str, file_path: str, content: str) -> str:
    """Write content to a file in the sandbox. Creates the file if it
    doesn't exist, overwrites if it does. Use this to create scripts,
    config files, or any other text file."""
    sandbox_client.upload_file(sandbox_id, file_path, content)
    return f"File written: {file_path}"

@tool
def read_file(sandbox_id: str, file_path: str) -> str:
    """Read the contents of a file from the sandbox. Returns the file
    content as a string. Use this to read test results, log files,
    source code, or any text file."""
    content = sandbox_client.download_file(sandbox_id, file_path)
    return content.decode("utf-8", errors="replace")

# ── Desktop tools (for the Analyst) ─────────────────────────────

@tool
def desktop_command(sandbox_id: str, command: str) -> str:
    """Execute a command on the desktop sandbox. The sandbox has a full
    XFCE desktop with LibreOffice, a web browser, and standard Linux tools.
    Use this to run GUI applications, generate documents, and produce
    reports."""
    result = sandbox_client.execute(sandbox_id, command)
    output = result.get("stdout", "")
    errors = result.get("stderr", "")
    exit_code = result.get("exit_code", -1)
    return f"Exit code: {exit_code}\n\nSTDOUT:\n{output}\n\nSTDERR:\n{errors}"

@tool
def take_desktop_screenshot(sandbox_id: str) -> str:
    """Take a screenshot of the desktop sandbox. Saves the screenshot
    locally and returns the file path. Use this to verify that GUI
    applications are rendering correctly."""
    image_bytes = sandbox_client.take_screenshot(sandbox_id)
    path = f"/tmp/screenshot_{sandbox_id}.png"
    with open(path, "wb") as f:
        f.write(image_bytes)
    return f"Screenshot saved to {path}"

from crewai import Agent
from sandbox_tools import (
    browse_url, search_web,
    run_command, write_file, read_file,
    desktop_command, take_desktop_screenshot,
)


researcher = Agent(
    role="Senior Research Analyst",
    goal="Thoroughly research a GitHub repository by browsing its documentation, "
         "README, wiki, and related web resources. Produce a structured summary "
         "of the project's purpose, architecture, dependencies, and notable patterns.",
    backstory="You are an expert technical researcher who specializes in analyzing "
              "open-source projects. You have access to a dedicated browser sandbox. "
              "You browse real web pages to gather information. You never fabricate "
              "content — you only report what you actually read on the page.",
    tools=[browse_url, search_web],
    verbose=True,
    memory=True,
)

developer = Agent(
    role="Senior Software Developer",
    goal="Clone a GitHub repository, analyze its code structure, install "
         "dependencies, run the test suite, and report on code quality, "
         "test coverage, and any issues found.",
    backstory="You are a seasoned developer with deep expertise in Python, "
              "JavaScript, and system administration. You have a dedicated terminal "
              "sandbox with full shell access. You clone repos, run tests, read code, "
              "and produce detailed technical assessments.",
    tools=[run_command, write_file, read_file],
    verbose=True,
    memory=True,
)

analyst = Agent(
    role="Technical Report Analyst",
    goal="Compile research findings and technical assessments into a polished "
         "report with structured sections, tables, and a professional format. "
         "Use the desktop environment to produce a finished document.",
    backstory="You are a technical writer and analyst who produces executive-grade "
              "reports. You have a desktop sandbox with LibreOffice and full Linux "
              "GUI tools. You write reports in Markdown, convert them to PDF using "
              "pandoc, and verify the output looks correct.",
    tools=[desktop_command, take_desktop_screenshot, read_file],
    verbose=True,
    memory=True,
)

from crewai import Task
from agents import researcher, developer, analyst


def create_tasks(repo_url: str, sandbox_ids: dict) -> list:
    """Create tasks for analyzing a GitHub repo.

    sandbox_ids should be:
        {"researcher": "sandbox_xxx", "developer": "sandbox_yyy", "analyst": "sandbox_zzz"}
    """

    research_task = Task(
        description=f"""Research the GitHub repository at {repo_url}.

Your browser sandbox ID is: {sandbox_ids['researcher']}
Pass this sandbox_id to every tool call.

Steps:
1. Browse the repository's main page at {repo_url}
2. Read the README.md by browsing {repo_url}/blob/main/README.md
3. Check for documentation in /docs or a wiki
4. Search the web for "{repo_url.split('/')[-1]} tutorial" and
   "{repo_url.split('/')[-1]} architecture" to find community resources
5. Browse at least 2 of the most relevant search results

Produce a structured research summary with:
- Project purpose and description
- Key features and capabilities
- Architecture overview (if documented)
- Dependencies and technology stack
- Community size and activity
- Notable patterns or design decisions""",
        expected_output="A detailed research summary in Markdown format with sections for "
                        "purpose, features, architecture, dependencies, community, and patterns.",
        agent=researcher,
    )

    development_task = Task(
        description=f"""Analyze the codebase of {repo_url} by cloning it and running its tests.

Your terminal sandbox ID is: {sandbox_ids['developer']}
Pass this sandbox_id to every tool call.

Steps:
1. Clone the repository:
   run_command(sandbox_id, "git clone {repo_url} /workspace/repo")
2. Explore the project structure:
   run_command(sandbox_id, "find /workspace/repo -type f -name '*.py' | head -50")
   run_command(sandbox_id, "cat /workspace/repo/README.md")
3. Check for a requirements.txt or pyproject.toml:
   run_command(sandbox_id, "ls /workspace/repo/requirements*.txt /workspace/repo/pyproject.toml 2>/dev/null")
4. Install dependencies:
   run_command(sandbox_id, "cd /workspace/repo && pip install -e '.[dev]' 2>/dev/null || pip install -r requirements.txt 2>/dev/null || pip install -e .")
5. Run the test suite:
   run_command(sandbox_id, "cd /workspace/repo && python -m pytest --tb=short -q 2>&1 | tail -40")
6. Check code quality:
   run_command(sandbox_id, "cd /workspace/repo && pip install ruff && ruff check . 2>&1 | tail -30")
7. Count lines of code:
   run_command(sandbox_id, "find /workspace/repo -name '*.py' -not -path '*/venv/*' | xargs wc -l 2>/dev/null | tail -5")

Produce a technical assessment covering:
- Project structure and organization
- Dependency analysis (what it requires, any concerns)
- Test results (pass/fail counts, coverage if available)
- Code quality issues found
- Lines of code and project size
- Build/install experience (smooth or problematic)""",
        expected_output="A technical assessment in Markdown format with sections for structure, "
                        "dependencies, test results, code quality, project size, and build experience.",
        agent=developer,
    )

    analysis_task = Task(
        description=f"""Compile the research findings and technical assessment into a
polished final report.

Your desktop sandbox ID is: {sandbox_ids['analyst']}
Pass this sandbox_id to every tool call.

You will receive the outputs from the researcher and developer agents.
Use them to produce a comprehensive report.

Steps:
1. Write the report as a Markdown file:
   desktop_command(sandbox_id, "cat > /workspace/report.md << 'REPORT'\n...\nREPORT")

   The report should include these sections:
   - Executive Summary (3-4 sentences)
   - Project Overview (from research)
   - Technical Architecture (from research + code analysis)
   - Code Quality Assessment (from developer)
   - Test Suite Analysis (from developer)
   - Strengths and Weaknesses
   - Recommendations
   - Appendix: Raw Metrics

2. Convert to PDF using pandoc:
   desktop_command(sandbox_id, "apt-get install -y pandoc texlive-latex-base texlive-fonts-recommended 2>/dev/null")
   desktop_command(sandbox_id, "cd /workspace && pandoc report.md -o report.pdf --pdf-engine=pdflatex -V geometry:margin=1in -V fontsize=11pt")

3. Verify the PDF was created:
   desktop_command(sandbox_id, "ls -la /workspace/report.pdf")

4. Take a screenshot of the desktop to confirm:
   take_desktop_screenshot(sandbox_id)""",
        expected_output="A comprehensive analysis report in Markdown format. Also confirm that "
                        "a PDF version was generated at /workspace/report.pdf.",
        agent=analyst,
        context=[research_task, development_task],  # depends on both
    )

    return [research_task, development_task, analysis_task]

#!/usr/bin/env python3
"""
CrewAI + Sandbox Platform: Analyze a GitHub repo with 3 dedicated agents.

Each agent gets its own sandbox:
  - Researcher: browser sandbox (Chrome) for web research
  - Developer:  terminal sandbox for cloning, building, testing
  - Analyst:    desktop sandbox (XFCE + LibreOffice) for report generation

Usage:
    export SANDBOX_API_KEY="ab0t_sk_live_YOUR_KEY"
    export SANDBOX_URL="https://sandbox.dev.ab0t.com"
    export OPENAI_API_KEY="sk-YOUR_KEY"
    python analyze_repo.py https://github.com/pydantic/pydantic
"""

import sys
import time
from crewai import Crew, Process
from sandbox_client import sandbox_client
from agents import researcher, developer, analyst
from tasks import create_tasks


def provision_sandboxes() -> dict:
    """Create three sandboxes, one per agent role."""
    print("Provisioning 3 sandboxes...")
    start = time.time()

    # Create all three in parallel-ish (the API handles queueing)
    sandbox_configs = {
        "researcher": {
            "name": "crew-researcher",
            "type": "browser",        # Chrome browser sandbox
            "instance": "ab0t.small",   # 2 vCPU, 2 GB RAM
        },
        "developer": {
            "name": "crew-developer",
            "type": "terminal",      # Terminal-only sandbox
            "instance": "ab0t.medium",  # 2 vCPU, 4 GB (needs RAM for builds)
        },
        "analyst": {
            "name": "crew-analyst",
            "type": "desktop",       # Full XFCE desktop sandbox
            "instance": "ab0t.medium",  # 2 vCPU, 4 GB (desktop needs memory)
        },
    }

    sandbox_ids = {}
    for role, config in sandbox_configs.items():
        sandbox = sandbox_client.create_sandbox(
            name=config["name"],
            sandbox_type=config["type"],
            instance_type=config["instance"],
        )
        sandbox_ids[role] = sandbox["sandbox_id"]
        print(f"  {role}: {sandbox['sandbox_id']} ({config['type']}, {config['instance']})")

    elapsed = time.time() - start
    print(f"All sandboxes ready in {elapsed:.0f}s\n")
    return sandbox_ids


def cleanup_sandboxes(sandbox_ids: dict):
    """Stop all sandboxes to stop billing."""
    print("\nCleaning up sandboxes...")
    for role, sid in sandbox_ids.items():
        try:
            sandbox_client.stop_sandbox(sid)
            print(f"  Stopped {role}: {sid}")
        except Exception as e:
            print(f"  Warning: could not stop {role} ({sid}): {e}")


def main():
    if len(sys.argv) < 2:
        print("Usage: python analyze_repo.py <github_repo_url>")
        print("Example: python analyze_repo.py https://github.com/pydantic/pydantic")
        sys.exit(1)

    repo_url = sys.argv[1]
    print(f"Analyzing: {repo_url}\n")

    # Step 1: Provision sandboxes
    sandbox_ids = provision_sandboxes()

    try:
        # Step 2: Create tasks with sandbox IDs baked in
        tasks = create_tasks(repo_url, sandbox_ids)

        # Step 3: Assemble the crew
        crew = Crew(
            agents=[researcher, developer, analyst],
            tasks=tasks,
            process=Process.sequential,  # researcher & developer first, then analyst
            verbose=True,
            memory=True,
            max_rpm=30,  # rate-limit LLM calls
        )

        # Step 4: Kick off the crew
        print("Starting crew execution...\n")
        result = crew.kickoff()

        # Step 5: Print the final report
        print("\n" + "=" * 72)
        print("FINAL REPORT")
        print("=" * 72)
        print(result)

        # Step 6: Download the PDF from the analyst's sandbox
        try:
            pdf_bytes = sandbox_client.download_file(
                sandbox_ids["analyst"], "/workspace/report.pdf"
            )
            with open("report.pdf", "wb") as f:
                f.write(pdf_bytes)
            print(f"\nPDF report saved to: report.pdf ({len(pdf_bytes):,} bytes)")
        except Exception as e:
            print(f"\nNote: Could not download PDF: {e}")
            print("The Markdown report above is the primary output.")

    finally:
        # Always clean up sandboxes
        cleanup_sandboxes(sandbox_ids)


if __name__ == "__main__":
    main()

python analyze_repo.py https://github.com/pydantic/pydantic

# Output:
# Analyzing: https://github.com/pydantic/pydantic
#
# Provisioning 3 sandboxes...
#   researcher: sandbox_r1a2b3 (browser, ab0t.small)
#   developer: sandbox_d4e5f6 (terminal, ab0t.medium)
#   analyst: sandbox_a7b8c9 (desktop, ab0t.medium)
# All sandboxes ready in 38s
#
# Starting crew execution...
#
# [Researcher] Browsing https://github.com/pydantic/pydantic ...
# [Developer]  Cloning repository ...
# [Developer]  Running pytest ...
# [Analyst]    Compiling final report ...
#
# ========================================================================
# FINAL REPORT
# ========================================================================
# # Pydantic Analysis Report
# ## Executive Summary ...
#
# PDF report saved to: report.pdf (48,291 bytes)
#
# Cleaning up sandboxes...
#   Stopped researcher: sandbox_r1a2b3
#   Stopped developer: sandbox_d4e5f6
#   Stopped analyst: sandbox_a7b8c9

# Two browser sandboxes, different sizes
fast_researcher = sandbox_client.create_sandbox(
    name="crew-researcher-fast",
    sandbox_type="browser",
    instance_type="ab0t.small",    # 2 GB — for static docs
)

heavy_researcher = sandbox_client.create_sandbox(
    name="crew-researcher-heavy",
    sandbox_type="browser",
    instance_type="ab0t.medium",   # 4 GB — for JS-heavy SPAs
)

# Two terminal sandboxes, different purposes
builder = sandbox_client.create_sandbox(
    name="crew-builder",
    sandbox_type="terminal",
    instance_type="ab0t.large",    # 8 GB — for compiling large projects
)

linter = sandbox_client.create_sandbox(
    name="crew-linter",
    sandbox_type="terminal",
    instance_type="ab0t.micro",    # 1 GB — just running ruff/eslint
)

from crewai import Crew, Process

# The key: research_task and development_task have NO context dependencies.
# Only analysis_task depends on both via context=[research_task, development_task].
# CrewAI will run the independent tasks in parallel automatically
# when you use Process.hierarchical with a manager agent.

manager = Agent(
    role="Project Manager",
    goal="Coordinate the research, development, and analysis tasks efficiently.",
    backstory="You manage a technical team. Delegate tasks and ensure quality.",
    verbose=True,
)

crew = Crew(
    agents=[researcher, developer, analyst],
    tasks=[research_task, development_task, analysis_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
)

# The manager will run researcher and developer in parallel,
# then hand off both outputs to the analyst.
result = crew.kickoff()

import json
from pathlib import Path

SANDBOX_CACHE = Path("sandbox_ids.json")

def get_or_create_sandboxes() -> dict:
    """Reuse existing sandboxes if they're still running."""

    if SANDBOX_CACHE.exists():
        cached = json.loads(SANDBOX_CACHE.read_text())
        # Verify they're still running
        all_running = True
        for role, sid in cached.items():
            try:
                status = sandbox_client.get_sandbox(sid)
                if status["status"] != "running":
                    all_running = False
                    break
            except Exception:
                all_running = False
                break
        if all_running:
            print("Reusing cached sandboxes")
            return cached

    # Create fresh sandboxes
    sandbox_ids = provision_sandboxes()
    SANDBOX_CACHE.write_text(json.dumps(sandbox_ids))
    return sandbox_ids

def transfer_file(from_sandbox: str, from_path: str,
                    to_sandbox: str, to_path: str):
    """Copy a file from one sandbox to another."""
    content = sandbox_client.download_file(from_sandbox, from_path)
    sandbox_client.upload_file(
        to_sandbox, to_path, content.decode("utf-8", errors="replace")
    )
    print(f"Transferred {from_path} -> {to_path}")

# Example: copy test results from developer to analyst
transfer_file(
    from_sandbox=sandbox_ids["developer"],
    from_path="/workspace/repo/test_results.xml",
    to_sandbox=sandbox_ids["analyst"],
    to_path="/workspace/test_results.xml",
)

import httpx
from crewai.tools import tool

@tool
def run_command_safe(sandbox_id: str, command: str) -> str:
    """Execute a shell command with automatic retry on transient failures.
    Retries up to 2 times on timeout or connection errors."""
    max_retries = 2
    for attempt in range(max_retries + 1):
        try:
            result = sandbox_client.execute(sandbox_id, command)
            exit_code = result.get("exit_code", -1)
            stdout = result.get("stdout", "")
            stderr = result.get("stderr", "")

            if exit_code != 0:
                return (f"Command failed (exit code {exit_code}).\n"
                        f"STDOUT:\n{stdout}\nSTDERR:\n{stderr}\n"
                        f"Try a different approach or fix the command.")

            return f"Success (exit code 0).\nOutput:\n{stdout}"

        except httpx.TimeoutException:
            if attempt < max_retries:
                continue
            return "Command timed out after multiple retries. The command may be " \
                   "taking too long. Try breaking it into smaller steps."

        except httpx.HTTPStatusError as e:
            return f"API error: {e.response.status_code} - {e.response.text}"

developer = Agent(
    role="Senior Software Developer",
    goal="Analyze code and run tests.",
    backstory="...",
    tools=[run_command, write_file, read_file],
    max_iter=15,        # Maximum 15 tool calls per task
    max_rpm=10,         # Maximum 10 LLM requests per minute
    verbose=True,
)

# In your run_command tool, truncate long output:
@tool
def run_command(sandbox_id: str, command: str) -> str:
    """Execute a shell command. Output is truncated to 4000 characters."""
    result = sandbox_client.execute(sandbox_id, command)
    output = result.get("stdout", "")
    if len(output) > 4000:
        output = output[:2000] + "\n\n... (truncated) ...\n\n" + output[-2000:]
    return f"Exit: {result.get('exit_code')}\n{output}"

import logging

logging.basicConfig(
    filename="crew_run.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

# In SandboxClient.execute():
def execute(self, sandbox_id, command, **kwargs):
    logging.info(f"EXEC [{sandbox_id}]: {command[:200]}")
    result = self.client.post(...)
    logging.info(f"RESULT [{sandbox_id}]: exit={result.json().get('exit_code')}")
    return result.json()

# After create_sandbox returns, verify the sandbox is responsive:
result = sandbox_client.execute(sandbox_id, "echo ready")
assert result["stdout"].strip() == "ready", "Sandbox not responsive"