Code Execution Skill

You are an expert in secure sandboxed code execution for AI agents, implementing

the patterns used by OpenAI Code Interpreter, Anthropic's tool use, and E2B cloud sandboxes.

You write safe, production-ready execution environments in British English.

IMI colours: Navy #1A1A2E · Teal #0F3D66 · Gold #E2B95A

Part 1 — E2B Cloud Sandboxes (Production Grade)

# pip install e2b-code-interpreter

from e2b_code_interpreter import CodeInterpreter
import os


def run_in_e2b_sandbox(code: str, timeout: int = 30) -> dict:
    """
    Run Python code in an isolated E2B cloud sandbox.
    No risk to local system — full Python environment available.
    Returns stdout, stderr, and any generated files.
    """
    with CodeInterpreter(api_key=os.getenv("E2B_API_KEY")) as sandbox:
        execution = sandbox.notebook.exec_cell(code, timeout=timeout)

        result = {
            "stdout": execution.text or "",
            "stderr": "\n".join([str(e) for e in execution.error]) if execution.error else "",
            "results": [],
            "success": execution.error is None
        }

        # Extract output files / charts
        for output in execution.results:
            if hasattr(output, "png"):
                result["results"].append({"type": "image", "data": output.png})
            elif hasattr(output, "text"):
                result["results"].append({"type": "text", "data": output.text})

        return result


def e2b_install_packages(sandbox_code: str, packages: list[str]) -> str:
    """Pre-install packages in E2B before running analysis code."""
    install_code = f"!pip install {' '.join(packages)} -q\n\n{sandbox_code}"
    return install_code


def imi_data_analysis_agent(
    dataset_csv: str,
    analysis_request: str
) -> dict:
    """
    AI agent that writes and executes data analysis code in E2B sandbox.
    Safe: runs in isolated cloud container, never touches local files.
    """
    import anthropic
    client = anthropic.Anthropic()

    # Ask Claude to write analysis code
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1500,
        system=(
            "You are a Python data analyst for IMI sports fan research. "
            "Write clean, runnable Python code to answer the analysis request. "
            "Assume pandas and matplotlib are available. "
            "The CSV data is already loaded as df = pd.read_csv('data.csv'). "
            "Use British English in outputs. Output ONLY the Python code, no explanation."
        ),
        messages=[{"role": "user", "content": f"Dataset columns: {dataset_csv[:200]}\n\nAnalysis request: {analysis_request}"}]
    )

    code = response.content[0].text.strip().strip("```python").strip("```").strip()

    # Prepend data loading
    full_code = f"""
import pandas as pd
import matplotlib.pyplot as plt
import io

csv_data = '''{dataset_csv}'''
df = pd.read_csv(io.StringIO(csv_data))

{code}
"""
    return run_in_e2b_sandbox(full_code)

Part 2 — Subprocess Execution with Timeout & Safety

import subprocess, tempfile, os, signal, resource
from pathlib import Path


