┌─────────────────────────────────────────────────────────────────┐

│                         CrewAI Crew                              │

│                                                                  │

│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐       │

│  │  Researcher  │───▶│ Data Analyst │───▶│Report Writer │       │

│  └──────────────┘    └──────────────┘    └──────────────┘       │

│         │                   │                    │               │

│         │                   │                    │               │

│         ▼                   ▼                    ▼               │

│    ┌─────────┐        ┌─────────┐          ┌─────────┐          │

│    │ Search  │        │  HopX   │          │  File   │          │

│    │  Tool   │        │Sandbox  │          │  Tool   │          │

│    └─────────┘        └─────────┘          └─────────┘          │

│                            │                                     │

│                   Isolated Code Execution                        │

└─────────────────────────────────────────────────────────────────┘

pip install crewai crewai-tools hopx-ai

export OPENAI_API_KEY="sk-..."

export HOPX_API_KEY="..."

from crewai.tools import BaseTool

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type, Optional

class PythonCodeInput(BaseModel):

    """Input schema for Python execution."""

    code: str = Field(description="Python code to execute")

class PythonExecutorTool(BaseTool):

    """Execute Python code in a secure HopX sandbox."""

    name: str = "python_executor"

    description: str = """Execute Python code in an isolated sandbox.

Use this tool to:

- Perform calculations and data analysis

- Process and transform data with pandas

- Create visualizations with matplotlib

- Run any Python computation

The sandbox has pandas, numpy, matplotlib, seaborn, scikit-learn installed.

Always print() results you want to see.

For charts, use plt.savefig('/app/chart.png') then mention the file.

"""

    args_schema: Type[BaseModel] = PythonCodeInput

    def _run(self, code: str) -> str:

        """Execute code in sandbox."""

        sandbox = None

        try:

            sandbox = Sandbox.create(template="code-interpreter")

            result = sandbox.runCode(code, language="python", timeout=60)

            if result.exitCode == 0:

                output = result.stdout or "Code executed successfully (no output)"

                return f"✅ Success:\n{output}"

            else:

                return f"❌ Error:\n{result.stderr}"

        except Exception as e:

            return f"❌ Sandbox error: {str(e)}"

        finally:

            if sandbox:

                sandbox.kill()

class BashCommandInput(BaseModel):

    """Input schema for bash commands."""

    command: str = Field(description="Bash command to execute")

class BashExecutorTool(BaseTool):

    """Execute bash commands in a secure sandbox."""

    name: str = "bash_executor"

    description: str = """Execute bash/shell commands in an isolated sandbox.

Use for:

- File operations (ls, cat, head, tail)

- Installing packages (pip install)

- System commands

"""

    args_schema: Type[BaseModel] = BashCommandInput

    def _run(self, command: str) -> str:

        sandbox = None

        try:

            sandbox = Sandbox.create(template="code-interpreter")

            result = sandbox.runCode(command, language="bash", timeout=60)

            if result.exitCode == 0:

                return f"✅ Success:\n{result.stdout}"

            else:

                return f"❌ Error (exit {result.exitCode}):\n{result.stderr}"

        except Exception as e:

            return f"❌ Error: {str(e)}"

        finally:

            if sandbox:

                sandbox.kill()

from crewai.tools import BaseTool

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type, Optional, ClassVar

import threading

class SharedSandboxManager:

    """Singleton manager for shared sandbox across agents."""

    _instance: Optional['SharedSandboxManager'] = None

    _lock: ClassVar[threading.Lock] = threading.Lock()

    def __init__(self):

        self.sandbox: Optional[Sandbox] = None

        self.ttl = 600  # 10 minutes

    @classmethod

    def get_instance(cls) -> 'SharedSandboxManager':

        if cls._instance is None:

            with cls._lock:

                if cls._instance is None:

                    cls._instance = cls()

        return cls._instance

    def get_sandbox(self) -> Sandbox:

        if self.sandbox is None:

            self.sandbox = Sandbox.create(

                template="code-interpreter",

                ttl=self.ttl

            )

        return self.sandbox

    def cleanup(self):

        if self.sandbox:

            try:

                self.sandbox.kill()

            except:

                pass

            self.sandbox = None

class SharedPythonInput(BaseModel):

    code: str = Field(description="Python code to execute")

