Building a code-executing agent in Haystack 2.0

This tutorial walks through building a code-executing agent in Haystack 2.0 from scratch. You’ll wire up a tool that runs Python in a sandboxed subprocess, hook it into the agent loop, and add the guardrails that keep arbitrary code from eating your server. The result is a reusable component you can drop into any Haystack pipeline.

Step 1: Install dependencies and set up the environment

Start with a clean virtual environment. Haystack 2.0 requires Python 3.10+.

python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install "haystack-ai>=2.0.0" "ipykernel>=6.29" "pydantic>=2.0"

If you plan to use an LLM via OpenRouter or another OpenAI-compatible endpoint, set your API key now:

export OPENROUTER_API_KEY="sk-or-..."
export OPENROUTER_BASE_URL="https://openrouter.ai/api/v1"

Create a project layout:

code_agent/
├── main.py
├── tools/
│   ├── __init__.py
│   └── python_tool.py
├── sandbox/
│   ├── __init__.py
│   └── executor.py
└── requirements.txt

Step 2: Build a secure Python executor

Never exec() untrusted code in the main process. Use a subprocess with resource limits, a stripped environment, and no network access. The executor below writes the snippet to a temp file, runs it under python -I (isolated mode), and enforces a timeout.

# sandbox/executor.py
import subprocess
import tempfile
import os
import signal
import shlex
from dataclasses import dataclass
from typing import Optional

@dataclass
class ExecutionResult:
    stdout: str
    stderr: str
    exit_code: int
    timed_out: bool

class PythonExecutor:
    def __init__(
        self,
        timeout_seconds: int = 30,
        max_memory_mb: int = 512,
        allowed_imports: Optional[list[str]] = None,
    ):
        self.timeout_seconds = timeout_seconds
        self.max_memory_mb = max_memory_mb
        self.allowed_imports = allowed_imports or []

    def _build_command(self, script_path: str) -> list[str]:
        # -I: isolated mode (no PYTHONPATH, no site-packages, no .pyc)
        # -S: don't import site (faster startup, less attack surface)
        # -B: don't write .pyc files
        return ["python3", "-I", "-S", "-B", script_path]

    def _preexec_fn(self):
        # Limit CPU time (seconds) and virtual memory (bytes)
        import resource
        resource.setrlimit(resource.RLIMIT_CPU, (self.timeout_seconds, self.timeout_seconds))
        resource.setrlimit(
            resource.RLIMIT_AS,
            (self.max_memory_mb * 1024 * 1024, self.max_memory_mb * 1024 * 1024),
        )
        # Drop privileges if running as non-root user
        # os.setuid(65534)  # nobody user — enable in production

    def run(self, code: str) -> ExecutionResult:
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".py", delete=False, prefix="agent_"
        ) as f:
            f.write(code)
            script_path = f.name

        try:
            cmd = self._build_command(script_path)
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.timeout_seconds + 2,  # buffer for process overhead
                preexec_fn=self._preexec_fn,
                env={"PATH": "/usr/bin:/bin", "HOME": "/tmp"},  # minimal env
            )
            return ExecutionResult(
                stdout=proc.stdout,
                stderr=proc.stderr,
                exit_code=proc.returncode,
                timed_out=False,
            )
        except subprocess.TimeoutExpired as e:
            return ExecutionResult(
                stdout=e.stdout or "",
                stderr=e.stderr or "Execution timed out",
                exit_code=-1,
                timed_out=True,
            )
        finally:
            try:
                os.unlink(script_path)
            except OSError:
                pass

Verification: Run a quick sanity check in a REPL:

from sandbox.executor import PythonExecutor
ex = PythonExecutor(timeout_seconds=5)
result = ex.run("print('hello'); import sys; print(sys.version)")
assert result.exit_code == 0
assert "hello" in result.stdout
print("Executor works")

Step 3: Wrap the executor as a Haystack tool

Haystack 2.0 tools are callables with a @tool decorator or classes implementing __call__ with a JSON-serializable signature. We’ll use the decorator for clarity and add input validation via Pydantic.