def safe_python_exec(
    code: str,
    timeout: int = 15,
    max_output_bytes: int = 10_000
) -> dict:
    """
    Execute Python code in a subprocess with strict timeout and output limits.
    Safer than exec() — runs in separate process with resource limits.
    """
    with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
        f.write(code)
        temp_path = f.name

    try:
        result = subprocess.run(
            ["python3", temp_path],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        stdout = result.stdout[:max_output_bytes]
        stderr = result.stderr[:max_output_bytes]
        return {
            "stdout": stdout,
            "stderr": stderr,
            "return_code": result.returncode,
            "success": result.returncode == 0
        }
    except subprocess.TimeoutExpired:
        return {"stdout": "", "stderr": f"Execution timed out after {timeout}s", "success": False}
    finally:
        os.unlink(temp_path)


# ── Dangerous patterns to block before executing ─────────────────────────────
BLOCKED_PATTERNS = [
    "import os", "import sys", "import subprocess",
    "__import__", "open(", "exec(", "eval(",
    "shutil", "socket", "requests", "urllib",
    "rm -rf", "os.remove", "os.system",
    "pathlib", "glob"
]

def is_safe_code(code: str) -> tuple[bool, str]:
    """Check code for dangerous patterns before execution."""
    for pattern in BLOCKED_PATTERNS:
        if pattern in code:
            return False, f"Blocked pattern detected: '{pattern}'"
    return True, ""


def sandboxed_exec(code: str, timeout: int = 10) -> dict:
    """Safety check + subprocess execution."""
    safe, reason = is_safe_code(code)
    if not safe:
        return {"stdout": "", "stderr": reason, "success": False, "blocked": True}
    return safe_python_exec(code, timeout=timeout)

Part 3 — Python REPL Pattern for Agents

from io import StringIO
import sys, traceback, contextlib


@contextlib.contextmanager
def capture_output():
    """Context manager to capture stdout/stderr."""
    old_stdout, old_stderr = sys.stdout, sys.stderr
    sys.stdout = StringIO()
    sys.stderr = StringIO()
    try:
        yield sys.stdout, sys.stderr
    finally:
        sys.stdout = old_stdout
        sys.stderr = old_stderr


class PythonREPL:
    """
    Persistent Python REPL for agents.
    Variables persist across calls — like a Jupyter notebook.
    WARNING: Only use with trusted code (no user input).
    """

    def __init__(self):
        self.globals = {
            "pd": __import__("pandas"),
            "np": __import__("numpy"),
            "__builtins__": __builtins__
        }

    def run(self, code: str) -> dict:
        with capture_output() as (stdout, stderr):
            error = None
            try:
                exec(compile(code, "<repl>", "exec"), self.globals)
            except Exception:
                error = traceback.format_exc()

        return {
            "stdout": stdout.getvalue(),
            "stderr": stderr.getvalue(),
            "error": error,
            "success": error is None
        }

    def run_expression(self, expr: str):
        """Evaluate a single expression and return its value."""
        try:
            return eval(expr, self.globals)
        except Exception as e:
            return f"Error: {e}"


# ── Agent tool definition for Claude tool use ─────────────────────────────────
PYTHON_REPL_TOOL = {
    "name": "python_repl",
    "description": (
        "Execute Python code for data analysis, calculations, and visualisation. "
        "pandas (pd), numpy (np) are pre-imported. "
        "Use for: statistical analysis, data manipulation, chart generation. "
        "Do NOT use for file system operations or network requests."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "code": {
                "type": "string",
                "description": "Python code to execute"
            }
        },
        "required": ["code"]
    }
}


def run_code_tool(tool_input: dict, repl: PythonREPL) -> str:
    """Handle python_repl tool call from Claude."""
    code = tool_input.get("code", "")
    result = repl.run(code)
    if result["error"]:
        return f"Error:\n{result['error']}"
    output = result["stdout"]
    if not output:
        return "Code executed successfully (no output)"
    return output[:2000]  # limit output to 2000 chars

Part 4 — Code Interpreter Agent Loop

import anthropic, json


def code_interpreter_agent(
    task: str,
    data: str | None = None,
    max_iterations: int = 5
) -> str:
    """
    Full code interpreter agent loop.
    The AI writes code → executes → sees output → iterates until done.
    """
    client = anthropic.Anthropic()
    repl = PythonREPL()

    # Seed with data if provided
    if data:
        repl.run(f"import io, pandas as pd\ndf = pd.read_csv(io.StringIO({repr(data)}))")

    messages = [{"role": "user", "content": task}]

    for _ in range(max_iterations):
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=2000,
            tools=[PYTHON_REPL_TOOL],
            system=(
                "You are an IMI data analysis agent. Use the python_repl tool to "
                "compute, analyse, and verify your work. Use British English."
            ),
            messages=messages
        )

        # Collect text output
        text_parts = [b.text for b in response.content if hasattr(b, "text")]

        if response.stop_reason == "end_turn":
            return "\n".join(text_parts)

        # Handle tool use
        tool_uses = [b for b in response.content if b.type == "tool_use"]
        if not tool_uses:
            return "\n".join(text_parts)

        # Execute all tool calls
        messages.append({"role": "assistant", "content": response.content})
        tool_results = []
        for tool_use in tool_uses:
            output = run_code_tool(tool_use.input, repl)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": output
            })

        messages.append({"role": "user", "content": tool_results})

    return "Max iterations reached"

Output Standards

• Production code: use E2B sandboxes — never exec() on untrusted input
• Dev/trusted workflows: PythonREPL with pre-vetted code is acceptable
• Timeout: always set timeout (10-30s) to prevent runaway execution
• Output limits: always cap stdout at 10KB to prevent memory issues
• Safety check: run is_safe_code() before any subprocess execution
• British English in all agent outputs and error messages

pip install

pip install e2b-code-interpreter anthropic pandas numpy

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)