Integration: exec-sandbox component for hardware-isolated code execution

Summary

Haystack has 96 integrations spanning model providers, document stores, monitoring, and data ingestion -- but no sandboxed code execution component. As agents and pipelines increasingly generate and run code (data analysis, tool use, code interpreters), there is a gap: no built-in way to execute arbitrary Python/JavaScript/raw shell scripts safely inside a pipeline.

exec-sandbox provides hardware-isolated code execution via QEMU microVMs (KVM on Linux, HVF on macOS). Each execution runs in a dedicated VM that is destroyed after use. It ships as a single pip install, requires no cloud account, and is Apache-2.0 licensed.

This issue proposes an ExecSandboxRunner component and an ExecSandboxTool that follow Haystack's @component contract and Tool interface respectively.

Motivation

• Agent tool calling -- Haystack's Agent component supports tool-calling workflows via Tool, ComponentTool, and @tool. An LLM that generates Python or JS code today has no safe place to run it inside a Haystack pipeline. The common workaround is eval()/exec() in the host process, which is a security liability.
• Data analysis pipelines -- RAG pipelines that retrieve data and then compute over it (aggregations, charts, statistical tests) need a code execution step between retrieval and generation.
• Code interpreter pattern -- The increasingly common "write code, execute, observe output, iterate" loop requires a sandboxed executor that an agent can call repeatedly with state persistence.

Proposed API

1. Pipeline component -- ExecSandboxRunner

A standard Haystack component for one-shot code execution inside a pipeline.

from haystack import component
from exec_sandbox import Scheduler, SchedulerConfig, ExecutionResult