# tools/python_tool.py
from haystack import tool
from pydantic import BaseModel, Field
from sandbox.executor import PythonExecutor, ExecutionResult

_executor = PythonExecutor(timeout_seconds=30, max_memory_mb=512)

class PythonToolInput(BaseModel):
    code: str = Field(
        description="Python code to execute. Must be a complete script. "
                    "Use print() to return values. No network, no filesystem access beyond /tmp."
    )

@tool(name="python_executor", description="Execute Python code in a sandboxed subprocess. Returns stdout, stderr, and exit code.")
def python_executor(code: str) -> dict:
    """
    Run arbitrary Python code safely. The sandbox enforces:
    - 30 second CPU timeout
    - 512 MB memory limit
    - No network access
    - No access to parent process environment
    - Isolated Python interpreter (-I -S -B flags)
    """
    result: ExecutionResult = _executor.run(code)
    return {
        "stdout": result.stdout,
        "stderr": result.stderr,
        "exit_code": result.exit_code,
        "timed_out": result.timed_out,
    }

Verification: Test the tool directly:

from tools.python_tool import python_executor
out = python_executor.invoke({"code": "x = 2 + 2; print(f'result: {x}')"})
assert out["exit_code"] == 0
assert "result: 4" in out["stdout"]
print("Tool invocation works")

Step 4: Create the agent with a system prompt that teaches tool use

Haystack’s Agent class (introduced in 2.0) expects a ChatGenerator and a list of tools. The system prompt is where you define the agent’s behavior — how it should think, when to call tools, and how to format responses.

# main.py
import os
from haystack import Pipeline
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from tools.python_tool import python_executor

SYSTEM_PROMPT = """
You are a code-executing assistant. You have access to a Python sandbox via the `python_executor` tool.

Rules:
1. When the user asks a question that requires computation, data transformation, or verification, write Python code and call the tool.
2. The tool returns stdout, stderr, and exit_code. Read stdout for results. If exit_code != 0, debug and retry.
3. Always print() your final answer so the tool captures it. Do not rely on implicit returns.
4. You can import standard library modules (json, math, statistics, datetime, itertools, collections, etc.).
5. No network requests, no filesystem access outside /tmp, no subprocess spawning.
6. If the user asks for code, provide it in a markdown block AND offer to execute it.
7. Think step by step. For multi-step problems, write one script that does everything and run it once.
"""

def build_agent(model: str = "openai/gpt-4o-mini") -> Agent:
    generator = OpenAIChatGenerator(
        model=model,
        api_key=os.getenv("OPENROUTER_API_KEY"),
        api_base_url=os.getenv("OPENROUTER_BASE_URL"),
        generation_kwargs={"temperature": 0.1, "max_tokens": 4096},
    )
    return Agent(
        chat_generator=generator,
        tools=[python_executor],
        system_prompt=SYSTEM_PROMPT,
        max_agent_steps=10,
    )

Note: If you’re using a local model via Ollama or vLLM, swap OpenAIChatGenerator for OllamaChatGenerator or OpenAIChatGenerator with a local base URL. The agent interface stays the same.

Step 5: Wire the agent into a pipeline for observability

A bare Agent works, but wrapping it in a Pipeline lets you add logging, tracing, and fallback components later.

# main.py (continued)
from haystack.components.others import Multiplexer
from haystack.components.preprocessors import DocumentCleaner
from haystack import component
from typing import Any

@component
class AgentLogger:
    @component.output_types(response=str)
    def run(self, messages: list[ChatMessage]) -> dict:
        last = messages[-1]
        print(f"[AGENT] {last.role.value}: {last.text[:200]}...")
        return {"response": last.text or ""}

def build_pipeline() -> Pipeline:
    agent = build_agent()
    pipe = Pipeline()
    pipe.add_component("agent", agent)
    pipe.add_component("logger", AgentLogger())
    pipe.connect("agent.replies", "logger.messages")
    return pipe

