langchain-k8s

Kubernetes-native sandbox execution for LangChain Deep Agents

Give your AI agents their own isolated Kubernetes pods to run shell commands and file operations — fully self-hosted, zero vendor lock-in.

Implements the BaseSandbox / SandboxBackendProtocol contract using kubernetes-sigs/agent-sandbox as the execution backend.

Why langchain-k8s?

• Isolated execution — each agent gets its own pod with its own filesystem
• Self-hosted — runs on any Kubernetes cluster you control
• Disposable or durable — tear a pod down at the end of a task, or keep it alive and reconnect to it across turns
• Drop-in compatible — implements the LangChain Deep Agents sandbox protocol
• Enterprise-ready — path policies, virtual filesystems, sticky sessions, reconnection

Installation

pip install langchain-k8s

Or with uv:

uv add langchain-k8s

Prerequisites

• A Kubernetes cluster with the agent-sandbox controller v0.5.2 or newer installed
• A SandboxTemplate resource defining the pod spec for your sandboxes
• A SandboxWarmPool resource referencing that template — this is what sandboxes are claimed from, and it is mandatory (use replicas: 0 for pure on-demand cold start)
• kubectl configured with cluster access

Upgrading to 0.6.0 (breaking)

agent-sandbox v0.5.0 graduated its CRDs to v1beta1, and a v1beta1 SandboxClaim has no template reference at all — spec.warmPoolRef is required instead. A claim now points at a SandboxWarmPool, and the pool is what points at the SandboxTemplate. This package follows that model:

There is no deprecated alias: passing template_name= now raises TypeError. To migrate, create a SandboxWarmPool pointing at your existing template and pass its name. SandboxInClusterConnectionConfig(use_pod_ip=...) is also gone; the pod IP is now preferred automatically with a cluster-DNS fallback.

Quick start

Ecosystem-standard mode (recommended)

Pass a pre-created k8s_agent_sandbox.Sandbox handle — the standard LangChain Deep Agents sandbox backend pattern. Lifecycle management stays with the caller:

from k8s_agent_sandbox import SandboxClient
from k8s_agent_sandbox.models import SandboxLocalTunnelConnectionConfig
from langchain_anthropic import ChatAnthropic
from deepagents import create_deep_agent
from langchain_k8s import KubernetesSandbox

client = SandboxClient(
    connection_config=SandboxLocalTunnelConnectionConfig(),
)
handle = client.create_sandbox(
    warmpool="python-sandbox-pool",
    namespace="agent-sandbox-system",
)

backend = KubernetesSandbox(sandbox=handle)

agent = create_deep_agent(
    model=ChatAnthropic(model="claude-sonnet-4-20250514"),
    system_prompt="You are a Python coding assistant with sandbox access.",
    backend=backend,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Create a Python script that prints the Fibonacci sequence"}]}
)

client.delete_sandbox(handle.claim_name, "agent-sandbox-system")

Config-based mode (convenience)

For simpler setups, pass warmpool_name and connection parameters directly. The sandbox is created lazily on first use and destroyed on stop():

from langchain_k8s import KubernetesSandbox

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
)

agent = create_deep_agent(model=model, backend=backend, ...)
result = agent.invoke({"messages": [...]})

backend.stop()

Usage

Context manager

with KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
) as backend:
    agent = create_deep_agent(model=model, backend=backend, ...)
    result = agent.invoke({"messages": [...]})
# Sandbox pod is automatically cleaned up

Direct execution

from k8s_agent_sandbox import SandboxClient
from k8s_agent_sandbox.models import SandboxLocalTunnelConnectionConfig
from langchain_k8s import KubernetesSandbox

client = SandboxClient(
    connection_config=SandboxLocalTunnelConnectionConfig(),
)
handle = client.create_sandbox(
    warmpool="python-sandbox-pool",
    namespace="agent-sandbox-system",
)

backend = KubernetesSandbox(sandbox=handle)

# Execute commands (with optional per-call timeout)
resp = backend.execute("echo 'Hello from K8s!'")
print(resp.output, resp.exit_code)

resp = backend.execute("python3 long_script.py", timeout=600)

# Upload files
backend.upload_files([("/workspace/script.py", b"print('hello')\n")])

