# ⚠️ DANGEROUS - Don't use in production

from langchain_experimental.tools import PythonREPLTool

tool = PythonREPLTool()

result = tool.run("import os; os.system('rm -rf /')")  # Game over

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

│                    LangChain Agent                          │

│                                                             │

│  "I need to run this Python code to analyze the data..."   │

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

                              │

                              ▼

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

│                   HopX Sandbox Tool                         │

│                   (Custom LangChain Tool)                   │

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

                              │

                              ▼

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

│                    HopX MicroVM                             │

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

│  │  Isolated execution environment                     │    │

│  │  • No access to host filesystem                    │    │

│  │  • Network policies enforced                       │    │

│  │  • Resource limits applied                         │    │

│  │  • Destroyed after execution                       │    │

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

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

                              │

                              ▼

                    Results returned to agent

pip install langchain langchain-openai hopx-ai

export OPENAI_API_KEY="sk-..."

export HOPX_API_KEY="your-hopx-key"

from langchain.tools import BaseTool

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type, Optional

class PythonCodeInput(BaseModel):

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

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

class SecurePythonTool(BaseTool):

    """Execute Python code securely in an isolated HopX sandbox."""

    name: str = "python_executor"

    description: str = """Execute Python code in a secure, isolated sandbox.

Use this tool when you need to:

- Perform calculations or data analysis

- Process files or data structures

- Run any Python code safely

The sandbox has pandas, numpy, matplotlib, requests, and standard libraries.

For visualizations, save to /app/output.png.

"""

    args_schema: Type[BaseModel] = PythonCodeInput

    # Sandbox configuration

    template: str = "code-interpreter"

    timeout: int = 60

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

        """Execute code in isolated sandbox."""

        sandbox = None

        try:

            # Create isolated sandbox

            sandbox = Sandbox.create(template=self.template)

            # Execute code with timeout

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

            # Format output

            if result.exitCode == 0:

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

                return f"✅ Execution successful:\n{output}"

            else:

                error = result.stderr or "Unknown error"

                return f"❌ Execution failed:\n{error}"

        except Exception as e:

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

        finally:

            if sandbox:

                sandbox.kill()

    async def _arun(self, code: str) -> str:

        """Async version - runs sync for simplicity."""

        return self._run(code)

from langchain_openai import ChatOpenAI

from langchain.agents import AgentExecutor, create_openai_tools_agent

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Initialize the LLM

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Create the secure tool

python_tool = SecurePythonTool()

# Define the prompt

prompt = ChatPromptTemplate.from_messages([

    ("system", """You are a helpful AI assistant with access to a Python executor.

When users ask questions that require computation, data analysis, or code execution:

1. Write clear, well-commented Python code

2. Use the python_executor tool to run it

3. Analyze the results and provide a helpful response

Available libraries: pandas, numpy, matplotlib, seaborn, requests, json, csv, datetime

Tips:

- Always print() results you want to see

- For charts, save to /app/output.png using plt.savefig()

- Handle potential errors in your code

"""),

    MessagesPlaceholder(variable_name="chat_history", optional=True),

    ("human", "{input}"),

    MessagesPlaceholder(variable_name="agent_scratchpad"),

])

# Create the agent

agent = create_openai_tools_agent(llm, [python_tool], prompt)

agent_executor = AgentExecutor(

    agent=agent, 

    tools=[python_tool], 

    verbose=True,

    max_iterations=5

)

# Run it

response = agent_executor.invoke({

    "input": "Calculate the first 50 prime numbers and their sum"

})

print(response["output"])

from langchain.tools import BaseTool

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type

class BashCommandInput(BaseModel):

    """Input for bash commands."""

    command: str = Field(description="The bash command to execute")

class SecureBashTool(BaseTool):

    """Execute bash commands in isolated sandbox."""

    name: str = "bash_executor"

    description: str = """Execute bash/shell commands securely.

Use for: file operations, system commands, package installation.

Example: ls -la, cat file.txt, pip install package

"""

    args_schema: Type[BaseModel] = BashCommandInput

    template: str = "code-interpreter"

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

        sandbox = None

        try:

            sandbox = Sandbox.create(template=self.template)

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

            if result.exitCode == 0:

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

            else:

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

        except Exception as e:

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

        finally:

            if sandbox:

                sandbox.kill()

class FileReadInput(BaseModel):

    """Input for reading files."""

    path: str = Field(description="Path to the file to read")