Step 6: Run the agent and verify end-to-end behavior

Create a small driver script that exercises the agent with a few test cases.

# main.py (driver)
def main():
    pipe = build_pipeline()
    
    test_cases = [
        "What is the 100th Fibonacci number?",
        "Calculate the standard deviation of [12, 15, 14, 10, 18, 20, 22, 17]",
        "Write a quicksort implementation and test it on a random list of 50 integers",
        "Parse this JSON and extract all email addresses: "
        '{"users": [{"name": "Alice", "email": "alice@example.com"}, '
        '{"name": "Bob", "email": "bob@test.org"}]}',
    ]

    for i, query in enumerate(test_cases, 1):
        print(f"\n{'='*60}")
        print(f"Test {i}: {query}")
        print(f"{'='*60}")
        
        result = pipe.run({"agent": {"messages": [ChatMessage.from_user(query)]}})
        print(f"\nFinal answer:\n{result['logger']['response']}")

if __name__ == "__main__":
    main()

Run it:

python main.py

Verification checklist:

• Each test prints a final answer with the computed result
• No exit_code: -1 (timeout) appears in tool output
• Agent completes within max_agent_steps (default 10)
• Memory stays bounded — watch top while running the quicksort test

Step 7: Add streaming for real-time feedback

Long-running code benefits from streaming the agent’s reasoning. Haystack 2.0 supports streaming via the streaming_callback on the generator.

# main.py (streaming variant)
import sys
from haystack.components.generators.chat import OpenAIChatGenerator

def build_streaming_agent(model: str = "openai/gpt-4o-mini") -> Agent:
    def stream_handler(chunk: str) -> None:
        sys.stdout.write(chunk)
        sys.stdout.flush()

    generator = OpenAIChatGenerator(
        model=model,
        api_key=os.getenv("OPENROUTER_API_KEY"),
        api_base_url=os.getenv("OPENROUTER_BASE_URL"),
        generation_kwargs={"temperature": 0.1, "max_tokens": 4096},
        streaming_callback=stream_handler,
    )
    return Agent(
        chat_generator=generator,
        tools=[python_executor],
        system_prompt=SYSTEM_PROMPT,
        max_agent_steps=10,
    )

Now the agent’s thought process prints token-by-token. Tool results still arrive as complete blocks — the sandbox doesn’t stream partial output.

Step 8: Harden the sandbox for production

The executor in Step 2 is a starting point. For production workloads, layer on these defenses:

1. Seccomp filter (Linux only) — restrict syscalls to a minimal allowlist:

# sandbox/executor.py (add to _preexec_fn)
try:
    import seccomp
    f = seccomp.SyscallFilter(seccomp.ALLOW)
    # Allow only essential syscalls
    for sc in ["read", "write", "exit", "exit_group", "brk", "mmap", "munmap",
               "openat", "close", "fstat", "lseek", "rt_sigaction", "rt_sigprocmask",
               "getpid", "getuid", "getgid", "geteuid", "getegid", "arch_prctl"]:
        f.add_rule(seccomp.ALLOW, sc)
    f.load()
except ImportError:
    pass  # seccomp not available — log warning in production

2. Network namespace isolation — run in a container with no network:

# Docker approach (preferred for multi-tenant)
docker run --rm --network none --memory=512m --cpus=1 \
  -v /tmp:/tmp python:3.11-slim python -I -S -B /tmp/agent_xyz.py

3. Filesystem virtualization — use bubblewrap or landlock (Linux 5.13+) to deny all FS access except a dedicated /tmp/agent_<uuid> mount.

4. Static analysis gate — before execution, run ast.parse() and reject code containing:

• import os, import subprocess, import socket, import urllib
• __import__, eval, exec, compile
• Attribute access on dangerous modules (os.system, subprocess.run)

# tools/python_tool.py (add before executor call)
import ast

