Sandbox API

sandbox

orqest.sandbox — safe execution of dynamic tool implementations.

The Protocol is small and explicit: validate then execute.

Three backends ship today:

• :class:InProcessSandbox — Tier 0; in-process exec() with AST-level static restriction and a curated __builtins__ set. Requires explicit unsafe=True at construction because there is no real isolation. Suitable for tests and tightly-controlled dev workflows. NOT for LLM-generated code from untrusted sources.
• :class:SubprocessSandbox — Tier 1, the production default. Subprocess isolation with RLIMIT_AS + RLIMIT_CPU (POSIX) + outer asyncio.wait_for timeout. The wrapper subprocess runs against the same curated __builtins__ (from :mod:orqest.sandbox._safe_builtins) so reflection helpers and the unsafe builtins are never reachable. JSON args/result; stdin/stdout transport.
• :class:orqest.sandbox.docker.DockerSandbox — Tier 2; per-session container running the published orqest/agent-runtime image. Requires uv sync --group docker. Imported lazily from the orqest.sandbox.docker submodule so the host-side docker SDK stays an optional dep. Adds scope-separated JWT auth (agent for execution, operator for promote_tool / forget_tool) and per-user persisted tool library.

Every tier rejects the same set of static escapes (default-deny imports, reflection helpers, dunder reach-through, string-keyed subscript access — see :mod:orqest.sandbox._static) and runs with the same curated __builtins__ (:mod:orqest.sandbox._safe_builtins). Identifier paths that feed into per-agent workspaces are bounded by :mod:orqest.sandbox._identifiers.

Third parties can ship :class:E2BSandbox / :class:WasmSandbox / :class:FirecrackerSandbox against the same :class:Sandbox Protocol — no changes required in core.

The matching consumer is :class:orqest.autonomy.tool_factory.DynamicToolFactory, which turns a :class:GeneratedToolSpec (carrying an implementation string) into a runnable pydantic_ai.Tool via this Protocol.

SandboxRunError

SandboxRunError(
    message: str,
    *,
    stage: str,
    code_snippet: str,
    underlying: ExecutionResult
    | ValidationError
    | None = None,
)

Raised by :func:run_in_sandbox on validation / execution failure.

Attributes:

Name	Type	Description
stage		"validate" (AST rejected the code) or "execute" (sandbox launched the code but it errored / timed out / produced unparseable output).
code_snippet		First ~200 chars of the offending implementation, for diagnostics.
underlying		The raw :class:ExecutionResult (when stage="execute") or :class:ValidationError (when stage="validate") — for consumers who need the structured error fields.

Source code in orqest/sandbox/helpers.py

def __init__(
    self,
    message: str,
    *,
    stage: str,
    code_snippet: str,
    underlying: ExecutionResult | ValidationError | None = None,
) -> None:
    super().__init__(message)
    self.stage = stage
    self.code_snippet = code_snippet
    self.underlying = underlying

InProcessSandbox

InProcessSandbox(*, unsafe: bool = False)

Tier-0 sandbox — see module docstring. Refuses unsafe=False.

Source code in orqest/sandbox/inprocess.py

def __init__(self, *, unsafe: bool = False) -> None:
    if not unsafe:
        raise ValueError(
            "InProcessSandbox requires unsafe=True — there is no real "
            "isolation. Use SubprocessSandbox for production. See the "
            "module docstring for what InProcessSandbox does NOT protect."
        )
    self._unsafe = unsafe

validate async

validate(code: str, *, allowed_imports: set[str]) -> None

Static AST check; raise :class:ValidationError on issues.

Source code in orqest/sandbox/inprocess.py

async def validate(
    self,
    code: str,
    *,
    allowed_imports: set[str],
) -> None:
    """Static AST check; raise :class:`ValidationError` on issues."""
    issues = collect_issues(code, allowed_imports=allowed_imports)
    if issues:
        raise ValidationError(
            format_issues(issues),
            code_snippet=code[:200],
        )

execute async

execute(
    code: str,
    *,
    args: dict[str, Any],
    allowed_imports: set[str],
    timeout_s: float = 5.0,
    memory_mb: int = 128,
    agent_id: str | None = None,
    dependencies: list[str] | None = None,
) -> ExecutionResult

Validate, then run the code in a restricted namespace.

Note that timeout_s, memory_mb, agent_id, and dependencies are accepted for Protocol conformance but not enforced by this backend — the in-process path has no process boundary to enforce them on, and there are no per-agent venvs in-process.

Source code in orqest/sandbox/inprocess.py

