langchain-kubernetes-agent-sandbox 0.2.1

A LangChain sandbox tool that runs agent-generated code inside Kubernetes pods

langchain-kubernetes-agent-sandbox

Run agent-generated code inside Kubernetes pods via k8s-agent-sandbox.

KubernetesAgentSandbox adapts a k8s_agent_sandbox.Sandbox to the deepagents.backends.sandbox.BaseSandbox protocol, so it can be used in two ways:

1. As a deepagents backend — the pod is the agent's filesystem and shell; state persists across turns.
2. As a LangChain tool — the agent keeps an in-memory virtual filesystem and calls an execute_python tool that spawns an ephemeral pod per run.

	Sandbox as backend	Sandbox as tool
Lifetime	Lives for the whole agent session	Spawned per run, terminated at end
State across turns	Yes (files + shell)	No
Best for	Long, stateful coding sessions	Stateless code execution, lower idle cost
Wiring	One kwarg on create_deep_agent	Custom @tool + VFS seeding

Installation

pip install langchain-kubernetes-agent-sandbox

Requires Python 3.12+. You also need access to a Kubernetes cluster that the k8s-agent-sandbox client can reach (your local kubeconfig context is used by default).

Pattern 1: Sandbox as a deepagents backend

The sandbox is created once and passed to create_deep_agent as backend=. The agent's shell and file tools are routed to the pod for the lifetime of the agent.

from deepagents import create_deep_agent
from k8s_agent_sandbox import SandboxClient
from k8s_agent_sandbox.models import SandboxLocalTunnelConnectionConfig

from langchain_kubernetes_agent_sandbox import KubernetesAgentSandbox

SANDBOX_TEMPLATE = "python-3.12"   # any template available in your cluster
SANDBOX_NAMESPACE = "agent-sandboxes"


def build_sandbox() -> KubernetesAgentSandbox:
    client = SandboxClient(connection_config=SandboxLocalTunnelConnectionConfig())
    raw_sandbox = client.create_sandbox(
        template=SANDBOX_TEMPLATE,
        namespace=SANDBOX_NAMESPACE,
    )
    return KubernetesAgentSandbox(sandbox=raw_sandbox)


sandbox = build_sandbox()

agent = create_deep_agent(
    model="claude-sonnet-4-6",
    backend=sandbox,
)

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "Write a Python script that prints the first 10 primes and run it."},
    ],
})

# When you're done with the session:
sandbox.sandbox.terminate()

Files uploaded with sandbox.upload_files([...]) (absolute paths only) are visible to the agent for the rest of the session.

Pattern 2: Sandbox as a tool (execute_python)

The agent runs with an in-memory virtual filesystem (state["files"]). When it calls execute_python, the tool lazily spawns a sandbox, copies the VFS into /app/, runs the code, and the caller terminates the sandbox at the end of the run.

import base64
from typing import Annotated

from deepagents import create_deep_agent
from k8s_agent_sandbox import SandboxClient
from k8s_agent_sandbox.models import SandboxLocalTunnelConnectionConfig
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedState

from langchain_kubernetes_agent_sandbox import KubernetesAgentSandbox

SANDBOX_TEMPLATE = "python-3.12"
SANDBOX_NAMESPACE = "agent-sandboxes"
UPLOAD_DIR = "/app"

TOOL_MODE_PROMPT_SUFFIX = (
    "\n\nYou have an `execute_python(code)` tool that runs Python 3 in a fresh "
    "Kubernetes sandbox. The sandbox is spawned on the first call within this "
    "turn and terminated when the turn ends — do not rely on state across "
    "tool calls. Any files in your virtual filesystem are copied into `/app/` "
    "inside the sandbox before each call, so reference uploaded files by their "
    "`/app/<name>` path."
)


def _new_sandbox() -> KubernetesAgentSandbox:
    client = SandboxClient(connection_config=SandboxLocalTunnelConnectionConfig())
    raw = client.create_sandbox(template=SANDBOX_TEMPLATE, namespace=SANDBOX_NAMESPACE)
    return KubernetesAgentSandbox(sandbox=raw)


# Per-run sandbox handle. Replace this module-level dict with whatever
# request-scoped storage your app uses (e.g. Streamlit session_state, a FastAPI
# request, an asyncio context var).
_run: dict = {}


def _get_or_spawn_run_sandbox(vfs_files: dict) -> KubernetesAgentSandbox:
    sb = _run.get("sandbox")
    if sb is not None and sb.sandbox.is_active:
        return sb

    sb = _new_sandbox()
    _run["sandbox"] = sb

    payload: list[tuple[str, bytes]] = []
    for path, data in (vfs_files or {}).items():
        content = data.get("content", "")
        encoding = data.get("encoding", "utf-8")
        raw = base64.b64decode(content) if encoding == "base64" else content.encode("utf-8")
        sb_path = path if path.startswith("/") else f"{UPLOAD_DIR}/{path}"
        payload.append((sb_path, raw))
    if payload:
        sb.upload_files(payload)
    return sb


def _terminate_run_sandbox() -> None:
    sb = _run.pop("sandbox", None)
    if sb is not None:
        try:
            sb.sandbox.terminate()
        except Exception:
            pass


@tool
def execute_python(code: str, state: Annotated[dict, InjectedState]) -> str:
    """Execute a Python 3 snippet in an ephemeral Kubernetes sandbox.

    Files in the agent's virtual filesystem are copied into /app/ before
    execution. Returns combined stdout/stderr with the exit code.
    """
    sb = _get_or_spawn_run_sandbox(state.get("files") or {})
    b64 = base64.b64encode(code.encode("utf-8")).decode("ascii")
    cmd = (
        "python3 -c "
        "'import base64,sys;"
        f"exec(compile(base64.b64decode(\"{b64}\").decode(),\"<agent>\",\"exec\"))'"
    )
    resp = sb.execute(cmd)
    return f"exit={resp.exit_code}\n{resp.output}"


agent = create_deep_agent(
    model="claude-sonnet-4-6",
    tools=[execute_python],
    system_prompt="You are a helpful assistant." + TOOL_MODE_PROMPT_SUFFIX,
)

try:
    result = agent.invoke({
        "messages": [{"role": "user", "content": "Plot the first 10 primes."}],
        # Seed the VFS with files the agent should see at /app/<name>:
        # "files": {"/app/data.csv": {"content": "...", "encoding": "utf-8"}},
    })
finally:
    _terminate_run_sandbox()

Key points:

• The sandbox is created lazily on the first execute_python call within a run and reused for subsequent calls in the same run.
• VFS files are seeded into /app/ before each call; the agent should reference them by absolute path (/app/<name>).
• The caller is responsible for terminating the per-run sandbox in a finally block — the tool itself does not own the run's lifecycle.

Configuring the command timeout

Each execute call is bounded by a timeout (default 300 seconds). Override the default per sandbox, or per call:

sandbox = KubernetesAgentSandbox(sandbox=raw_sandbox, timeout=120)
sandbox.execute("pytest -q", timeout=600)

Commands that exceed the timeout return an ExecuteResponse with exit code 124 instead of raising.

Capabilities

KubernetesAgentSandbox implements the BaseSandbox interface:

• execute(command, *, timeout=None) — run a shell command in the pod.
• upload_files([(path, bytes), ...]) — write files into the pod (absolute paths only).
• download_files([path, ...]) — read files from the pod (absolute paths only).
• id — the underlying sandbox ID.

License

MIT — see LICENSE.