Using the OpenAI Agents SDK with Daytona Sandboxes

from openai.types.responses import ResponseTextDeltaEvent

from agents import Runner
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.sandbox.capabilities import Shell
from agents.sandbox.entries import File
from agents.extensions.sandbox import DaytonaSandboxClient, DaytonaSandboxClientOptions

DAYTONA_ROOT = "/home/daytona/workspace"

# Declare workspace contents declaratively
# Use Daytona's home directory as root instead of the default /workspace.
manifest = Manifest(root=DAYTONA_ROOT, entries={
    "data/sales.csv": File(content=b"quarter,revenue\nQ1,3200000\nQ2,3600000\nQ3,4200000\nQ4,3900000"),
    "requirements.txt": File(content=b"pandas\nmatplotlib"),
})

agent = SandboxAgent(
    name="Data Analyst",
    model="gpt-5.4",
    instructions=(
        "You're a data analyst with shell access to a sandbox. "
        "Inspect the workspace, install dependencies, write and run code to answer questions."
    ),
    default_manifest=manifest,
    capabilities=[Shell()],
)

client = DaytonaSandboxClient()
run_config = RunConfig(
    sandbox=SandboxRunConfig(client=client, options=DaytonaSandboxClientOptions())
)

result = Runner.run_streamed(
    agent,
    "Which quarter had the highest revenue? Write a script to plot the trend and save it as chart.png.",
    run_config=run_config,
)
async for event in result.stream_events():
    if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
        print(event.data.delta, end="", flush=True)
    elif event.type == "run_item_stream_event":
        if event.name == "tool_called":
            raw = event.item.raw_item
            name = raw.get("name", "") if isinstance(raw, dict) else getattr(raw, "name", "")
            args = raw.get("arguments", "") if isinstance(raw, dict) else getattr(raw, "arguments", "")
            print(f"\n[{name}] {args}")
        elif event.name == "tool_output":
            print(f"  → {event.item.output[:200]}")

await client.close()

client = DaytonaSandboxClient()
session = await client.create(manifest=manifest, options=DaytonaSandboxClientOptions())
await session.start()

run_config = RunConfig(sandbox=SandboxRunConfig(session=session))

conversation = []
while True:
    question = input("> ")
    if question.strip().lower() == "exit":
        break

    input_items = conversation + [{"role": "user", "content": question}]
    result = Runner.run_streamed(agent, input_items, run_config=run_config)

    async for event in result.stream_events():
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            print(event.data.delta, end="", flush=True)
    print()

    # Carry conversation history forward so the agent remembers previous turns
    conversation = result.to_input_list()

await session.aclose()
await client.close()

from pathlib import Path
from agents.extensions.sandbox import (
    DaytonaSandboxClient,
    DaytonaSandboxClientOptions,
    DaytonaSandboxSessionState,
)

STATE_FILE = Path(".session_state.json")

client = DaytonaSandboxClient()
options = DaytonaSandboxClientOptions(pause_on_exit=True)

# Try to resume a previously paused sandbox
session = None
if STATE_FILE.exists():
    saved = DaytonaSandboxSessionState.model_validate_json(STATE_FILE.read_text())
    old_sandbox_id = saved.sandbox_id  # snapshot before resume() mutates it
    try:
        session = await client.resume(saved)
        if session.state.sandbox_id == old_sandbox_id:
            print("Reconnected to existing sandbox.")
        else:
            print("Previous sandbox expired. Created a new one.")
    except Exception:
        session = None  # fall through to fresh creation

if session is None:
    session = await client.create(manifest=manifest, options=options)

# Save state immediately so crashes don't orphan the sandbox
STATE_FILE.write_text(session.state.model_dump_json(indent=2))

# ... run your agent ...

# On clean exit: aclose() persists the workspace, then pauses (or deletes) the remote sandbox
await session.aclose()
await client.close()

from agents import Agent, Runner
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.sandbox.capabilities import Shell
from agents.sandbox.entries import File
from agents.extensions.sandbox import DaytonaSandboxClient, DaytonaSandboxClientOptions

manifest = Manifest(root="/home/daytona/workspace", entries={
    "data/sales.csv": File(content=b"quarter,region,revenue\nQ1,NA,3200000\nQ1,EU,2100000\n..."),
})

# The copywriter receives the analyst's findings (no sandbox needed)
copywriter = Agent(
    name="Client Email Drafter",
    model="gpt-5.4",
    instructions="Turn the analyst's findings into a short, friendly client-facing email.",
)

# The analyst has shell access to crunch data, then hands off to the copywriter
analyst = SandboxAgent(
    name="Data Analyst",
    model="gpt-5.4",
    instructions=(
        "Analyze the sales data in the workspace. Write and run code to compute trends. "
        "Then hand off your findings to the Client Email Drafter."
    ),
    default_manifest=manifest,
    capabilities=[Shell()],
    handoffs=[copywriter],
)