async def execute(
    self,
    code: str,
    *,
    args: dict[str, Any],
    allowed_imports: set[str],
    timeout_s: float = 5.0,
    memory_mb: int = 128,
    agent_id: str | None = None,
    dependencies: list[str] | None = None,
) -> ExecutionResult:
    """Validate, then run the code in a restricted namespace.

    Note that ``timeout_s``, ``memory_mb``, ``agent_id``, and
    ``dependencies`` are accepted for Protocol conformance but
    **not enforced** by this backend — the in-process path has no
    process boundary to enforce them on, and there are no per-agent
    venvs in-process.
    """
    # Re-validate (defense in depth — also done at spawn time)
    await self.validate(code, allowed_imports=allowed_imports)

    # Build the curated namespace from the shared safe-builtins module
    # so every sandbox tier runs against the same allowlist.
    import importlib

    builtins = build_safe_builtins(allowed_imports)
    namespace: dict[str, Any] = {
        "__builtins__": builtins,
        "args": dict(args),
    }
    for module_name in allowed_imports:
        try:
            namespace[module_name.split(".", 1)[0]] = importlib.import_module(
                module_name
            )
        except ImportError as exc:
            return ExecutionResult(
                success=False,
                error=f"allowed import {module_name!r} not available: {exc}",
                duration_ms=0.0,
            )

    # User code is wrapped in a function so its `return` becomes the
    # captured output. The wrapper assigns the return value to a
    # well-known name (``__orqest_result``) we can read after exec.
    wrapped = (
        "def __orqest_tool(args):\n"
        + "\n".join("    " + line for line in code.splitlines())
        + "\n__orqest_result = __orqest_tool(args)\n"
    )

    stdout_buf = io.StringIO()
    t0 = monotonic()
    try:
        with redirect_stdout(stdout_buf):
            # ruff: noqa: S102  (sandbox is the whole point)
            exec(wrapped, namespace)  # noqa: S102
    except Exception as exc:  # noqa: BLE001
        return ExecutionResult(
            success=False,
            error=f"{type(exc).__name__}: {exc}",
            stdout=stdout_buf.getvalue(),
            duration_ms=(monotonic() - t0) * 1000.0,
        )

    result = namespace.get("__orqest_result")
    # JSON-roundtrip the result to enforce serializability (matches
    # the SubprocessSandbox boundary so behaviour is identical).
    try:
        json.dumps(result)
    except (TypeError, ValueError) as exc:
        return ExecutionResult(
            success=False,
            error=f"return value is not JSON-serializable: {exc}",
            stdout=stdout_buf.getvalue(),
            duration_ms=(monotonic() - t0) * 1000.0,
        )

    return ExecutionResult(
        success=True,
        output=result,
        stdout=stdout_buf.getvalue(),
        duration_ms=(monotonic() - t0) * 1000.0,
    )

ExecutionResult

Outcome of one :meth:Sandbox.execute call.

Always emitted regardless of success — :class:Sandbox implementations MUST NOT raise for user-code failures (those land in :attr:error). Reserved for infrastructure failures (e.g., subprocess crash before handshake completes) which still raise from :meth:Sandbox.execute.

success instance-attribute

success: bool

True when the implementation ran to completion and returned a value without raising. False for any user-code exception, timeout, memory cap, or non-JSON-serializable return.

output class-attribute instance-attribute

output: Any = None

JSON-serializable value the implementation returned. None on failure or when the implementation explicitly returned None.

error class-attribute instance-attribute

error: str | None = None

Captured exception message (or timeout / memory-cap signal) when :attr:success is False. None on success.

stdout class-attribute instance-attribute

stdout: str = ''

Captured stdout from the implementation. Useful for debugging LLM-generated code that prints intermediate values.

duration_ms class-attribute instance-attribute

duration_ms: float = Field(ge=0.0)

Wall-clock time spent inside the sandbox boundary.

Sandbox

Protocol for safe execution of dynamic tool implementations.

Two-stage contract — see module docstring. Implementations are async so they can do off-thread / off-process work (subprocess calls, HTTP to a sandbox provider, etc.) without blocking the agent loop.

validate async

validate(code: str, *, allowed_imports: set[str]) -> None

Static AST check. Raises :class:ValidationError on failure.

Source code in orqest/sandbox/protocol.py

async def validate(
    self,
    code: str,
    *,
    allowed_imports: set[str],
) -> None:
    """Static AST check. Raises :class:`ValidationError` on failure."""
    ...

execute async