# Download files
results = backend.download_files(["/workspace/script.py"])
print(results[0].content)

client.delete_sandbox(handle.claim_name, "agent-sandbox-system")

Connection modes

# Production — cluster Gateway
backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    gateway_name="sandbox-gateway",
)

# Development — automatic port-forward
backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
)

# Advanced — existing port-forward or in-cluster routing
backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    api_url="http://localhost:8080",
)

# InCluster — agent pod connecting directly to sandbox pods
from k8s_agent_sandbox.models import SandboxInClusterConnectionConfig

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    # Prefers the pod IP from the Sandbox status, falls back to cluster DNS.
    connection_config=SandboxInClusterConnectionConfig(),
)

Sandbox lifecycle

Thread-scoped (production)

Use create_kubernetes_sandbox() for thread-scoped sandboxes in production. It implements a get-or-create pattern: if a sandbox with the given claim_name already exists, it is reused; otherwise a new one is created.

Define a graph factory that provisions a sandbox per conversation thread:

from langchain_anthropic import ChatAnthropic
from deepagents import create_deep_agent
from k8s_agent_sandbox import SandboxClient
from k8s_agent_sandbox.models import SandboxGatewayConnectionConfig
from langchain_k8s import create_kubernetes_sandbox
from langchain_core.runnables import RunnableConfig

client = SandboxClient(
    connection_config=SandboxGatewayConnectionConfig(gateway_name="sandbox-gw"),
)

async def make_agent(config: RunnableConfig):
    """Graph factory — each thread_id gets its own sandbox."""
    thread_id = config["configurable"]["thread_id"]
    backend = create_kubernetes_sandbox(
        client=client,
        claim_name=f"sandbox-{thread_id}",
        warmpool_name="python-sandbox-pool",
        namespace="agent-sandbox-system",
        labels={"thread_id": thread_id},
    )
    return create_deep_agent(
        model=ChatAnthropic(model="claude-sonnet-4-20250514"),
        backend=backend,
    )

Each conversation thread gets its own sandbox. When the thread resumes, the existing sandbox is found by claim_name and reused with its filesystem state intact:

# Turn 1 — user starts a new conversation, sandbox is created
config = {"configurable": {"thread_id": "user-session-abc"}}
agent = await make_agent(config)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Write a Python script that fetches weather data"}]}
)
# Agent writes files to the sandbox filesystem...

# Turn 2 — user continues the conversation, same sandbox is reused
agent = await make_agent(config)  # get-or-create finds the existing pod
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Now add error handling to the script"}]}
)
# Agent can read/modify files from turn 1 — filesystem state persists

# Clean up when the conversation ends
client.delete_sandbox(f"sandbox-user-session-abc", "agent-sandbox-system")

Config-based lifecycle

In config-based mode the sandbox pod is created lazily on first use and reused for every subsequent call. Filesystem state persists for the life of the backend. stop() deletes the SandboxClaim — unless you pass skip_cleanup=True, which leaves the pod running so you can reconnect to it later by claim_name.

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
)

To get a fresh pod, create a fresh backend — or call stop() then start(), which deletes the claim and provisions a new one.

reuse_sandbox — automatic reconnection

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    reuse_sandbox=False,
)

reuse_sandbox controls error recovery, not pod lifetime. With the default True, a failure inside execute() causes the backend to drop the dead sandbox, provision a replacement, and retry the command once before giving up. Setting it to False disables that retry, so connection errors propagate to the caller immediately.

It has no effect on start()/stop(), and it does not create a pod per invocation. Auto-reconnect is also inactive in handle mode, where the caller owns the lifecycle.

Enterprise features

Path access policy

Restrict which directories agents can write to using allow_prefixes. When set, only write() and edit() operations targeting paths under the specified prefixes are permitted. All other paths return an error without executing a command.

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    allow_prefixes=["/tmp/"],
)

When allow_prefixes is None (the default), no write restrictions are applied.

Note: allow_prefixes is a tool-level policy — it controls which paths write() and edit() accept, but does not block shell commands like execute("echo bad > /etc/passwd"). Use the Kubernetes pod securityContext (e.g. readOnlyRootFilesystem) for system-level protection. The allowed directories must also be writable inside the container — see Container permissions vs. sandbox policy.

