Code Execution Agents with E2B

The Code Execution Agent Pattern

Code execution transforms an agent from a text processor into a genuine problem solver. Instead of describing how to analyze data, the agent writes Python, runs it, sees the output, and iterates.

flowchart TD
    U(["User: Analyze this CSV\nfind top 5 customers by revenue"]) --> AG["AGENT\ngenerates Python code\nimport pandas; groupby; nlargest"]
    AG --> SB["E2B SANDBOX\nexecutes code in isolation"]
    SB -->|"stdout: {Acme: 150000...}"| OUT([AGENT formats response])
    SB -->|error traceback| AG
    style AG fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style SB fill:#f0fdf9,stroke:#0D9488,color:#0F172A

The agent can also handle errors by reading the traceback and rewriting the code.

E2B: Managed Sandbox Service

pip install e2b-code-interpreter

SandboxExecutor wraps the E2B API so every code execution happens in an isolated cloud VM, no LLM-generated code ever touches your host machine.

# src/tools/code_executor.py
from e2b_code_interpreter import Sandbox
import asyncio

class SandboxExecutor:
    def __init__(self):
        self.api_key = settings.e2b_api_key
    
    async def run_python(
        self,
        code: str,
        timeout_seconds: int = 30,
        files: dict[str, bytes] = None,
    ) -> dict:
        """Execute Python code in an isolated sandbox."""
        
        # Sandbox() creates an isolated cloud VM: code runs there, not on your host machine
        sandbox = Sandbox(api_key=self.api_key)
        
        try:
            # Upload any data files the generated code needs to read (e.g., the user's CSV)
            if files:
                for filename, content in files.items():
                    sandbox.filesystem.write(filename, content)
            
            # Run the code; timeout_seconds caps execution to prevent runaway loops or infinite computation
            execution = sandbox.run_code(code, timeout=timeout_seconds)
            
            return {
                "success": not execution.error,
                "stdout": execution.text or "",
                "stderr": execution.error or "",  # error message if the code raised an exception
                "results": [
                    {
                        "type": r.type,
                        # Exclude raw image bytes from the dict: just flag that an image was produced
                        "data": r.data if r.type != "image/png" else "<image>",
                    }
                    for r in (execution.results or [])
                ],
                "execution_count": 1,
            }
        
        except Exception as e:
            return {
                "success": False,
                "stdout": "",
                "stderr": str(e),
                "results": [],
            }
        finally:
            # sandbox.kill() is critical: forgetting this leaves running VMs that cost money
            sandbox.kill()

executor = SandboxExecutor()

Data Analysis Agent

The graph below implements an iterative generate-execute-correct loop: the LLM writes code, the sandbox runs it, and the output (or error traceback) feeds back to the LLM for the next iteration.

# src/agents/data_analyst_agent.py
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

class AnalystState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    code_iterations: int       # tracks how many code attempts have been made
    data_file: bytes | None
    last_code: str | None      # the most recent code block extracted from the LLM response
    last_output: str | None    # stdout from the last successful execution
    analysis_complete: bool

# MAX_CODE_ITERATIONS=4 prevents the agent from looping indefinitely on a broken task
MAX_CODE_ITERATIONS = 4

ANALYST_SYSTEM = """You are a Python data analyst. When given data analysis tasks:
1. Write clean, complete Python code that solves the problem
2. Use pandas, matplotlib, and numpy as needed
3. Always print() your results clearly
4. Handle edge cases (missing data, wrong types)
5. If you get an error, read it carefully and fix the specific issue

When writing code, put it in a ```python code block.
After seeing execution results, interpret them for the user."""

def extract_code_from_response(text: str) -> str | None:
    """Extract Python code from markdown code blocks."""
    import re
    # extract_code_from_response uses regex rather than JSON parsing because LLMs sometimes wrap code in prose
    match = re.search(r'```python\n(.*?)```', text, re.DOTALL)
    return match.group(1).strip() if match else None

async def analyst_llm_node(state: AnalystState) -> dict:
    """LLM generates or refines code."""
    # Bail out gracefully once the iteration cap is hit: more attempts rarely help
    if state["code_iterations"] >= MAX_CODE_ITERATIONS:
        return {
            "messages": [AIMessage(content="I've reached the maximum number of code iterations. Here are my findings so far: " + (state["last_output"] or "No output generated."))],
            "analysis_complete": True,
        }
    
    response = llm.invoke(
        [SystemMessage(content=ANALYST_SYSTEM)] + state["messages"]
    )
    return {
        "messages": [response],
        # Parse code from the response so the execute node receives clean Python, not markdown
        "last_code": extract_code_from_response(response.content),
    }

