DEV Community

Syeed Talha
Syeed Talha

Posted on

Build a Thread-Scoped AI Data Science Sandbox with Daytona and Deep Agents

#ai #programming #agentskills #langchain

AI coding agents can write files, install Python packages, run scripts, and create charts. But where should that code run?

Running arbitrary agent-generated code directly on a local laptop or production server can be risky. A sandbox gives an agent an isolated environment where it can use a filesystem and shell commands without directly accessing the host machine.

This article presents a small two-turn data science pipeline using:

  • Daytona for a remote development sandbox
  • Deep Agents for an AI agent with filesystem and shell access
  • LangGraph for conversation memory
  • NVIDIA NIM as the LLM provider

The objective is not only to train a model. The example demonstrates that:

  1. Python packages installed in the first turn remain available in the second turn.
  2. Files created in the first turn still exist in the second turn.
  3. The agent retains conversation context when the same thread_id is reused.

What the example builds

The program makes two agent calls:

Call Agent task Persistence proof
Call 1 Installs Python packages, generates data, trains a model, and saves a chart Creates data.csv, model.pkl, and chart.png
Call 2 Loads the saved model, predicts a class, and checks the chart Reads files created during Call 1

Both calls use the same sandbox name:

THREAD_ID = "ds-pipeline-demo-1"
SANDBOX_NAME = f"thread-{THREAD_ID}"
Enter fullscreen mode Exit fullscreen mode

This produces the following sandbox name:

thread-ds-pipeline-demo-1
Enter fullscreen mode Exit fullscreen mode

When Call 2 begins, the program searches for a Daytona sandbox with that name. If one exists, the program reuses it instead of creating a new environment.

Understanding the two types of persistence

It is important to separate conversation state from sandbox state.

State type Stored by What it contains Key used in this example
Conversation memory LangGraph checkpointer Earlier messages and agent state thread_id
Execution environment Daytona sandbox Files, installed packages, and workspace state thread-<thread_id>

The same thread_id serves two different purposes.

config = {"configurable": {"thread_id": THREAD_ID}}
Enter fullscreen mode Exit fullscreen mode

This gives LangGraph a stable identifier for the conversation.

The sandbox name is then derived from the same value:

SANDBOX_NAME = f"thread-{THREAD_ID}"
Enter fullscreen mode Exit fullscreen mode

This gives Daytona a predictable name for the remote workspace.

A thread_id does not automatically make Daytona reuse a sandbox. The application must implement a mapping strategy. In this example, the mapping is a stable sandbox name.

Architecture

The overall flow looks like this:

User prompt
   |
   v
Deep Agent running in the application
   |
   +-- LangGraph checkpointer
   |      |
   |      +-- Stores conversation state using thread_id
   |
   v
DaytonaSandbox backend
   |
   +-- File operations
   +-- Shell command execution
   |
   v
Remote Daytona sandbox
   |
   +-- /home/daytona/train.py
   +-- /home/daytona/data.csv
   +-- /home/daytona/model.pkl
   +-- /home/daytona/chart.png
Enter fullscreen mode Exit fullscreen mode

The agent runs in the application, while filesystem and shell operations run in the Daytona environment.

Prerequisites

Install the required packages:

pip install daytona langchain-daytona deepagents \
  langchain-nvidia-ai-endpoints python-dotenv
Enter fullscreen mode Exit fullscreen mode

Create a .env file:

DAYTONA_API_KEY=your_daytona_api_key
NVIDIA_API_KEY=your_nvidia_api_key
Enter fullscreen mode Exit fullscreen mode

The .env file and API keys should not be committed to version control.

Step 1: Create shared configuration

Start by loading environment variables and setting up the Daytona client.

from dotenv import load_dotenv
from daytona import Daytona, CreateSandboxFromSnapshotParams
from langchain_daytona import DaytonaSandbox
from langgraph.checkpoint.memory import InMemorySaver
from deepagents import create_deep_agent

load_dotenv()

DAYTONA = Daytona()

# This checkpointer stays alive during this Python process.
CHECKPOINTER = InMemorySaver()

# One ID represents one conversation and one sandbox workspace.
THREAD_ID = "ds-pipeline-demo-1"
SANDBOX_NAME = f"thread-{THREAD_ID}"