Virtual filesystem

When virtual_mode=True, all file-operation paths (read, write, edit, ls, grep, glob, uploads, downloads) are resolved under root_dir (default /tmp). Path traversal (.., ~) is rejected.

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    virtual_mode=True,
    root_dir="/tmp",
)

# Agent sees virtual paths — resolved under /tmp automatically:
#   write("/src/main.py", ...)  →  writes to /tmp/src/main.py
#   read("/src/main.py")        →  reads from /tmp/src/main.py
#   upload_files([("/data/input.csv", content)])  →  /tmp/data/input.csv

When combined with allow_prefixes, the policy check runs against the resolved path:

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    virtual_mode=True,
    root_dir="/tmp",
    allow_prefixes=["/tmp/"],
)
# Virtual path "/src/main.py" resolves to "/tmp/src/main.py" — allowed
# Virtual path "../../etc/passwd" — rejected (path traversal)

Container permissions vs. sandbox policy

allow_prefixes and virtual_mode are tool-level policies enforced by KubernetesSandbox before any command reaches the container. They do not grant filesystem permissions inside the container itself.

For file operations to succeed, two conditions must be met:

1. Sandbox policy allows the path — allow_prefixes check passes (or is not set)
2. Container OS user can write to the path — the directory exists and is writable inside the container

If a write() or edit() call passes the sandbox policy but fails with PermissionError, the container's filesystem permissions are the cause. The sandbox logs a warning when this happens:

WARNING  langchain_k8s.sandbox: write: container permission denied for path='/src/main.py'
(resolved='/workspace/src/main.py'). The sandbox policy (allow_prefixes) permits this path,
but the container's OS user cannot write to it.

Common writable locations (no extra configuration needed):

Making a custom directory writable in your SandboxTemplate:

apiVersion: extensions.agents.x-k8s.io/v1beta1
kind: SandboxTemplate
metadata:
  name: python-sandbox-template
  namespace: agent-sandbox-system
spec:
  podTemplate:
    spec:
      containers:
        - name: python-runtime
          image: registry.k8s.io/agent-sandbox/python-runtime-sandbox:v0.5.4
          volumeMounts:
            - name: workspace
              mountPath: /workspace
      volumes:
        - name: workspace
          emptyDir: {}

With this configuration, /workspace is backed by an emptyDir volume and is writable regardless of the container's root filesystem permissions. You can then safely use root_dir="/workspace" and allow_prefixes=["/workspace/"].

Durable workspaces with volume_claim_templates

emptyDir disappears with the pod. For a workspace that survives a pod restart, pass volume_claim_templates instead — the claim then requests a PersistentVolumeClaim per sandbox:

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    root_dir="/workspace",
    virtual_mode=True,
    volume_claim_templates=[
        {
            "metadata": {"name": "workspace"},
            "spec": {
                "accessModes": ["ReadWriteOnce"],
                "resources": {"requests": {"storage": "1Gi"}},
            },
        }
    ],
)

Two requirements that are easy to miss:

• The SandboxTemplate pod spec must still declare a matching volumeMounts entry (name workspace, mountPath: /workspace). The claim template supplies the volume; only the template can mount it.
• The cluster needs a working StorageClass. Kind ships local-path with WaitForFirstConsumer, so the PVC stays Pending until the sandbox pod is scheduled — expect the first start to be slower than with emptyDir.

Pod labels and annotations

labels land on the SandboxClaim object. To label the pod instead — so the values are visible to node-level tooling and readable from inside the container through the Downward API — use pod_labels / pod_annotations, which map to spec.additionalPodMetadata:

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    labels={"session": "s1"},            # -> SandboxClaim object
    pod_labels={"team": "platform"},     # -> the running Pod
    pod_annotations={"owner": "agents"},
)

These are validated client-side for RFC 1123 syntax only. The controller additionally rejects reserved and restricted-domain keys, which fails the claim quickly with an InvalidMetadata reason rather than timing out.

Note: the SDK always stamps agents.x-k8s.io/created-by: python-client on claims it creates, so a claim's label set is never exactly what you passed in. Account for it in any label selector you write.

Horizontal scaling and sticky sessions

