Add Docker sandbox

[InfernalAzazel]

Submission checklist

• This is a feature request, not a bug report.
• I searched existing issues and didn't find this feature.
• I checked the docs and README for existing functionality.
• This request applies to this repo (deepagents) and not an external package.

Area (Required)

• deepagents (SDK)
• cli
• acp
• evals
• harbor
• repl
• daytona
• modal
• quickjs
• runloop
• langsmith-sandbox
• Other / not sure / general

Feature description

At present, solutions such as Modal, Runloop, Daytona, and LangSmith do not directly provide Docker sandbox capabilities. The advantage of this built-in sandbox mechanism is that it can avoid dependence on additional Docker environments and simplify the overall architecture.

Proposed solution (optional)

e.g:

• docker box

"""Docker daemon + DeepAgents backend, layered like Modal + langchain_modal.

Mirrors::

    app = modal.App.lookup("your-app")
    modal_sandbox = modal.Sandbox.create(app=app)
    backend = ModalSandbox(sandbox=modal_sandbox)

with::

    app = DockerApp.lookup("your-app")
    docker_sandbox = DockerSandbox.create(app=app)
    backend = DockerBackend(sandbox=docker_sandbox)
"""

from __future__ import annotations

import contextlib
import io
import shlex
import tarfile
import time
from pathlib import PurePosixPath
from typing import TYPE_CHECKING, Any

import docker as docker_sdk
from docker import DockerClient

from deepagents.backends.protocol import (
    ExecuteResponse,
    FileDownloadResponse,
    FileUploadResponse,
)
from deepagents.backends.sandbox import BaseSandbox

if TYPE_CHECKING:
    from docker.models.containers import Container


class DockerApp:
    """Docker daemon scope (``modal.App`` analogue — holds :class:`~docker.DockerClient`)."""

    __slots__ = ("_client",)

    def __init__(self, client: DockerClient | None = None) -> None:
        self._client = client

    @classmethod
    def from_env(cls) -> DockerApp:
        return cls()

    @classmethod
    def lookup(cls, name: str | None = None) -> DockerApp:
        """Connect to Docker. ``name`` is unused (no Modal-style registry locally)."""
        _ = name
        return cls.from_env()

    @property
    def client(self) -> DockerClient:
        if self._client is None:
            self._client = docker_sdk.from_env()
        return self._client


class DockerSandbox:
    """Running container (``modal.Sandbox`` analogue); built via :meth:`create`."""

    __slots__ = ("_container",)

    def __init__(self, container: Container) -> None:
        self._container = container

    @classmethod
    def create(
        cls,
        app: DockerApp,
        *,
        image: str = "python:3.12-slim",
        command: list[str] | None = None,
        detach: bool = True,
        remove: bool = True,
        **run_kwargs: Any,
    ) -> DockerSandbox:
        """Start a long-lived container for ``exec_run`` / archive APIs."""
        cmd = command if command is not None else ["tail", "-f", "/dev/null"]
        container = app.client.containers.run(
            image,
            command=cmd,
            detach=detach,
            remove=remove,
            **run_kwargs,
        )
        return cls(container)

    @property
    def id(self) -> str:
        return self._container.id

    @property
    def container(self) -> Container:
        return self._container

    def stop(self, **kwargs: Any) -> None:
        self._container.stop(**kwargs)

    def __enter__(self) -> DockerSandbox:
        return self

    def __exit__(self, *exc: Any) -> None:
        with contextlib.suppress(Exception):
            self.stop()


class DockerBackend(BaseSandbox):
    """DeepAgents backend over :class:`DockerSandbox` (``ModalSandbox`` analogue)."""

    def __init__(self, *, sandbox: DockerSandbox) -> None:
        self._sandbox = sandbox
        self._timeout: int = 30 * 60

    @property
    def id(self) -> str:
        return self._sandbox.id

    def _read_file(self, path: str) -> FileDownloadResponse:
        if not path.startswith("/"):
            return FileDownloadResponse(path=path, content=None, error="invalid_path")

        try:
            strm, stat = self._sandbox.container.get_archive(path)
            file_like_object = io.BytesIO(b"".join(chunk for chunk in strm))
            with tarfile.open(fileobj=file_like_object, mode="r") as tar:
                file_name = stat.get("name") if isinstance(stat, dict) else None
                member = file_name or PurePosixPath(path).name
                with tar.extractfile(member) as f:
                    if f is None:
                        return FileDownloadResponse(
                            path=path, content=None, error="extract_failed"
                        )
                    return FileDownloadResponse(path=path, content=f.read(), error=None)
        except Exception as e:
            msg = str(e).lower()
            if "no such file" in msg or "not found" in msg:
                return FileDownloadResponse(path=path, content=None, error="file_not_found")
            if "is a directory" in msg:
                return FileDownloadResponse(path=path, content=None, error="is_directory")
            return FileDownloadResponse(path=path, content=None, error="read_failed")

    def _write_file(self, path: str, content: bytes) -> FileUploadResponse:
        if not path.startswith("/"):
            return FileUploadResponse(path=path, error="invalid_path")

        posix_path = PurePosixPath(path)
        parent_dir = str(posix_path.parent)
        filename = posix_path.name
        if not filename:
            return FileUploadResponse(path=path, error="invalid_path")

        try:
            mkdir_result = self._sandbox.container.exec_run(
                cmd=f"mkdir -p {shlex.quote(parent_dir)}",
                user="root",
                workdir="/root",
            )
            if mkdir_result.exit_code != 0:
                return FileUploadResponse(path=path, error="mkdir_failed")

            tar_stream = io.BytesIO()
            with tarfile.open(fileobj=tar_stream, mode="w") as tar:
                tarinfo = tarfile.TarInfo(name=filename)
                tarinfo.size = len(content)
                tarinfo.mtime = int(time.time())
                tarinfo.mode = 0o644
                tar.addfile(tarinfo, io.BytesIO(content))
            tar_stream.seek(0)
            self._sandbox.container.put_archive(parent_dir, tar_stream)
            return FileUploadResponse(path=path, error=None)
        except Exception as e:
            msg = str(e).lower()
            if "permission denied" in msg:
                return FileUploadResponse(path=path, error="permission_denied")
            return FileUploadResponse(path=path, error="write_failed")

    def execute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse:
        _ = timeout
        result = self._sandbox.container.exec_run(
            cmd=command, user="root", workdir="/root"
        )
        output = (
            result.output.decode("utf-8", errors="replace") if result.output else ""
        )
        return ExecuteResponse(output=output, exit_code=result.exit_code, truncated=False)

    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
        return [self._read_file(path) for path in paths]

    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
        return [self._write_file(path, content) for path, content in files]