class SecureFileReadTool(BaseTool):

    """Read files from the sandbox."""

    name: str = "read_file"

    description: str = "Read the contents of a file. Use after creating or downloading files."

    args_schema: Type[BaseModel] = FileReadInput

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

        sandbox = None

        try:

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

            content = sandbox.files.read(path)

            return f"File contents:\n{content[:10000]}"  # Truncate large files

        except Exception as e:

            return f"❌ Could not read file: {str(e)}"

        finally:

            if sandbox:

                sandbox.kill()

# Create multi-tool agent

tools = [

    SecurePythonTool(),

    SecureBashTool(),

    SecureFileReadTool(),

]

agent = create_openai_tools_agent(llm, tools, prompt)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

from langchain.tools import BaseTool

from hopx import Sandbox

from typing import Optional

import threading

class PersistentSandboxManager:

    """Manage a persistent sandbox for multi-step execution."""

    _instance: Optional['PersistentSandboxManager'] = None

    _lock = threading.Lock()

    def __init__(self):

        self.sandbox: Optional[Sandbox] = None

        self.ttl = 300  # 5 minutes

    @classmethod

    def get_instance(cls) -> 'PersistentSandboxManager':

        if cls._instance is None:

            with cls._lock:

                if cls._instance is None:

                    cls._instance = cls()

        return cls._instance

    def get_sandbox(self) -> Sandbox:

        """Get or create sandbox."""

        if self.sandbox is None:

            self.sandbox = Sandbox.create(

                template="code-interpreter",

                ttl=self.ttl

            )

        return self.sandbox

    def reset(self):

        """Destroy and recreate sandbox."""

        if self.sandbox:

            try:

                self.sandbox.kill()

            except:

                pass

        self.sandbox = None

class PersistentPythonTool(BaseTool):

    """Python tool with persistent sandbox state."""

    name: str = "python"

    description: str = """Execute Python code with persistent state.

Variables and imports persist between calls.

Use for multi-step data analysis where you need to build on previous results.

"""

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

        manager = PersistentSandboxManager.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:

            # Sandbox might have expired, reset and retry

            manager.reset()

            return f"Sandbox error (will retry with fresh sandbox): {str(e)}"

from langchain_openai import ChatOpenAI

from langchain.agents import AgentExecutor, create_openai_tools_agent

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain.tools import BaseTool

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type, Optional

import base64

class DataAnalysisTool(BaseTool):

    """Comprehensive data analysis tool with file support."""

    name: str = "analyze_data"

    description: str = """Analyze data using Python in a secure sandbox.

Capabilities:

- Load CSV, JSON, Excel files

- Statistical analysis with pandas

- Visualizations with matplotlib/seaborn

- Machine learning with scikit-learn

Input your Python code. For charts, save to /app/chart.png.

Uploaded files are available at /app/data/

"""

    # Keep sandbox alive for the session

    sandbox: Optional[Sandbox] = None

    def get_sandbox(self) -> Sandbox:

        if self.sandbox is None:

            self.sandbox = Sandbox.create(

                template="code-interpreter",

                ttl=300  # 5 minute TTL

            )

        return self.sandbox

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

        try:

            sandbox = self.get_sandbox()

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

            output_parts = []

            if result.stdout:

                output_parts.append(f"Output:\n{result.stdout}")

            if result.stderr and result.exitCode != 0:

                output_parts.append(f"Error:\n{result.stderr}")

            # Check if a chart was created

            try:

                chart_data = sandbox.files.read("/app/chart.png")

                output_parts.append("\n📊 Chart saved to /app/chart.png")

            except:

                pass

            return "\n\n".join(output_parts) or "Code executed successfully"

        except Exception as e:

            self.sandbox = None  # Reset on error

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

    def upload_data(self, filename: str, content: bytes):

        """Upload data file to sandbox."""

        sandbox = self.get_sandbox()

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

    def cleanup(self):

        """Destroy sandbox when done."""

        if self.sandbox:

            self.sandbox.kill()

            self.sandbox = None

# Create the analysis agent

llm = ChatOpenAI(model="gpt-4o", temperature=0)

data_tool = DataAnalysisTool()

analysis_prompt = ChatPromptTemplate.from_messages([

    ("system", """You are an expert data analyst assistant.

When users provide data or ask analytical questions:

1. First explore the data structure (head, info, describe)

2. Perform the requested analysis

3. Create visualizations when appropriate

4. Explain your findings clearly

Always show your work with code. Use pandas for data manipulation.

Save visualizations to /app/chart.png using plt.savefig('/app/chart.png', dpi=150, bbox_inches='tight')

"""),

    ("human", "{input}"),

    MessagesPlaceholder(variable_name="agent_scratchpad"),

])

agent = create_openai_tools_agent(llm, [data_tool], analysis_prompt)

data_agent = AgentExecutor(agent=agent, tools=[data_tool], verbose=True)