class SharedPythonTool(BaseTool):

    """Execute Python with persistent state across agents."""

    name: str = "shared_python"

    description: str = """Execute Python code with PERSISTENT STATE.

Variables, imports, and data persist between calls.

Use this when you need to build on previous computations.

All agents share the same execution environment.

Example workflow:

1. First call: import pandas; df = pd.read_csv('data.csv')

2. Second call: df['new_col'] = df['a'] * 2  # df still exists!

3. Third call: print(df.describe())  # works!

"""

    args_schema: Type[BaseModel] = SharedPythonInput

    def _run(self, code: str) -> str:

        manager = SharedSandboxManager.get_instance()

        try:

            sandbox = manager.get_sandbox()

            result = sandbox.runCode(code, language="python", timeout=60)

            if result.exitCode == 0:

                return result.stdout or "Executed (no output)"

            else:

                return f"Error: {result.stderr}"

        except Exception as e:

            manager.cleanup()

            return f"Sandbox error: {str(e)}"

from crewai import Agent, Task, Crew, Process

# Initialize tools

python_tool = PythonExecutorTool()

bash_tool = BashExecutorTool()

shared_python = SharedPythonTool()

# Research Agent - Gathers information

researcher = Agent(

    role="Senior Research Analyst",

    goal="Identify key data points and analysis requirements",

    backstory="""You are an experienced research analyst who excels at 

    breaking down complex problems into specific, measurable questions.

    You identify what data is needed and what analyses will provide insights.""",

    tools=[],  # Researcher doesn't need code execution

    verbose=True,

    allow_delegation=False

)

# Data Analyst Agent - Runs analysis

data_analyst = Agent(

    role="Senior Data Analyst",

    goal="Analyze data using Python to extract meaningful insights",

    backstory="""You are a skilled data analyst proficient in Python, pandas, 

    and statistical analysis. You write clean, efficient code to analyze data 

    and create visualizations. You always verify your results.""",

    tools=[shared_python, bash_tool],

    verbose=True,

    allow_delegation=False

)

# Report Writer Agent - Creates final report

report_writer = Agent(

    role="Technical Report Writer",

    goal="Create clear, actionable reports from analysis results",

    backstory="""You excel at translating complex technical findings into 

    clear, well-structured reports. You highlight key insights and provide 

    actionable recommendations.""",

    tools=[shared_python],  # Can run code to format/verify data

    verbose=True,

    allow_delegation=False

)

# Task 1: Research

research_task = Task(

    description="""Analyze the requirements for understanding sales performance.

    We have sales data with columns: date, product, region, quantity, revenue.

    Identify:

    1. Key metrics to calculate (totals, averages, trends)

    2. Important comparisons (by product, by region, over time)

    3. Specific questions the analysis should answer

    Output a structured analysis plan.""",

    expected_output="A detailed analysis plan with specific metrics and questions",

    agent=researcher

)

# Task 2: Data Analysis

analysis_task = Task(

    description="""Execute the analysis plan using Python.

    First, create sample sales data that matches the description:

    - 1000 rows of sales data

    - Columns: date, product, region, quantity, revenue

    - Date range: 2024-01-01 to 2024-12-31

    - Products: ["Laptop", "Phone", "Tablet", "Watch", "Headphones"]

    - Regions: ["North", "South", "East", "West"]

    Then perform the analysis:

    1. Calculate total revenue and quantity by product

    2. Calculate average order value by region

    3. Identify monthly trends

    4. Find top performing product-region combinations

    5. Calculate month-over-month growth rates

    Use the shared_python tool to maintain state between code executions.

    Print all results clearly with labels.""",

    expected_output="Complete analysis results with all calculated metrics",

    agent=data_analyst,

    context=[research_task]  # Depends on research

)

# Task 3: Report Writing

report_task = Task(

    description="""Create a comprehensive report based on the analysis.

    Structure:

    1. Executive Summary (key findings in 3-4 bullet points)

    2. Methodology (how data was analyzed)

    3. Key Findings (detailed results with numbers)

    4. Recommendations (3-5 actionable items)

    5. Appendix (any additional details)

    Use the shared_python tool if you need to access or verify any data.

    Format the report in clean markdown.""",

    expected_output="A well-structured markdown report with all sections",

    agent=report_writer,

    context=[analysis_task]  # Depends on analysis

)

# Create the crew

sales_analysis_crew = Crew(

    agents=[researcher, data_analyst, report_writer],

    tasks=[research_task, analysis_task, report_task],

    process=Process.sequential,  # Tasks run in order

    verbose=True

)

# Run it!