FORBIDDEN_NODES = {
    ast.Import: ["os", "subprocess", "socket", "urllib", "requests", "http"],
    ast.ImportFrom: ["os", "subprocess", "socket", "urllib"],
    ast.Call: ["eval", "exec", "compile", "__import__"],
}

def validate_code(code: str) -> None:
    tree = ast.parse(code)
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.split(".")[0] in FORBIDDEN_NODES[ast.Import]:
                    raise ValueError(f"Forbidden import: {alias.name}")
        elif isinstance(node, ast.ImportFrom):
            if node.module and node.module.split(".")[0] in FORBIDDEN_NODES[ast.ImportFrom]:
                raise ValueError(f"Forbidden import from: {node.module}")
        elif isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name) and node.func.id in FORBIDDEN_NODES[ast.Call]:
                raise ValueError(f"Forbidden call: {node.func.id}")
            if isinstance(node.func, ast.Attribute):
                # Catch os.system, subprocess.run, etc.
                full = ast.unparse(node.func)
                if any(full.startswith(f) for f in ["os.", "subprocess.", "socket."]):
                    raise ValueError(f"Forbidden attribute call: {full}")

Call validate_code(code) at the top of python_executor before invoking the sandbox.

Step 9: Handle tool errors gracefully

Agents sometimes generate buggy code. The tool should return structured error info the model can reason about, not raise exceptions that crash the pipeline.

# tools/python_tool.py (enhanced)
@tool(name="python_executor", description="Execute Python code in a sandboxed subprocess.")
def python_executor(code: str) -> dict:
    try:
        validate_code(code)
    except ValueError as e:
        return {
            "stdout": "",
            "stderr": f"Static analysis rejected code: {e}",
            "exit_code": -2,
            "timed_out": False,
            "error_type": "validation_error",
        }

    result: ExecutionResult = _executor.run(code)
    
    # Classify common error patterns for the agent
    error_type = "runtime_error"
    if result.timed_out:
        error_type = "timeout"
    elif result.exit_code == -1:
        error_type = "resource_limit"
    elif "SyntaxError" in result.stderr:
        error_type = "syntax_error"
    elif "ImportError" in result.stderr or "ModuleNotFoundError" in result.stderr:
        error_type = "import_error"
    
    return {
        "stdout": result.stdout,
        "stderr": result.stderr,
        "exit_code": result.exit_code,
        "timed_out": result.timed_out,
        "error_type": error_type,
    }

Update the system prompt to reference error_type:

8. The tool returns an error_type field: validation_error, syntax_error, import_error, 
   timeout, resource_limit, or runtime_error. Use this to decide how to retry.
   - validation_error: rewrite code without forbidden constructs
   - syntax_error: fix syntax, re-run
   - import_error: use stdlib only, re-run
   - timeout/resource_limit: optimize algorithm or reduce data size
   - runtime_error: read stderr, debug, re-run

Step 10: Add structured output for downstream consumers

If this agent feeds another pipeline stage, emit JSON instead of free text. Add a response_format hint to the generator and a final formatting step.

# main.py (structured output)
from haystack.dataclasses import ChatMessage
from pydantic import BaseModel
import json

class AgentResponse(BaseModel):
    answer: str
    code_executed: bool
    tool_calls: int
    execution_time_ms: int

def extract_structured_response(messages: list[ChatMessage]) -> AgentResponse:
    # Last assistant message with tool_calls = the final answer
    for msg in reversed(messages):
        if msg.role.value == "assistant" and msg.text:
            try:
                return AgentResponse.model_validate_json(msg.text)
            except Exception:
                pass
    # Fallback
    return AgentResponse(
        answer=messages[-1].text or "",
        code_executed=any("tool_call" in str(m) for m in messages),
        tool_calls=sum(1 for m in messages if "tool_call" in str(m)),
        execution_time_ms=0,
    )

Then update the system prompt’s final rule:

9. Your FINAL response must be a single JSON object matching this schema:
{
  "answer": "string - the user-facing answer",
  "code_executed: "boolean - whether you called python_executor",
  "tool_calls": "integer - number of tool invocations",
  "execution_time_ms": "integer - approximate total execution time"
}
No markdown, no extra text. Just the JSON.