# Remove the sandbox after 60 minutes of inactivity.
SANDBOX_TTL_MINUTES = 60
Enter fullscreen mode Exit fullscreen mode

What does InMemorySaver() do?

InMemorySaver() stores LangGraph state in memory. Since both agent calls happen inside the same Python script, the second call can access the conversation state created by the first call.

There is an important limitation:

InMemorySaver() is not durable. If the application stops and starts again, the conversation memory is lost.

The Daytona sandbox may still be reusable if it has not expired, because it is found by name. However, a production application should use a persistent LangGraph checkpointer backed by a database or other durable storage service.

Step 2: Find or create a sandbox

This function is the core of the sandbox reuse pattern:

def get_or_create_sandbox(name: str):
    """Reuse a sandbox with this name, or create it if missing."""

    for sb in DAYTONA.list():
        if getattr(sb, "name", None) == name:
            print(
                f"Reusing existing sandbox: "
                f"{name} (id={sb.id}, state={sb.state})"
            )

            # Start it again if it is stopped or paused.
            if str(sb.state).lower() not in ("running", "started"):
                sb.start()

            return sb

    print(
        f"Creating new sandbox: {name} "
        f"(ttl={SANDBOX_TTL_MINUTES}min)..."
    )

    sb = DAYTONA.create(
        CreateSandboxFromSnapshotParams(
            name=name,
            language="python",
            ttl_minutes=SANDBOX_TTL_MINUTES,
        )
    )

    print(f"Created sandbox: {name} (id={sb.id})")
    return sb
Enter fullscreen mode Exit fullscreen mode

Here is what happens in different situations:

Situation Result
No sandbox with the target name exists A new sandbox is created
A matching sandbox exists and is running It is reused immediately
A matching sandbox exists but is stopped It is started and reused
The sandbox expired because of TTL A new sandbox is created

Files and installed packages exist inside the Daytona sandbox, not in the LangGraph checkpointer.

That is why Call 2 can load the model:

Call 1 writes /home/daytona/model.pkl
                 |
                 v
The same Daytona sandbox is reused
                 |
                 v
Call 2 reads /home/daytona/model.pkl
Enter fullscreen mode Exit fullscreen mode

Step 3: Give the agent access to the sandbox

Next, the Daytona sandbox is wrapped in a Deep Agents backend.

def build_agent(sandbox):
    """Create an AI agent connected to one Daytona sandbox."""

    backend = DaytonaSandbox(sandbox=sandbox)

    return create_deep_agent(
        model="nvidia:nvidia/nemotron-3-ultra-550b-a55b",
        backend=backend,
        checkpointer=CHECKPOINTER,
        system_prompt=(
            "You are a data-science coding assistant with sandbox access. "
            "Use the filesystem and execute tools to install packages, "
            "write scripts, run them, and inspect outputs. "
            "Always verify files exist before reporting success.\n\n"
            "The writable working directory is /home/daytona. "
            "Use absolute paths under /home/daytona/ for file operations. "
            "When executing commands, first run: cd /home/daytona"
        ),
    )
Enter fullscreen mode Exit fullscreen mode

The sandbox backend gives the agent tools for tasks such as:

  • Writing files
  • Reading files
  • Editing files
  • Listing directories
  • Searching files
  • Running shell commands

These tools allow the agent to install packages, create Python scripts, run them, and inspect generated artifacts.

Why /home/daytona matters

The system prompt directs the agent to work only inside:

/home/daytona
Enter fullscreen mode Exit fullscreen mode

This convention keeps generated files in one writable and predictable location.

For example:

/home/daytona/train.py
/home/daytona/data.csv
/home/daytona/model.pkl
/home/daytona/chart.png
Enter fullscreen mode Exit fullscreen mode

Using absolute paths also reduces the chance of files being written to an unexpected location or to a protected root-level directory.

Step 4: Run the training turn

For the first call, the agent receives instructions to install dependencies, create a training script, execute it, and verify the outputs.

call1_prompt = (
    "Do the following in the sandbox, step by step:\n"
    "1. Install: pandas matplotlib scikit-learn (use pip).\n"
    "2. Write a Python script `train.py` that:\n"
    "   - generates a small synthetic classification dataset,\n"
    "   - saves the data to `data.csv`,\n"
    "   - trains a sklearn LogisticRegression model,\n"
    "   - saves the trained model to `model.pkl`,\n"
    "   - creates and saves a scatter plot as `chart.png`.\n"
    "3. Run `train.py` and verify `model.pkl` and `chart.png` exist.\n"
    "4. Report what you did and the file sizes."
)
Enter fullscreen mode Exit fullscreen mode