if __name__ == "__main__":

    try:

        result = sales_analysis_crew.kickoff()

        print("\n" + "="*60)

        print("FINAL REPORT")

        print("="*60)

        print(result)

    finally:

        # Clean up sandbox

        SharedSandboxManager.get_instance().cleanup()

from crewai import Agent, Task, Crew, Process

# Manager Agent

manager = Agent(

    role="Project Manager",

    goal="Coordinate the team to deliver high-quality analysis",

    backstory="""You are an experienced project manager who ensures 

    deliverables are on time and meet quality standards. You delegate 

    effectively and provide clear guidance.""",

    allow_delegation=True,

    verbose=True

)

# Specialist Agents

python_expert = Agent(

    role="Python Developer",

    goal="Write efficient, well-documented Python code",

    backstory="Expert Python developer with 10 years of experience.",

    tools=[shared_python, bash_tool],

    verbose=True

)

data_scientist = Agent(

    role="Data Scientist", 

    goal="Apply statistical methods and ML to extract insights",

    backstory="PhD in Statistics with expertise in ML and data analysis.",

    tools=[shared_python],

    verbose=True

)

visualizer = Agent(

    role="Data Visualization Expert",

    goal="Create clear, impactful visualizations",

    backstory="Expert in matplotlib, seaborn, and data storytelling.",

    tools=[shared_python],

    verbose=True

)

# Create hierarchical crew

hierarchical_crew = Crew(

    agents=[python_expert, data_scientist, visualizer],

    tasks=[...],  # Define tasks

    manager_agent=manager,

    process=Process.hierarchical,

    verbose=True

)

from crewai import Crew, Process

import asyncio

# Define independent analysis tasks

north_analysis = Task(

    description="Analyze North region sales data",

    agent=data_analyst,

    expected_output="North region analysis"

)

south_analysis = Task(

    description="Analyze South region sales data", 

    agent=data_analyst,

    expected_output="South region analysis"

)

east_analysis = Task(

    description="Analyze East region sales data",

    agent=data_analyst,

    expected_output="East region analysis"

)

west_analysis = Task(

    description="Analyze West region sales data",

    agent=data_analyst,

    expected_output="West region analysis"

)

# Consolidation task (runs after all parallel tasks)

consolidate_task = Task(

    description="Consolidate all regional analyses into final report",

    agent=report_writer,

    expected_output="Consolidated report",

    context=[north_analysis, south_analysis, east_analysis, west_analysis]

)

# Parallel crew

parallel_crew = Crew(

    agents=[data_analyst, report_writer],

    tasks=[north_analysis, south_analysis, east_analysis, west_analysis, consolidate_task],

    process=Process.sequential,  # CrewAI handles parallelization internally

    verbose=True

)

from crewai.tools import BaseTool

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type

import base64

class FileAnalysisInput(BaseModel):

    file_content: str = Field(description="Base64 encoded file content")

    filename: str = Field(description="Name of the file")

    analysis_code: str = Field(description="Python code to analyze the file")

class FileAnalysisTool(BaseTool):

    """Upload and analyze files in sandbox."""

    name: str = "file_analyzer"

    description: str = """Upload a file and analyze it with Python code.

Provide:

1. file_content: Base64 encoded file data

2. filename: Name to save as (e.g., 'data.csv')

3. analysis_code: Python code to analyze (file is at /app/{filename})

"""

    args_schema: Type[BaseModel] = FileAnalysisInput

    def _run(self, file_content: str, filename: str, analysis_code: str) -> str:

        sandbox = None

        try:

            sandbox = Sandbox.create(template="code-interpreter")

            # Decode and upload file

            file_bytes = base64.b64decode(file_content)

            sandbox.files.write(f"/app/{filename}", file_bytes)

            # Run analysis

            result = sandbox.runCode(analysis_code, language="python", timeout=120)

            if result.exitCode == 0:

                return f"✅ Analysis complete:\n{result.stdout}"

            else:

                return f"❌ Error:\n{result.stderr}"

        except Exception as e:

            return f"❌ Error: {str(e)}"

        finally:

            if sandbox:

                sandbox.kill()

from crewai.tools import BaseTool

from hopx import Sandbox

import time