# Example: Multi-step analysis

response = data_agent.invoke({

    "input": """Create a sample sales dataset with:

    - 500 rows

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

    - Random but realistic data

    Then:

    1. Show basic statistics

    2. Find top 5 products by revenue

    3. Create a bar chart of revenue by region

    """

})

print(response["output"])

# Cleanup

data_tool.cleanup()

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

from langchain_core.output_parsers import StrOutputParser

from langchain_openai import ChatOpenAI

from hopx import Sandbox

def execute_code_safely(code: str) -> str:

    """Execute code in sandbox and return result."""

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

    try:

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

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

    finally:

        sandbox.kill()

# Build an LCEL chain that generates and executes code

generate_code_prompt = ChatPromptTemplate.from_messages([

    ("system", "You are a Python expert. Generate ONLY executable Python code, no explanations."),

    ("human", "Write Python code to: {task}")

])

explain_result_prompt = ChatPromptTemplate.from_messages([

    ("system", "Explain the following code execution result in plain English."),

    ("human", "Task: {task}\n\nCode result:\n{result}")

])

llm = ChatOpenAI(model="gpt-4o")

# Chain: Generate code → Execute → Explain

chain = (

    {"task": RunnablePassthrough()}

    | RunnablePassthrough.assign(

        code=generate_code_prompt | llm | StrOutputParser()

    )

    | RunnablePassthrough.assign(

        result=lambda x: execute_code_safely(x["code"])

    )

    | explain_result_prompt

    | llm

    | StrOutputParser()

)

# Run it

result = chain.invoke("Calculate the factorial of 20 and check if it's divisible by 7")

print(result)

from langchain.tools import BaseTool

from hopx import Sandbox

from typing import Optional

import time

class RobustPythonTool(BaseTool):

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

    name: str = "python"

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

    max_retries: int = 3

    retry_delay: float = 1.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 (no output)"

                else:

                    # Code error - don't retry, return error for LLM to 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)

                    continue

            finally:

                if sandbox:

                    try:

                        sandbox.kill()

                    except:

                        pass

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

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

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

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

    output = result.stdout[:10000]  # Truncate large outputs

    return output

# For data science tasks

sandbox = Sandbox.create(template="data-science")

# For web scraping

sandbox = Sandbox.create(template="web-scraper")

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

agent_executor = AgentExecutor(

    agent=agent,

    tools=tools,

    memory=memory,

    verbose=True

)

import logging

logger = logging.getLogger("secure_python_tool")

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

    logger.info(f"Executing code: {code[:100]}...")

    result = self._execute(code)

    logger.info(f"Result: {result[:100]}...")

    return result

"""

Secure LangChain Agent with HopX Code Execution

"""

from langchain_openai import ChatOpenAI

from langchain.agents import AgentExecutor, create_openai_tools_agent

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain.tools import BaseTool

from langchain.memory import ConversationBufferMemory

from hopx import Sandbox

from pydantic import BaseModel, Field

from typing import Type, Optional

import os

# Ensure API keys are set

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

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

class CodeInput(BaseModel):

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

class SecurePythonExecutor(BaseTool):

    name: str = "execute_python"

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

Available: pandas, numpy, matplotlib, seaborn, scikit-learn, requests.

For charts: plt.savefig('/app/chart.png')

Print results you want to see."""

    args_schema: Type[BaseModel] = CodeInput

    sandbox: Optional[Sandbox] = None

    def get_or_create_sandbox(self) -> Sandbox:

        if self.sandbox is None:

            self.sandbox = Sandbox.create(template="code-interpreter", ttl=300)

        return self.sandbox

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

        try:

            sandbox = self.get_or_create_sandbox()

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

            if result.exitCode == 0:

                return result.stdout or "✅ Executed successfully"

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

        except Exception as e:

            self.sandbox = None

            return f"❌ Sandbox error: {e}"

    def cleanup(self):

        if self.sandbox:

            self.sandbox.kill()

            self.sandbox = None

def create_secure_agent():

    """Create a LangChain agent with secure code execution."""

    llm = ChatOpenAI(model="gpt-4o", temperature=0)

    tool = SecurePythonExecutor()

    prompt = ChatPromptTemplate.from_messages([

        ("system", """You are a helpful AI assistant that can execute Python code safely.

When you need to compute, analyze data, or run code:

1. Write clear Python code

2. Use the execute_python tool

3. Explain the results

Be concise and helpful."""),

        MessagesPlaceholder(variable_name="chat_history", optional=True),

        ("human", "{input}"),

        MessagesPlaceholder(variable_name="agent_scratchpad"),

    ])

    agent = create_openai_tools_agent(llm, [tool], prompt)