Inside the sandbox, the workflow is similar to:

cd /home/daytona
pip install pandas matplotlib scikit-learn
python train.py
ls -lh model.pkl chart.png
Enter fullscreen mode Exit fullscreen mode

After the first call, the Daytona workspace should contain:

/home/daytona/
├── train.py
├── data.csv
├── model.pkl
└── chart.png
Enter fullscreen mode Exit fullscreen mode

Step 5: Reuse the sandbox in the second call

Before the second call, the same lookup function runs again:

sandbox = get_or_create_sandbox(SANDBOX_NAME)
agent = build_agent(sandbox)
Enter fullscreen mode Exit fullscreen mode

Since SANDBOX_NAME is unchanged, the program finds and reuses the existing environment.

The next prompt uses artifacts from the earlier turn:

call2_prompt = (
    "Using the artifacts from the previous turn in the sandbox:\n"
    "1. Load `model.pkl` and run a prediction on this new row:\n"
    "   feature_1 = 0.5, feature_2 = -1.2\n"
    "   Print the predicted class.\n"
    "2. Confirm `chart.png` is still present and report its size and "
    "the list of files in the working directory.\n"
    "This proves the installed packages and files survived between turns."
)
Enter fullscreen mode Exit fullscreen mode

A successful second call demonstrates that:

scikit-learn is still installed
model.pkl still exists
chart.png still exists
Enter fullscreen mode Exit fullscreen mode

This is the practical value of a thread-scoped sandbox.

Streaming tool activity

The run_turn() function streams the agent’s actions while work is in progress:

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": prompt}]},
    config=config,
    stream_mode="updates",
):
    ...
Enter fullscreen mode Exit fullscreen mode

This makes the workflow observable:

[agent] -> execute({"command": "cd /home/daytona && pip install ..."})
[tools] <- execute: Successfully installed ...
[agent] -> write_file({"path": "/home/daytona/train.py", ...})
[agent] -> execute({"command": "cd /home/daytona && python train.py"})
[tools] <- execute: Model saved successfully
Enter fullscreen mode Exit fullscreen mode

Streaming is useful for code-running agents because it shows:

  • Which commands were executed
  • Which files were written
  • Whether a command failed
  • Whether the agent verified the final files

Step 6: Clean up the sandbox

The example uses a finally block to stop the sandbox after the script ends:

finally:
    try:
        sb = DAYTONA.get(SANDBOX_NAME)
        sb.stop()
        print(f"Stopped sandbox: {SANDBOX_NAME} (id={sb.id})")
    except Exception as exc:
        print(f"Could not stop sandbox {SANDBOX_NAME}: {exc}")
Enter fullscreen mode Exit fullscreen mode

This matters because sandboxes consume remote compute resources.

The example also configures:

ttl_minutes=60
Enter fullscreen mode Exit fullscreen mode

The TTL is a backup cleanup mechanism. If the application crashes or the session does not return, the sandbox can be removed after it remains idle for the configured period.

In this demo, the explicit stop() runs after Call 2, so the TTL mainly serves as a fallback.

Security considerations

A sandbox provides isolation between agent-generated code and the local machine or primary application server.

However, a sandbox does not solve every security issue automatically.

Risk Why it still matters
Prompt injection Untrusted input could convince an agent to run unwanted commands inside the sandbox
Network access If outbound networking is allowed, the agent may send data elsewhere
Secrets in prompts Sensitive information passed to the agent could be written into sandbox files
Resource usage An agent may install large packages, create large files, or run expensive commands

Practical safety rules include:

  • Keep API keys outside the sandbox when possible.
  • Avoid placing sensitive production data into the sandbox unless necessary.
  • Use short sandbox TTL values.
  • Use unique and difficult-to-guess thread IDs in multi-user systems.
  • Validate user-provided file paths and commands.
  • Apply runtime, storage, package, and network restrictions where available.
  • Clean up environments when tasks finish.

Full code 