class RobustPythonTool(BaseTool):

    """Python execution with retry and error recovery."""

    name: str = "robust_python"

    description: str = "Execute Python with automatic error recovery"

    max_retries: int = 3

    retry_delay: float = 2.0

    def _run(self, code: str) -> str:

        last_error = None

        for attempt in range(self.max_retries):

            sandbox = None

            try:

                sandbox = Sandbox.create(template="code-interpreter")

                result = sandbox.runCode(code, language="python", timeout=60)

                if result.exitCode == 0:

                    return result.stdout or "Success"

                else:

                    # Syntax/runtime error - don't retry, let agent fix

                    return f"Code error:\n{result.stderr}"

            except Exception as e:

                last_error = str(e)

                if attempt < self.max_retries - 1:

                    time.sleep(self.retry_delay)

            finally:

                if sandbox:

                    try:

                        sandbox.kill()

                    except:

                        pass

        return f"Sandbox failed after {self.max_retries} attempts: {last_error}"

"""

Market Research Crew with Secure Code Execution

"""

from crewai import Agent, Task, Crew, Process

from crewai.tools import BaseTool

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type, Optional

import os

# Ensure API keys

assert os.environ.get("OPENAI_API_KEY"), "Set OPENAI_API_KEY"

assert os.environ.get("HOPX_API_KEY"), "Set HOPX_API_KEY"

class SandboxManager:

    """Manage shared sandbox for crew."""

    _sandbox: Optional[Sandbox] = None

    @classmethod

    def get(cls) -> Sandbox:

        if cls._sandbox is None:

            cls._sandbox = Sandbox.create(template="code-interpreter", ttl=600)

        return cls._sandbox

    @classmethod

    def cleanup(cls):

        if cls._sandbox:

            cls._sandbox.kill()

            cls._sandbox = None

class CodeInput(BaseModel):

    code: str = Field(description="Python code to execute")

class AnalysisTool(BaseTool):

    name: str = "analyze"

    description: str = "Execute Python for data analysis. State persists."

    args_schema: Type[BaseModel] = CodeInput

    def _run(self, code: str) -> str:

        try:

            result = SandboxManager.get().runCode(code, language="python", timeout=60)

            return result.stdout if result.exitCode == 0 else f"Error: {result.stderr}"

        except Exception as e:

            return f"Error: {e}"

def create_market_research_crew():

    """Create a crew for market research analysis."""

    tool = AnalysisTool()

    # Agents

    market_analyst = Agent(

        role="Market Research Analyst",

        goal="Analyze market data to identify trends and opportunities",

        backstory="15 years of experience in market research and competitive analysis.",

        tools=[tool],

        verbose=True

    )

    quant_analyst = Agent(

        role="Quantitative Analyst",

        goal="Apply statistical methods to validate market insights",

        backstory="Former hedge fund quant with expertise in statistical modeling.",

        tools=[tool],

        verbose=True

    )

    strategist = Agent(

        role="Business Strategist",

        goal="Translate analysis into actionable business strategies",

        backstory="MBA from Wharton, 10 years in strategy consulting.",

        tools=[tool],

        verbose=True

    )

    # Tasks

    data_collection = Task(

        description="""Create a synthetic market dataset for analysis:

        Generate data with:

        - 500 companies

        - Columns: company, sector, revenue, growth_rate, market_share, employees, founded_year

        - Sectors: Tech, Healthcare, Finance, Retail, Manufacturing

        - Realistic distributions

        Save to /app/market_data.csv and show summary stats.""",

        expected_output="Dataset created with summary statistics",

        agent=market_analyst

    )

    statistical_analysis = Task(

        description="""Perform statistical analysis on market_data.csv:

        1. Correlation analysis between variables

        2. Sector comparison (ANOVA for revenue differences)

        3. Growth rate distribution analysis

        4. Identify statistical outliers

        5. Regression: what predicts growth_rate?

        Print all statistical results with interpretations.""",

        expected_output="Complete statistical analysis with interpretations",

        agent=quant_analyst,

        context=[data_collection]

    )

    strategic_report = Task(

        description="""Create strategic recommendations based on analysis:

        1. Executive Summary

        2. Key Market Insights (backed by data)

        3. Sector Opportunities (ranked)

        4. Risk Assessment

        5. Strategic Recommendations (5 specific actions)

        Reference specific numbers from the analysis.""",

        expected_output="Strategic report with data-backed recommendations",

        agent=strategist,

        context=[statistical_analysis]

    )

    return Crew(

        agents=[market_analyst, quant_analyst, strategist],

        tasks=[data_collection, statistical_analysis, strategic_report],

        process=Process.sequential,

        verbose=True

    )

if __name__ == "__main__":

    crew = create_market_research_crew()

    try:

        result = crew.kickoff()