• use docker box

from deepagents import create_deep_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver

from sandbox import DockerApp, DockerBackend, DockerSandbox

app = DockerApp.lookup("your-app")
with DockerSandbox.create(app=app) as docker_sandbox:
    backend = DockerBackend(sandbox=docker_sandbox)

    checkpointer = MemorySaver()

    model = init_chat_model(
        model="deepseek:deepseek-chat",
        api_key="sk-xxxx",
    )
    agent = create_deep_agent(
        model=model,
        backend=backend,
        # skills=["/Users/kylin/work/code/github/langdemo/skills/"],
        interrupt_on={
            "write_file": True,  # Default: approve, edit, reject
            "read_file": False,  # No interrupts needed
            "edit_file": True    # Default: approve, edit, reject
        },
        checkpointer=checkpointer,
    )

    result = agent.invoke(
        {
            "messages": [
                {"role": "user", "content": "运行一个python脚本，脚本内容为：print('Hello, World!')"},
            ],
        },
        config={"configurable": {"thread_id": "12345"}},
    )

    print(result["messages"][-1].content)

• display effect

(langdemo) kylin@mrkylindeMacBook-Pro langdemo % python main.py
已运行成功，输出：`Hello, World!`

Additional context (optional)

No response

[eyurtsev]

Hi kylin (@InfernalAzazel), are you interested on working on this functionality?

[InfernalAzazel]

Hi kylin (@InfernalAzazel), are you interested on working on this functionality?

Thanks for reaching out! I'm not planning to implement this myself, but I'm happy to share ideas or discuss the design if helpful.

reference: pydantic-deepagents

[keenborder786]

Eugene Yurtsev (@eyurtsev) I would love to work on this, possible if you can reassign to me?

[andybbruno]

Hi all 👋

For anyone who needs this today, I built deepagents-docker: a Docker sandbox backend for deepagents, available on PyPI as deepagents-docker.

Install with uv:

uv add deepagents-docker

or with pip:

pip install deepagents-docker

Example:

from deepagents import create_deep_agent
from deepagents_docker import DockerSandbox

agent = create_deep_agent(
    model="openai:gpt-5.5",
    backend=DockerSandbox(),
    system_prompt="You are a research assistant.",
)

result = agent.invoke({"messages": "Research the latest trends in AI and write a summary."})

Repo: https://github.com/andybbruno/deepagents-docker

Happy to help with an upstream PR if the maintainers want to pull it in or use it as a starting point.

[zhengjie9510]

Hi all 👋

I've also been working on a Docker sandbox backend — langchain-docker-backend (https://github.com/zhengjie9510/langchain-docker-backend), available on PyPI:

pip install langchain-docker-backend

It implements BaseSandbox and works as a drop-in backend for both deepagents and LangChain (via FilesystemMiddleware):

from deepagents import create_deep_agent
from langchain_docker_backend import DockerSandbox

with DockerSandbox() as sandbox:
    agent = create_deep_agent(
        model="openai:gpt-4o",
        backend=sandbox,
    )
    result = agent.invoke({"messages": "Write and run a fizzbuzz script"})

Why this one?

• Dual framework support — works with both deepagents and LangChain out of the box
• Full sandbox API — execute, read, write, edit, ls, plus binary upload_files / download_files
• Any Docker image — python:3.13-slim, node:20, ubuntu:22.04, or your own custom image
• Clean lifecycle — context manager (with ... as sandbox) handles container creation and teardown automatically
• Flexible deployment — local Docker, Docker-in-Docker, or remote Docker via TCP/TLS
• Configurable — timeout, output truncation, volume mounts, working directory