client = DaytonaSandboxClient()
result = await Runner.run(
    analyst,
    "Summarize Q1 performance by region for the client.",
    run_config=RunConfig(sandbox=SandboxRunConfig(client=client, options=DaytonaSandboxClientOptions())),
)
await client.close()
print(result.final_output)  # a polished email, written by the copywriter

from agents import handoff

copywriter.handoffs = [handoff(analyst)]
analyst.handoffs = [handoff(copywriter)]

import json
from pydantic import BaseModel

class PricingReview(BaseModel):
    risk: str
    summary: str

class RolloutReview(BaseModel):
    risk: str
    blockers: list[str]

# By default, Pydantic output_type results are stringified (repr) when passed back
# as tool output. This extractor ensures the orchestrator receives clean JSON instead.
async def structured_output_extractor(result) -> str:
    final_output = result.final_output
    if isinstance(final_output, BaseModel):
        return json.dumps(final_output.model_dump(mode="json"), sort_keys=True)
    return str(final_output)

# Each reviewer gets its own isolated workspace
pricing_agent = SandboxAgent(
    name="Pricing Reviewer",
    default_manifest=pricing_docs_manifest,
    capabilities=[Shell()],
    output_type=PricingReview,
    ...
)
rollout_agent = SandboxAgent(
    name="Rollout Reviewer",
    default_manifest=rollout_docs_manifest,
    capabilities=[Shell()],
    output_type=RolloutReview,
    ...
)

# Orchestrator calls them like tools, each in its own sandbox
client = DaytonaSandboxClient()
orchestrator = Agent(
    name="Deal Desk Coordinator",
    instructions="Use both review tools, then synthesize a recommendation.",
    tools=[
        pricing_agent.as_tool(
            tool_name="review_pricing",
            tool_description="Review the pricing packet.",
            custom_output_extractor=structured_output_extractor,
            run_config=RunConfig(sandbox=SandboxRunConfig(client=client, options=DaytonaSandboxClientOptions())),
        ),
        rollout_agent.as_tool(
            tool_name="review_rollout",
            tool_description="Review the rollout plan.",
            custom_output_extractor=structured_output_extractor,
            run_config=RunConfig(sandbox=SandboxRunConfig(client=client, options=DaytonaSandboxClientOptions())),
        ),
    ],
)

result = await Runner.run(orchestrator, "Review the Acme Corp renewal deal.")
print(result.final_output)

await client.close()

from agents.sandbox import LocalSnapshotSpec, SandboxRunConfig
from agents.sandbox.capabilities import ApplyPatch, Memory, Shell

agent = SandboxAgent(
    name="Data Analyst",
    model="gpt-5.4",
    instructions="Analyze the workspace and answer questions.",
    default_manifest=manifest,
    capabilities=[
        Shell(),
        ApplyPatch(),
        Memory(),
    ],
)

snapshot = LocalSnapshotSpec(base_path=Path("/tmp/my-agent-snapshots"))

# First run: agent learns user preferences.
# Memory artifacts are written to the workspace when the session closes.
session = await client.create(manifest=manifest, snapshot=snapshot)
async with session:
    run_config = RunConfig(sandbox=SandboxRunConfig(session=session))
    result1 = await Runner.run(agent, "Fix the bug. I prefer minimal patches.", run_config=run_config)

# Second run: resume the workspace so the agent sees the memory files from run 1.
resumed = await client.resume(session.state)
async with resumed:
    run_config = RunConfig(sandbox=SandboxRunConfig(session=resumed))
    result2 = await Runner.run(agent, "Add a test for the fix.", run_config=run_config)

from agents.sandbox.config import MemoryReadConfig, MemoryWriteConfig

# Write-only (no auto-injection of memory into instructions):
Memory(read=None)

# Read-only (no background memory generation):
Memory(write=None)

# Custom write settings:
Memory(write=MemoryWriteConfig(
    batch_size=2,
    extra_prompt="Pay attention to which SQL patterns work best for this dataset.",
))

# Disable live updates (agent reads memory but won't repair stale entries):
Memory(read=MemoryReadConfig(live_update=False))

from agents.sandbox.capabilities.capability import Capability
from agents.tool import Tool, function_tool

class ExposePort(Capability):
    type: str = "expose_port"

    def tools(self) -> list[Tool]:
        session = self.session  # bound automatically by the framework

        @function_tool
        async def get_app_url(port: int) -> str:
            """Get the public URL for a port running in this sandbox."""
            endpoint = await session.resolve_exposed_port(port)
            return endpoint.url_for("http")

        return [get_app_url]