@component
class ExecSandboxRunner:
    """Execute code in a hardware-isolated QEMU microVM."""

    def __init__(
        self,
        language: str = "python",
        timeout_seconds: int = 30,
        memory_mb: int = 192,
        packages: list[str] | None = None,
        allow_network: bool = False,
        allowed_domains: list[str] | None = None,
        warm_pool_size: int = 2,
    ):
        self.language = language
        self.timeout_seconds = timeout_seconds
        self.memory_mb = memory_mb
        self.packages = packages
        self.allow_network = allow_network
        self.allowed_domains = allowed_domains
        self._scheduler_config = SchedulerConfig(warm_pool_size=warm_pool_size)
        self._scheduler: Scheduler | None = None
        self._loop = None

    def to_dict(self) -> dict:
        """Serialize component config for pipeline serialization."""
        return {
            "language": self.language,
            "timeout_seconds": self.timeout_seconds,
            "memory_mb": self.memory_mb,
            "packages": self.packages,
            "allow_network": self.allow_network,
            "allowed_domains": self.allowed_domains,
            "warm_pool_size": self._scheduler_config.warm_pool_size,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "ExecSandboxRunner":
        """Deserialize component config from pipeline serialization."""
        return cls(**data)

    def warm_up(self):
        """Called by the pipeline before first run. Starts the VM pool."""
        import asyncio
        self._scheduler = Scheduler(self._scheduler_config)
        self._loop = asyncio.new_event_loop()
        self._loop.run_until_complete(self._scheduler.__aenter__())

    def tear_down(self):
        """Called by the pipeline on shutdown. Destroys all VMs and cleans up."""
        if self._scheduler:
            try:
                self._loop.run_until_complete(self._scheduler.__aexit__(None, None, None))
            finally:
                if self._loop:
                    self._loop.close()

    @component.output_types(
        stdout=str,
        stderr=str,
        exit_code=int,
    )
    def run(self, code: str) -> dict:
        result = self._loop.run_until_complete(
            self._scheduler.run(
                code=code,
                language=self.language,
                timeout_seconds=self.timeout_seconds,
                memory_mb=self.memory_mb,
                packages=self.packages,
                allow_network=self.allow_network,
                allowed_domains=self.allowed_domains,
            )
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.exit_code,
        }

    # Async variant for AsyncPipeline
    @component.output_types(
        stdout=str,
        stderr=str,
        exit_code=int,
    )
    async def run_async(self, code: str) -> dict:
        result = await self._scheduler.run(
            code=code,
            language=self.language,
            timeout_seconds=self.timeout_seconds,
            memory_mb=self.memory_mb,
            packages=self.packages,
            allow_network=self.allow_network,
            allowed_domains=self.allowed_domains,
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.exit_code,
        }

Pipeline usage:

from haystack import Pipeline, component
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage

@component
class CodeExtractor:
    """Extract code text from the first ChatMessage reply."""

    @component.output_types(code=str)
    def run(self, replies: list[ChatMessage]) -> dict:
        return {"code": replies[0].text}

pipe = Pipeline()
pipe.add_component("prompt", ChatPromptBuilder(...))
pipe.add_component("llm", OpenAIChatGenerator())
pipe.add_component("extract", CodeExtractor())
pipe.add_component("sandbox", ExecSandboxRunner(language="python", packages=["pandas==2.2.0"]))
pipe.connect("llm.replies", "extract.replies")   # list[ChatMessage] -> list[ChatMessage]
pipe.connect("extract.code", "sandbox.code")      # str -> str

2. Agent tool -- ExecSandboxTool

A Tool for Haystack's Agent that enables tool-calling LLMs to execute code.

from haystack.tools import Tool
from exec_sandbox import Scheduler

def _make_exec_sandbox_tool(
    scheduler: Scheduler,
    language: str = "python",
    timeout_seconds: int = 30,
) -> Tool:
    """Create a Haystack Tool backed by exec-sandbox."""

    import asyncio

    loop = asyncio.new_event_loop()

    def _execute(code: str) -> dict:
        """Sync wrapper — Haystack Tool rejects async callables."""
        result = loop.run_until_complete(
            scheduler.run(
                code=code,
                language=language,
                timeout_seconds=timeout_seconds,
            )
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.exit_code,
        }

    return Tool(
        name="code_interpreter",
        description=(
            f"Execute {language} code in a secure sandbox. "
            "Returns stdout, stderr, and exit_code. "
            "Use this to run computations, analyze data, or test code."
        ),
        parameters={
            "type": "object",
            "properties": {
                "code": {
                    "type": "string",
                    "description": f"The {language} code to execute",
                },
            },
            "required": ["code"],
        },
        function=_execute,
        outputs_to_string={"source": "stdout"},
    )

Agent usage:

from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

async with Scheduler() as scheduler:
    tool = _make_exec_sandbox_tool(scheduler, language="python")
    agent = Agent(
        chat_generator=OpenAIChatGenerator(model="gpt-4o"),
        tools=[tool],
    )
    result = agent.run(
        messages=[ChatMessage.from_user("Calculate the first 20 Fibonacci numbers")]
    )

3. ComponentTool wrapping (zero-code path)

Since ExecSandboxRunner is a standard @component, it can also be wrapped as a tool with no additional code:

runner = ExecSandboxRunner(language="python")
tool = ComponentTool(component=runner, name="code_interpreter")
agent = Agent(chat_generator=..., tools=[tool])

Design notes

Concern	Approach
Lifecycle	warm_up() starts the Scheduler and VM pool; pipeline teardown destroys VMs.
Async	run_async() mirrors run() for AsyncPipeline support. Haystack auto-detects __haystack_supports_async__ from the presence of run_async.
Serialization	All init params are primitives -- to_dict()/from_dict() work out of the box with Haystack's default serializer.
State	One-shot by default (each run() gets a fresh VM). Stateful sessions (multi-turn code interpreter) can be added via a ExecSandboxSession component that wraps scheduler.session().
Isolation	Hardware VM boundary (QEMU with KVM/HVF). Not a container, not a namespace. Each execution is a separate machine.
Performance	1-2ms warm pool start. Warm pool pre-boots VMs during warm_up() so the first run() is fast.
License	exec-sandbox is Apache-2.0, same as Haystack.

Prior art and alternatives

Alternative	Limitation
eval()/exec() in host	No isolation. Full access to host filesystem, network, and process.
Docker-based execution	Shared kernel. Container escapes are a known attack surface.
E2B	Cloud-only. Data leaves your infrastructure.
Modal	Cloud-only. Proprietary.

References

• exec-sandbox: github.com/dualeai/exec-sandbox | PyPI
• Haystack custom components: docs.haystack.deepset.ai/docs/custom-components
• Haystack Agent and Tools: docs.haystack.deepset.ai/docs/agent
• Haystack AsyncPipeline: docs.haystack.deepset.ai/docs/asyncpipeline
• ComponentTool: docs.haystack.deepset.ai/docs/componenttool
• Haystack integrations: haystack.deepset.ai/integrations

Start by reviewing the repository’s existing integration packages alongside the referenced Haystack custom-component, Agent/Tool, AsyncPipeline, and ComponentTool entry points. Clarify the scope and integration boundaries for ExecSandboxRunner and ExecSandboxTool, then verify that the agreed design covers lifecycle, serialization, synchronous and asynchronous execution, isolation, and documentation.