async def execute_code_node(state: AnalystState) -> dict:
    """Execute the generated Python code in a sandbox."""
    if not state["last_code"]:
        return {}
    
    files = {}
    if state["data_file"]:
        # Pass the uploaded file into the sandbox under a predictable name the code can open
        files["data.csv"] = state["data_file"]
    
    result = await executor.run_python(
        code=state["last_code"],
        timeout_seconds=30,
        files=files,
    )
    
    # Format result as an observation for the LLM: success path shows output, failure path shows the traceback
    if result["success"]:
        output = f"Code executed successfully.\nOutput:\n{result['stdout']}"
        if result["results"]:
            output += f"\nGenerated {len(result['results'])} output objects"
    else:
        output = f"Code failed with error:\n{result['stderr']}"
    
    # Wrap the result in a HumanMessage so it enters the conversation as the next "user turn"
    observation_msg = HumanMessage(content=f"Code execution result:\n{output}")
    
    return {
        "messages": [observation_msg],
        "code_iterations": state["code_iterations"] + 1,
        "last_output": result["stdout"],
    }

def should_execute_or_finish(state: AnalystState) -> str:
    """Route: does the last message contain code to execute?"""
    if state.get("analysis_complete"):
        return "finish"
    # If the LLM produced a code block, run it; otherwise the analysis is a prose-only answer
    if state["last_code"]:
        return "execute"
    return "finish"

# Build the graph
analyst_builder = StateGraph(AnalystState)
analyst_builder.add_node("llm", analyst_llm_node)
analyst_builder.add_node("execute", execute_code_node)

analyst_builder.set_entry_point("llm")
analyst_builder.add_conditional_edges("llm", should_execute_or_finish, {
    "execute": "execute",
    "finish": END,
})
# After each execution the result re-enters the LLM node so the model can react to the output
analyst_builder.add_edge("execute", "llm")

data_analyst = analyst_builder.compile()

Running the Data Analyst

import asyncio

async def analyze_data(question: str, csv_content: bytes = None):
    # All state fields must be initialized: LangGraph does not supply defaults
    initial_state: AnalystState = {
        "messages": [HumanMessage(content=question)],
        "code_iterations": 0,    # counter starts at zero; incremented by execute_code_node
        "data_file": csv_content,
        "last_code": None,
        "last_output": None,
        "analysis_complete": False,
    }
    
    result = await data_analyst.ainvoke(initial_state)
    
    # Filter to AI messages only: the list also contains the human turns and execution observations
    ai_messages = [m for m in result["messages"] if isinstance(m, AIMessage)]
    return ai_messages[-1].content  # return the final interpretation, not intermediate code drafts

# Usage
with open("sales_data.csv", "rb") as f:
    csv_data = f.read()

analysis = await analyze_data(
    "What are the top 3 products by revenue? Show monthly trends.",
    csv_content=csv_data,
)
print(analysis)

Modal for GPU Workloads

For agents that need GPU acceleration (image generation, ML inference):

# src/tools/modal_executor.py
import modal

# modal.App groups all GPU functions under one deployment: referenced together in `modal deploy`
app = modal.App("agent-gpu-tools")

# @app.function provisions a fresh GPU container on demand: no GPU sits idle between calls
@app.function(
    gpu="A10G",                  # A10G is cost-effective for Stable Diffusion; use A100 for larger models
    image=modal.Image.debian_slim().pip_install(["torch", "diffusers", "transformers"]),
    timeout=120,                 # Modal kills the container after 120 s, prevents runaway GPU spend
)
def generate_image(prompt: str, negative_prompt: str = "") -> bytes:
    """Generate an image using Stable Diffusion on a GPU."""
    from diffusers import StableDiffusionPipeline
    import torch
    
    pipe = StableDiffusionPipeline.from_pretrained(
        "runwayml/stable-diffusion-v1-5",
        torch_dtype=torch.float16,  # float16 halves VRAM usage with negligible quality loss on A10G
    ).to("cuda")
    
    image = pipe(prompt, negative_prompt=negative_prompt).images[0]
    
    # Serialize the PIL image to PNG bytes so it can cross the Modal RPC boundary
    import io
    buffer = io.BytesIO()
    image.save(buffer, format="PNG")
    return buffer.getvalue()

# Use as a tool in your agent
async def generate_image_tool(prompt: str) -> dict:
    """Generate an image with Stable Diffusion on a GPU."""
    with modal.enable_output():
        # .remote() sends execution to the Modal cloud GPU; the calling process waits for the result
        image_bytes = generate_image.remote(prompt)
    
    # Save locally: in production you'd upload to S3/GCS and return a signed URL instead
    filename = f"generated_{uuid.uuid4().hex[:8]}.png"
    with open(f"/tmp/{filename}", "wb") as f:
        f.write(image_bytes)
    
    return {"filename": filename, "size_bytes": len(image_bytes)}

Code execution agents, especially data analysts and debugging assistants: are among the most valuable AI agent use cases because they can do things LLMs fundamentally cannot: run code, see real outputs, and iterate to correct answers.