execute(
    code: str,
    *,
    args: dict[str, Any],
    allowed_imports: set[str],
    timeout_s: float = 5.0,
    memory_mb: int = 128,
    agent_id: str | None = None,
    dependencies: list[str] | None = None,
) -> ExecutionResult

Run the (pre-validated) code with args and return the result.

Always returns :class:ExecutionResult; success=False captures user-code failures, timeouts, memory caps, or unserializable output. Infrastructure failures (subprocess crash before handshake) still raise.

Parameters:

Name	Type	Description	Default
code	str	The implementation source.	required
args	dict[str, Any]	Per-call arguments passed to the implementation as args.	required
allowed_imports	set[str]	Static-validation allowlist of top-level modules permitted in code.	required
timeout_s	float	Wall-clock cap inside the sandbox.	5.0
memory_mb	int	Memory cap (enforced by Tier-1+; Tier-0 ignores).	128
agent_id	str | None	Optional agent identifier — Tier-2 (Docker) routes execution into the agent's per-agent subfolder + .venv. Tier-0 / Tier-1 ignore.	None
dependencies	list[str] | None	Optional list of pip specifiers (e.g. ["pandas>=2.0"]). Tier-2 installs them into the agent's .venv before executing (gated by allowed_packages on the sandbox). Tier-0 / Tier-1 ignore.	None

Source code in orqest/sandbox/protocol.py

async def execute(
    self,
    code: str,
    *,
    args: dict[str, Any],
    allowed_imports: set[str],
    timeout_s: float = 5.0,
    memory_mb: int = 128,
    agent_id: str | None = None,
    dependencies: list[str] | None = None,
) -> ExecutionResult:
    """Run the (pre-validated) code with *args* and return the result.

    Always returns :class:`ExecutionResult`; ``success=False`` captures
    user-code failures, timeouts, memory caps, or unserializable output.
    Infrastructure failures (subprocess crash before handshake) still
    raise.

    Args:
        code: The implementation source.
        args: Per-call arguments passed to the implementation as ``args``.
        allowed_imports: Static-validation allowlist of top-level modules
            permitted in ``code``.
        timeout_s: Wall-clock cap inside the sandbox.
        memory_mb: Memory cap (enforced by Tier-1+; Tier-0 ignores).
        agent_id: Optional agent identifier — Tier-2 (Docker) routes
            execution into the agent's per-agent subfolder + ``.venv``.
            Tier-0 / Tier-1 ignore.
        dependencies: Optional list of pip specifiers (e.g.
            ``["pandas>=2.0"]``). Tier-2 installs them into the agent's
            ``.venv`` before executing (gated by ``allowed_packages``
            on the sandbox). Tier-0 / Tier-1 ignore.

    """
    ...

__aenter__ async

__aenter__() -> Sandbox

Optional context-manager hook for backends that hold resources.

Source code in orqest/sandbox/protocol.py

async def __aenter__(self) -> Sandbox:
    """Optional context-manager hook for backends that hold resources."""
    ...

__aexit__ async

__aexit__(*args: Any) -> None

Optional context-manager teardown.

Source code in orqest/sandbox/protocol.py

async def __aexit__(self, *args: Any) -> None:
    """Optional context-manager teardown."""
    ...

ValidationError

ValidationError(reason: str, *, code_snippet: str = '')

Raised by :meth:Sandbox.validate when static checks fail.

Examples: a disallowed import, a syntax error, an explicit eval call, or dunder attribute access (obj.__class__).

Source code in orqest/sandbox/protocol.py

def __init__(self, reason: str, *, code_snippet: str = "") -> None:
    super().__init__(reason)
    self.reason = reason
    self.code_snippet = code_snippet

SubprocessSandbox

SubprocessSandbox()

Tier-1 sandbox — subprocess with resource limits + outer timeout.

See module docstring for what this DOES protect against (memory, CPU, timeout, in-process namespace bleed) and what it does NOT (network, filesystem reads, child subprocesses if CPU cap permits).

Source code in orqest/sandbox/subprocess.py

def __init__(self) -> None:
    _warn_once_windows()

execute async

execute(
    code: str,
    *,
    args: dict[str, Any],
    allowed_imports: set[str],
    timeout_s: float = 5.0,
    memory_mb: int = 128,
    agent_id: str | None = None,
    dependencies: list[str] | None = None,
) -> ExecutionResult

Run code in a fresh subprocess. agent_id and dependencies are accepted for Protocol conformance but ignored — Tier-1 has no per-agent venv concept (use Tier-2 :class:DockerSandbox for that).

Source code in orqest/sandbox/subprocess.py

