Daytona Executor: Sandbox Code Execution for AG2 Agents#

import os

from autogen import ConversableAgent, LLMConfig
from autogen.coding import DaytonaCodeExecutor

# Ensure required API keys are set
assert os.getenv("DAYTONA_API_KEY"), "Please set DAYTONA_API_KEY environment variable"
assert os.getenv("OPENAI_API_KEY"), "Please set OPENAI_API_KEY environment variable (or configure another LLM provider)"

# Create a Daytona executor — a sandbox is provisioned immediately
executor = DaytonaCodeExecutor(timeout=60)

# Create an executor agent that runs code but doesn't call an LLM
code_executor_agent = ConversableAgent(
    "code_executor_agent",
    llm_config=False,
    code_execution_config={"executor": executor},
    human_input_mode="NEVER",
)

print(f"Sandbox ready — ID: {executor._sandbox.id}")

# Execute a simple Python block
task = """
Run this Python code:
```python
import datetime
import math

now = datetime.datetime.now()
print(f"Current time (inside Daytona sandbox): {now}")
print(f"Pi to 10 decimal places: {math.pi:.10f}")
print(f"Square root of 1764: {math.sqrt(1764):.1f}")
```
"""

reply = code_executor_agent.generate_reply(messages=[{"role": "user", "content": task}])
print(reply)

# Don't forget to clean up when done with a standalone executor
executor.delete()

llm_config = LLMConfig({
    "model": "gpt-4o-mini",
    "api_key": os.environ["OPENAI_API_KEY"],
})

code_writer_system_message = """
You are a helpful AI assistant that writes and executes Python code.
When writing code:
- Use only the Python standard library or packages available in the default Python environment
- Include informative print statements so results are clearly visible
- Write complete, runnable code blocks
Reply TERMINATE when the task is complete.
"""

with DaytonaCodeExecutor(timeout=120) as executor:
    code_writer_agent = ConversableAgent(
        "code_writer",
        system_message=code_writer_system_message,
        llm_config=llm_config,
        max_consecutive_auto_reply=5,
        code_execution_config=False,
    )

    code_executor_agent = ConversableAgent(
        "code_executor",
        llm_config=False,
        code_execution_config={"executor": executor},
        human_input_mode="NEVER",
        is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    )

    chat_result = code_executor_agent.run(
        recipient=code_writer_agent,
        message=(
            "Generate a list of the first 20 Fibonacci numbers, "
            "compute their sum and mean, and print a formatted table."
        ),
        max_turns=6,
    ).process()

print("\nSandbox automatically deleted — context manager exited.")

SNAPSHOT_NAME = os.getenv("DAYTONA_SNAPSHOT", "my-data-science-snapshot")

with DaytonaCodeExecutor(snapshot=SNAPSHOT_NAME, timeout=120) as executor:
    agent = ConversableAgent(
        "snapshot_agent",
        llm_config=False,
        code_execution_config={"executor": executor},
        human_input_mode="NEVER",
    )

    reply = agent.generate_reply(
        messages=[
            {
                "role": "user",
                "content": """
Run this Python code:
```python
import sys, platform
print(f"Python  : {sys.version}")
print(f"Platform: {platform.platform()}")
print("Snapshot sandbox is ready!")
```
""",
            }
        ]
    )
    print(reply)

from daytona import Image

# Declarative image build: start from a slim Python base and add packages
custom_image = Image.base("python:3.12-slim").pip_install("numpy", "pandas", "scipy")

with DaytonaCodeExecutor(image=custom_image, timeout=180) as executor:
    agent = ConversableAgent(
        "data_agent",
        llm_config=False,
        code_execution_config={"executor": executor},
        human_input_mode="NEVER",
    )

    task = """
Run this Python code:
```python
import numpy as np
import pandas as pd
from scipy import stats

# Generate random sales data
rng = np.random.default_rng(42)
sales = rng.normal(loc=500, scale=150, size=200).clip(0)

df = pd.DataFrame({"sales": sales})
print(df["sales"].describe().round(2).to_string())
print(f"\nSkewness : {stats.skew(sales):.4f}")
print(f"Kurtosis : {stats.kurtosis(sales):.4f}")
```
"""

    reply = agent.generate_reply(messages=[{"role": "user", "content": task}])
    print(reply)