"""
Thread-scoped Daytona Sandbox: multi-turn data science pipeline.

Demonstrates sandbox + conversation reuse across two calls with the
same `thread_id`:

  Call 1: install pandas/matplotlib/scikit-learn, generate a CSV, train a
          small model, save `model.pkl` + `chart.png`.
  Call 2: load the saved model and predict on a new row; re-display the
          chart. Proves both the installed packages and the files
          survived between turns.

How reuse works:
  - The sandbox is named `thread-<thread_id>`. On Call 2 we list existing
    sandboxes and find it by name -> reuse it (files + installed packages
    are still there). If not found, we create it.
  - A LangGraph checkpointer keyed by the same `thread_id` keeps the
    conversation memory across the two invocations.
  - `ttl_minutes` auto-deletes the sandbox after idle time so you don't
    pay forever if you never come back.

Prereqs (already in pyproject.toml):
    pip install daytona langchain-daytona deepagents langchain-nvidia-ai-endpoints

Env (already in .env):
    DAYTONA_API_KEY     - for Daytona()
    NVIDIA_API_KEY      - for the LLM

Run:
    python daytona_thread_scoped.py
"""

from dotenv import load_dotenv
from daytona import Daytona, CreateSandboxFromSnapshotParams
from langchain_daytona import DaytonaSandbox
from langgraph.checkpoint.memory import InMemorySaver

from deepagents import create_deep_agent

# Load DAYTONA_API_KEY / NVIDIA_API_KEY from .env
load_dotenv()

DAYTONA = Daytona()

# A single checkpointer kept alive for the whole script so both calls
# share conversation memory keyed by thread_id.
CHECKPOINTER = InMemorySaver()

# The conversation/sandbox key. Same value on both calls => reuse.
THREAD_ID = "ds-pipeline-demo-1"
SANDBOX_NAME = f"thread-{THREAD_ID}"
SANDBOX_TTL_MINUTES = 60  # auto-delete after 1h idle


def get_or_create_sandbox(name: str):
    """Find an existing Daytona sandbox by name, or create a new one.

    Returns the (running) Sandbox. Reusing the same name across turns is
    what makes files and installed packages persist between calls.
    """
    for sb in DAYTONA.list():
        if getattr(sb, "name", None) == name:
            print(f"Reusing existing sandbox: {name} (id={sb.id}, state={sb.state})")
            # Make sure it's running (it may have been stopped/paused).
            if str(sb.state).lower() not in ("running", "started"):
                sb.start()
            return sb

    print(f"Creating new sandbox: {name} (ttl={SANDBOX_TTL_MINUTES}min)...")
    sb = DAYTONA.create(
        CreateSandboxFromSnapshotParams(
            name=name,
            language="python",
            ttl_minutes=SANDBOX_TTL_MINUTES,
        )
    )
    print(f"Created sandbox: {name} (id={sb.id})")
    return sb


def build_agent(sandbox):
    """Build a Deep Agent bound to the given sandbox backend."""
    backend = DaytonaSandbox(sandbox=sandbox)
    return create_deep_agent(
        model="nvidia:nvidia/nemotron-3-ultra-550b-a55b",
        backend=backend,
        checkpointer=CHECKPOINTER,
        system_prompt=(
            "You are a data-science coding assistant with sandbox access. "
            "Use the provided filesystem and execute tools to install "
            "packages, write scripts, run them, and read outputs. "
            "Always verify results (e.g. check that a file exists) before "
            "reporting success. Keep shell commands short.\n\n"
            "IMPORTANT -- working directory:\n"
            "- The sandbox's writable working directory is `/home/daytona`. "
            "The filesystem root `/` is NOT writable.\n"
            "- When using write_file / read_file / edit_file, ALWAYS pass "
            "absolute paths under `/home/daytona/` "
            "(e.g. `/home/daytona/train.py`, NOT `train.py` or `/train.py`).\n"
            "- When using the execute tool, first `cd /home/daytona` so that "
            "relative paths in shell commands also land there.\n"
            "- Keep all created files (scripts, data.csv, model.pkl, "
            "chart.png) under `/home/daytona/`."
        ),
    )