Step 11: Test edge cases and failure modes

Create a dedicated test file that exercises the guardrails.

# test_edge_cases.py
from tools.python_tool import python_executor

def test_forbidden_import():
    result = python_executor.invoke({
        "code": "import os; print(os.listdir('/'))"
    })
    assert result["exit_code"] == -2
    assert result["error_type"] == "validation_error"
    print("✓ Forbidden import blocked")

def test_timeout():
    result = python_executor.invoke({
        "code": "import time; time.sleep(60)"
    })
    assert result["timed_out"] is True
    assert result["error_type"] == "timeout"
    print("✓ Timeout enforced")

def test_memory_limit():
    result = python_executor.invoke({
        "code": "x = 'a' * (1024**3)"  # 1 GB string
    })
    assert result["exit_code"] != 0
    assert result["error_type"] in ("resource_limit", "runtime_error")
    print("✓ Memory limit enforced")

def test_network_blocked():
    result = python_executor.invoke({
        "code": "import urllib.request; urllib.request.urlopen('http://example.com')"
    })
    # Should fail at validation or runtime
    assert result["exit_code"] != 0
    print("✓ Network access blocked")

def test_valid_computation():
    result = python_executor.invoke({
        "code": "import math; print(math.factorial(20))"
    })
    assert result["exit_code"] == 0
    assert "2432902008176640000" in result["stdout"]
    print("✓ Valid computation works")

if __name__ == "__main__":
    test_forbidden_import()
    test_timeout()
    test_memory_limit()
    test_network_blocked()
    test_valid_computation()
    print("\nAll edge case tests passed")

Run with python test_edge_cases.py. Every test should pass.

Step 12: Deploy behind an API endpoint

Wrap the pipeline in FastAPI for production serving. Add request validation, rate limiting, and structured logging.

# api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from haystack import Pipeline
from haystack.dataclasses import ChatMessage
from main import build_pipeline
import time
import uuid

app = FastAPI(title="Code Agent API")
pipeline = build_pipeline()

class QueryRequest(BaseModel):
    query: str
    session_id: str | None = None

class QueryResponse(BaseModel):
    answer: str
    session_id: str
    tool_calls: int
    latency_ms: int

@app.post("/agent/query", response_model=QueryResponse)
async def query_agent(req: QueryRequest):
    session_id = req.session_id or str(uuid.uuid4())
    start = time.perf_counter()
    
    try:
        result = pipeline.run({
            "agent": {"messages": [ChatMessage.from_user(req.query)]}
        })
        latency_ms = int((time.perf_counter() - start) * 1000)
        
        # Parse structured response from logger output
        response_text = result["logger"]["response"]
        # ... extract fields from JSON response ...
        
        return QueryResponse(
            answer=response_text,
            session_id=session_id,
            tool_calls=0,  # compute from messages
            latency_ms=latency_ms,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "ok"}

Run with uvicorn api:app --host 0.0.0.0 --port 8000 --workers 4.

Verification: Hit the endpoint:

curl -X POST http://localhost:8000/agent/query \
  -H "Content-Type: application/json" \
  -d '{"query": "Compute SHA256 of hello world"}'

You should get a JSON response with the hash and latency under 5 seconds.

What you now have

A production-ready code-executing agent in Haystack 2.0 with:

• Secure sandbox: subprocess isolation, resource limits, static analysis gate
• Observable pipeline: logging, streaming, structured output
• Graceful error handling: typed error codes the agent can reason about
• Tested guardrails: validation, timeout, memory, network blocking
• Deployable API: FastAPI wrapper with health checks

The pattern scales. Swap the executor for a container runtime (gVisor, Firecracker) if you need stronger isolation. Add a vector store tool for RAG. Chain multiple agents via pipeline branches. The Haystack 2.0 primitives — Agent, Pipeline, @tool — compose cleanly.