with DaytonaCodeExecutor(image="python:3.12-slim", timeout=60) as executor:
    agent = ConversableAgent(
        "slim_agent",
        llm_config=False,
        code_execution_config={"executor": executor},
        human_input_mode="NEVER",
    )

    reply = agent.generate_reply(
        messages=[
            {
                "role": "user",
                "content": """
Run this Python code:
```python
import sys
print(f"Python {sys.version}")
print("Running in a slim Python 3.12 sandbox!")
```
""",
            }
        ]
    )
    print(reply)

from autogen.coding import CodeBlock

with DaytonaCodeExecutor(timeout=60) as executor:
    # Bash — list files and show system info
    bash_result = executor.execute_code_blocks([
        CodeBlock(language="bash", code="echo 'Hello from Bash!' && uname -a && echo 'Disk usage:' && df -h /"),
    ])
    print("=== Bash ===")
    print(bash_result.output)

    # JavaScript — run via Node.js
    js_result = executor.execute_code_blocks([
        CodeBlock(
            language="javascript",
            code=(
                "const nums = Array.from({length: 10}, (_, i) => i + 1);\n"
                "const sum = nums.reduce((a, b) => a + b, 0);\n"
                "console.log('Numbers:', nums.join(', '));\n"
                "console.log('Sum:', sum);\n"
                "console.log('Mean:', sum / nums.length);"
            ),
        )
    ])
    print("\n=== JavaScript ===")
    print(js_result.output)

    # TypeScript — transpiled on-the-fly via ts-node
    ts_result = executor.execute_code_blocks([
        CodeBlock(
            language="typescript",
            code=(
                "function greet(name: string): string {\n"
                "  return `Hello, ${name}! Running TypeScript in Daytona.`;\n"
                "}\n"
                "console.log(greet('AG2'));"
            ),
        )
    ])
    print("\n=== TypeScript ===")
    print(ts_result.output)

# Secrets are injected into the sandbox environment at creation time
with DaytonaCodeExecutor(
    timeout=60,
    env_vars={
        "MY_API_KEY": "super-secret-key-123",
        "DATABASE_URL": "postgresql://user:pass@host/db",
        "APP_ENV": "production",
    },
) as executor:
    agent = ConversableAgent(
        "secrets_agent",
        llm_config=False,
        code_execution_config={"executor": executor},
        human_input_mode="NEVER",
    )

    reply = agent.generate_reply(
        messages=[
            {
                "role": "user",
                "content": """
Run this Python code:
```python
import os

api_key = os.environ.get("MY_API_KEY", "not set")
db_url = os.environ.get("DATABASE_URL", "not set")
app_env = os.environ.get("APP_ENV", "not set")

# Mask the secret values in output
print(f"MY_API_KEY   : {api_key[:8]}... (masked)")
print(f"DATABASE_URL : {db_url[:20]}... (masked)")
print(f"APP_ENV      : {app_env}")
```
""",
            }
        ]
    )
    print(reply)

data_science_image = Image.base("python:3.12-slim").pip_install("numpy", "pandas", "scipy", "scikit-learn")

llm_config = LLMConfig({
    "model": "gpt-4o-mini",
    "api_key": os.environ["OPENAI_API_KEY"],
})

system_message = """
You are a data science assistant. Write Python code using numpy, pandas, scipy,
and scikit-learn. The sandbox has these libraries pre-installed.
Include clear print statements and format numbers to 4 decimal places.
When the results of your code are executed by the executor and have a successful execution result, reply with TERMINATE to end the workflow.
"""

with DaytonaCodeExecutor(image=data_science_image, timeout=180) as executor:
    writer = ConversableAgent(
        "data_scientist",
        system_message=system_message,
        llm_config=llm_config,
        max_consecutive_auto_reply=5,
        code_execution_config=False,
    )

    runner = ConversableAgent(
        "executor",
        llm_config=False,
        code_execution_config={"executor": executor},
        human_input_mode="NEVER",
        is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    )

    chat_result = runner.run(
        writer,
        message=(
            "Generate a synthetic dataset with 300 samples and 4 features "
            "(make_classification from sklearn). "
            "Train a logistic regression model, evaluate it with cross-validation, "
            "and print the mean accuracy, precision, recall, and F1 score."
        ),
        max_turns=8,
    ).process()

print("\nSandbox deleted — workflow complete.")