def run_turn(agent, config, label: str, prompt: str) -> str:
    """Invoke the agent for one turn, streaming tool calls and results."""
    print("\n" + "=" * 72)
    print(f"{label}")
    print("=" * 72)
    print(f"USER: {prompt}\n")

    final_text = ""
    for chunk in agent.stream(
        {"messages": [{"role": "user", "content": prompt}]},
        config=config,
        stream_mode="updates",
    ):
        for node_name, update in chunk.items():
            if not update:
                continue
            for msg in update.get("messages", []):
                if msg.type == "ai" and getattr(msg, "tool_calls", None):
                    for call in msg.tool_calls:
                        name = call.get("name")
                        args = call.get("args", {})
                        summary = str(args)
                        if len(summary) > 200:
                            summary = summary[:200] + "..."
                        print(f"[{node_name}] -> {name}({summary})")
                elif msg.type == "tool":
                    content = str(msg.content)
                    if len(content) > 300:
                        content = content[:300] + "..."
                    print(f"[{node_name}] <- {msg.name}: {content}")
                elif msg.type == "ai" and not getattr(msg, "tool_calls", None):
                    final_text = msg.content
                    print(f"[{node_name}] AI: {final_text}\n")
    return final_text


def main():
    config = {"configurable": {"thread_id": THREAD_ID}}

    # --- Call 1: create sandbox + run the data-science pipeline -------
    sandbox = get_or_create_sandbox(SANDBOX_NAME)
    agent = build_agent(sandbox)

    call1_prompt = (
        "Do the following in the sandbox, step by step:\n"
        "1. Install: pandas matplotlib scikit-learn (use pip).\n"
        "2. Write a Python script `train.py` that:\n"
        "   - generates a small synthetic classification dataset (e.g. "
        "make_classification, ~200 rows, 2 numeric features),\n"
        "   - saves the data to `data.csv`,\n"
        "   - trains a sklearn LogisticRegression model,\n"
        "   - saves the trained model to `model.pkl` using joblib/pickle,\n"
        "   - plots the two features colored by class and saves it as "
        "`chart.png` (do not call plt.show).\n"
        "3. Run `train.py` and verify `model.pkl` and `chart.png` exist.\n"
        "4. Report what you did and the file sizes."
    )
    run_turn(agent, config, "CALL 1: train + save artifacts", call1_prompt)

    # --- Call 2: reuse sandbox + memory; prove files survived ---------
    # Same thread_id => same checkpointer memory AND same sandbox.
    sandbox = get_or_create_sandbox(SANDBOX_NAME)
    agent = build_agent(sandbox)  # rebuild bound to the (reused) sandbox

    call2_prompt = (
        "Using the artifacts from the previous turn in the sandbox:\n"
        "1. Load `model.pkl` and run a prediction on this new row:\n"
        "   feature_1 = 0.5, feature_2 = -1.2\n"
        "   Print the predicted class.\n"
        "2. Confirm `chart.png` is still present and report its size and "
        "the list of files in the working directory.\n"
        "This proves the installed packages and files survived between turns."
    )
    run_turn(agent, config, "CALL 2: load model + reuse files", call2_prompt)


if __name__ == "__main__":
    try:
        main()
    finally:
        # Clean up the sandbox so it doesn't keep running / billing.
        # (ttl_minutes would also clean it up, but we stop it now to be safe.)
        try:
            sb = DAYTONA.get(SANDBOX_NAME)
            sb.stop()
            print(f"\nStopped sandbox: {SANDBOX_NAME} (id={sb.id})")
        except Exception as exc:
            print(f"\nCould not stop sandbox {SANDBOX_NAME}: {exc}")
Enter fullscreen mode Exit fullscreen mode

Final takeaway

A thread-scoped sandbox gives each AI conversation its own isolated workspace.

In this example:

  1. THREAD_ID identifies the conversation.
  2. thread-<THREAD_ID> identifies the Daytona sandbox.
  3. Call 1 installs packages and creates artifacts.
  4. Call 2 finds the same sandbox and reuses its packages and files.
  5. LangGraph memory preserves conversational context during the script.
  6. Daytona TTL and explicit cleanup help control resource usage.

This pattern can be used for more than a small machine-learning example:

  • AI coding assistants
  • Data analysis agents
  • File-processing workflows
  • Report and chart generation
  • Automated testing environments
  • Per-user development workspaces

Conversation state and execution state are separate, but a shared thread identifier can connect them in a clean and reliable way.

Top comments (0)