When deploying a service that uses KubernetesSandbox behind a load balancer with multiple replicas, requests from the same user or session must be routed to the same service instance. The sandbox state (pod, port-forward) is held in-process, so different instances cannot share a sandbox.

Configure sticky sessions using one of these approaches:

Kubernetes Service with session affinity:

apiVersion: v1
kind: Service
metadata:
  name: my-agent-service
spec:
  sessionAffinity: ClientIP
  sessionAffinityConfig:
    clientIP:
      timeoutSeconds: 3600
  selector:
    app: my-agent
  ports:
    - port: 80
      targetPort: 8080

NGINX Ingress with cookie affinity:

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: my-agent-ingress
  annotations:
    nginx.ingress.kubernetes.io/affinity: "cookie"
    nginx.ingress.kubernetes.io/session-cookie-name: "AGENT_SESSION"
    nginx.ingress.kubernetes.io/session-cookie-max-age: "3600"
spec:
  rules:
    - host: agent.example.com
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: my-agent-service
                port:
                  number: 80

Preserving sandboxes and reconnection

Set skip_cleanup=True to prevent sandbox pod destruction when stop() is called. The Kubernetes SandboxClaim is preserved so the sandbox pod continues running.

When an agent process restarts (pod eviction, rolling update, crash), it can reconnect to the still-running sandbox by passing the original claim_name. Persist the claim_name property after start() and use it on the next instantiation:

# First run — create sandbox and persist claim name
backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    skip_cleanup=True,
)
backend.start()
redis.set("sandbox:user-123", backend.claim_name)  # persist for reconnection

# After restart — reconnect to the same pod
backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    claim_name=redis.get("sandbox:user-123"),  # re-attach, no new pod
    skip_cleanup=True,
)
backend.start()  # re-establishes connection to existing pod
resp = backend.execute("cat /workspace/previous-work.py")  # state is preserved

The sandbox pod must be cleaned up externally when no longer needed (e.g. Kubernetes TTL controller, CronJob, or manual deletion).

Sandbox labels

Tag sandboxes at creation with Kubernetes labels for discovery, filtering, and cleanup policies:

backend = KubernetesSandbox(
    warmpool_name="python-sandbox-pool",
    namespace="agent-sandbox-system",
    labels={
        "session": "abc-123",
        "agent-id": "code-reviewer",
        "team": "platform",
    },
    skip_cleanup=True,
)

Labels are applied to the SandboxClaim resource and can be used with kubectl:

# List sandboxes for a specific session
kubectl get sandboxclaims -l session=abc-123

# Clean up all sandboxes for a team
kubectl delete sandboxclaims -l team=platform

Labels are only applied at creation time. When reconnecting via claim_name, the existing labels on the SandboxClaim are preserved.

Configuration reference

Development

# Clone and install
git clone https://github.com/uesleilima/langchain-k8s.git
cd langchain-k8s
uv sync

# Run unit tests (no cluster needed)
uv run pytest tests/unit/ -v

# Lint and type check
uv run ruff check src/ tests/
uv run pyright src/

Integration tests with Kind

Integration tests require a Kubernetes cluster. The repository includes scripts and manifests to set up a Kind cluster with everything needed.

Prerequisites: kind, kubectl, docker

# Create the Kind cluster and deploy agent-sandbox components
./scripts/kind-setup.sh

# Run integration tests
uv run pytest tests/integration/ -v -m integration

# Tear down when done
./scripts/kind-teardown.sh

The setup script will:

1. Create a Kind cluster named langchain-k8s
2. Install the agent-sandbox controller and extension CRDs from the all-in-one release asset (v0.5.4, overridable via AGENT_SANDBOX_VERSION)
3. Deploy the sandbox router
4. Apply the python-sandbox-template SandboxTemplate
5. Apply the python-sandbox-pool SandboxWarmPool (replicas: 0 by default; set WARMPOOL_REPLICAS to pre-warm pods instead)

k8s/
├── sandbox-router.yaml            # Router Deployment + Service
├── sandbox-template.yaml          # SandboxTemplate for Python runtime
└── sandbox-warmpool.yaml          # SandboxWarmPool claims reference

License

MIT — see LICENSE for details.