async def execute(
    self,
    code: str,
    *,
    args: dict[str, Any],
    allowed_imports: set[str],
    timeout_s: float = 5.0,
    memory_mb: int = 128,
    agent_id: str | None = None,
    dependencies: list[str] | None = None,
) -> ExecutionResult:
    """Run code in a fresh subprocess. ``agent_id`` and ``dependencies``
    are accepted for Protocol conformance but ignored — Tier-1 has no
    per-agent venv concept (use Tier-2 :class:`DockerSandbox` for that).
    """
    # Re-validate at the parent (defense in depth)
    await self.validate(code, allowed_imports=allowed_imports)

    spec_json = json.dumps(
        {
            "code": code,
            "args": dict(args),
            "allowed_imports": sorted(allowed_imports),
            "_helpers": {
                "safe_builtins_path": _SAFE_BUILTINS_PATH,
                "static_path": _STATIC_PATH,
            },
        }
    )

    t0 = monotonic()
    try:
        proc = await asyncio.create_subprocess_exec(
            sys.executable,
            "-c",
            _WRAPPER_SCRIPT,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            preexec_fn=_make_preexec(memory_mb, timeout_s),
        )
    except OSError as exc:
        # Infrastructure failure — couldn't even start subprocess.
        return ExecutionResult(
            success=False,
            error=f"failed to start subprocess: {exc}",
            duration_ms=(monotonic() - t0) * 1000.0,
        )

    try:
        stdout_bytes, stderr_bytes = await asyncio.wait_for(
            proc.communicate(input=spec_json.encode("utf-8")),
            timeout=timeout_s,
        )
    except TimeoutError:
        proc.kill()
        try:
            await proc.wait()
        except Exception:  # noqa: BLE001
            pass
        return ExecutionResult(
            success=False,
            error=f"sandbox execution timed out after {timeout_s:.2f}s",
            duration_ms=(monotonic() - t0) * 1000.0,
        )

    duration_ms = (monotonic() - t0) * 1000.0
    stderr_text = stderr_bytes.decode("utf-8", errors="replace")
    stdout_text = stdout_bytes.decode("utf-8", errors="replace")

    if not stdout_text.strip():
        return ExecutionResult(
            success=False,
            error=(
                f"subprocess produced no output (exit code {proc.returncode}); "
                f"stderr: {stderr_text[:500]}"
            ),
            stdout=stderr_text,
            duration_ms=duration_ms,
        )

    try:
        payload = json.loads(stdout_text)
    except json.JSONDecodeError as exc:
        return ExecutionResult(
            success=False,
            error=(
                f"subprocess output was not valid JSON ({exc}); "
                f"raw stdout: {stdout_text[:500]}; stderr: {stderr_text[:200]}"
            ),
            stdout=stdout_text,
            duration_ms=duration_ms,
        )

    return ExecutionResult(
        success=bool(payload.get("success")),
        output=payload.get("output"),
        error=payload.get("error"),
        stdout=str(payload.get("stdout", "")),
        duration_ms=duration_ms,
    )

run_in_sandbox async

run_in_sandbox(
    code: str,
    *,
    return_expression: str | None = None,
    args: dict[str, Any] | None = None,
    allowed_imports: set[str] | None = None,
    sandbox: Sandbox | None = None,
    timeout_s: float = 5.0,
    memory_mb: int = 128,
) -> Any

Run code in a sandbox; return the value (or raise :class:SandboxRunError).

Parameters:

Name	Type	Description	Default
code	str	Python source — typically a function definition. The body runs inside an isolated def __orqest_tool(args) wrapper supplied by the sandbox.	required
return_expression	str | None	When given, appended as \nreturn {expr}\n so the wrapper returns it. When None, code must already include a top-level return statement.	None
args	dict[str, Any] | None	Dict passed as args into the wrapper. Default {}.	None
allowed_imports	set[str] | None	Top-level module names the candidate may import. Default empty (any import statement in the candidate fails static validation).	None
sandbox	Sandbox | None	An optional :class:Sandbox instance. Default is a fresh :class:SubprocessSandbox (Tier 1 — subprocess + RLIMIT caps + outer timeout). Pass an existing sandbox to share lifecycle.	None
timeout_s	float	Wall-clock cap on execution. Default 5.0s.	5.0
memory_mb	int	Memory cap in MB. Default 128.	128

Returns:

Type	Description
Any	Whatever the wrapped function returns (JSON-serialisable values only,
Any	per the sandbox Protocol).

Raises:

Type	Description
SandboxRunError	Validation failure (stage="validate") or execution failure / timeout (stage="execute"). The exception carries code_snippet and underlying for diagnostics.

Source code in orqest/sandbox/helpers.py

async def run_in_sandbox(
    code: str,
    *,
    return_expression: str | None = None,
    args: dict[str, Any] | None = None,
    allowed_imports: set[str] | None = None,
    sandbox: Sandbox | None = None,
    timeout_s: float = 5.0,
    memory_mb: int = 128,
) -> Any:
    """Run *code* in a sandbox; return the value (or raise :class:`SandboxRunError`).

    Args:
        code: Python source — typically a function definition. The body runs
            inside an isolated ``def __orqest_tool(args)`` wrapper supplied
            by the sandbox.
        return_expression: When given, appended as ``\\nreturn {expr}\\n`` so
            the wrapper returns it. When ``None``, *code* must already
            include a top-level ``return`` statement.
        args: Dict passed as ``args`` into the wrapper. Default ``{}``.
        allowed_imports: Top-level module names the candidate may import.
            Default empty (any ``import`` statement in the candidate fails
            static validation).
        sandbox: An optional :class:`Sandbox` instance. Default is a fresh
            :class:`SubprocessSandbox` (Tier 1 — subprocess + RLIMIT caps +
            outer timeout). Pass an existing sandbox to share lifecycle.
        timeout_s: Wall-clock cap on execution. Default 5.0s.
        memory_mb: Memory cap in MB. Default 128.

    Returns:
        Whatever the wrapped function returns (JSON-serialisable values only,
        per the sandbox Protocol).

    Raises:
        SandboxRunError: Validation failure (``stage="validate"``) or
            execution failure / timeout (``stage="execute"``). The exception
            carries ``code_snippet`` and ``underlying`` for diagnostics.

    """
    sandbox = sandbox or SubprocessSandbox()
    implementation = _build_implementation(code, return_expression)
    snippet = implementation[:200]
    effective_imports = set(allowed_imports) if allowed_imports else set()
    effective_args = dict(args) if args else {}

    try:
        await sandbox.validate(implementation, allowed_imports=effective_imports)
    except ValidationError as exc:
        raise SandboxRunError(
            f"sandbox validation rejected the implementation: {exc}",
            stage="validate",
            code_snippet=snippet,
            underlying=exc,
        ) from exc

    result = await sandbox.execute(
        implementation,
        args=effective_args,
        allowed_imports=effective_imports,
        timeout_s=timeout_s,
        memory_mb=memory_mb,
    )

    if not result.success:
        raise SandboxRunError(
            f"sandbox execution failed: {result.error or 'unknown error'}",
            stage="execute",
            code_snippet=snippet,
            underlying=result,
        )

    return result.output

run_in_sandbox_safe async

run_in_sandbox_safe(
    code: str,
    *,
    return_expression: str | None = None,
    args: dict[str, Any] | None = None,
    allowed_imports: set[str] | None = None,
    sandbox: Sandbox | None = None,
    timeout_s: float = 5.0,
    memory_mb: int = 128,
) -> tuple[bool, Any, str | None]

Non-raising variant of :func:run_in_sandbox.

Returns (success, output, error_message). On success: (True, result, None). On any failure (validation or execution): (False, None, <error_msg>).

Suited for code that wants to handle failure inline without try/except — e.g., test-driven loops that iterate over many candidates and need a per-candidate verdict without exception bookkeeping.

Source code in orqest/sandbox/helpers.py

async def run_in_sandbox_safe(
    code: str,
    *,
    return_expression: str | None = None,
    args: dict[str, Any] | None = None,
    allowed_imports: set[str] | None = None,
    sandbox: Sandbox | None = None,
    timeout_s: float = 5.0,
    memory_mb: int = 128,
) -> tuple[bool, Any, str | None]:
    """Non-raising variant of :func:`run_in_sandbox`.

    Returns ``(success, output, error_message)``. On success: ``(True, result, None)``.
    On any failure (validation or execution): ``(False, None, <error_msg>)``.

    Suited for code that wants to handle failure inline without try/except —
    e.g., test-driven loops that iterate over many candidates and need a
    per-candidate verdict without exception bookkeeping.
    """
    try:
        output = await run_in_sandbox(
            code,
            return_expression=return_expression,
            args=args,
            allowed_imports=allowed_imports,
            sandbox=sandbox,
            timeout_s=timeout_s,
            memory_mb=memory_mb,
        )
    except SandboxRunError as exc:
        return False, None, str(exc)
    return True, output, None