OpenAI Agents SDK + Runloop

export OPENAI_API_KEY=sk-...

export RUNLOOP_API_KEY=...

uv sync --extra runloop

uv run python -c "from agents.extensions.sandbox import RunloopSandboxClient; print('✓ Runloop support installed!')"

"""
basic_runloop_example.py - Simple Runloop sandbox demonstration
"""
import asyncio
import os

from agents import Runner
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.extensions.sandbox import RunloopSandboxClient, RunloopSandboxClientOptions


async def main():
    # Verify API keys are set
    if not os.environ.get("OPENAI_API_KEY"):
        raise SystemExit("OPENAI_API_KEY must be set")
    if not os.environ.get("RUNLOOP_API_KEY"):
        raise SystemExit("RUNLOOP_API_KEY must be set")

    # Create a simple workspace manifest
    manifest = Manifest(
        root="/home/user",  # Default workspace root for Runloop
        entries={
            "README.md": {
                "type": "text",
                "text": (
                    "# Demo Project\n\n"
                    "This workspace demonstrates Runloop sandbox integration.\n"
                ),
            },
            "data.txt": {
                "type": "text",
                "text": "Sample data: 42, 100, 256\n",
            },
        },
    )

    # Configure the sandbox agent
    agent = SandboxAgent(
        name="Runloop Assistant",
        model="gpt-4o",
        instructions=(
            "You are a helpful assistant with access to a cloud sandbox. "
            "Inspect the workspace files and answer questions about them."
        ),
        default_manifest=manifest,
    )

    # Create Runloop sandbox client
    client = RunloopSandboxClient()

    # Configure the run with Runloop backend
    run_config = RunConfig(
        sandbox=SandboxRunConfig(
            client=client,
            options=RunloopSandboxClientOptions(
                # Optional: specify a blueprint for pre-configured environments
                # blueprint_name="runloop/universal-ubuntu-24.04-x86_64-dnd",
                pause_on_exit=False,  # Set to True to suspend instead of delete
            ),
        ),
        workflow_name="Basic Runloop Example",
    )

    try:
        # Run the agent
        result = await Runner.run(
            agent,
            "What files are in the workspace? Summarize their contents.",
            run_config=run_config,
        )

        print("Agent response:")
        print(result.final_output)

    finally:
        # Clean up the client
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())

uv run python basic_runloop_example.py

"""
dockerindocker_postgres_example.py - Advanced Runloop suspend/resume demonstration

This example demonstrates:
- Using custom blueprints (Docker-in-Docker)
- Root user access for package installation
- Database operations in a sandbox
- Suspending and resuming sandbox state
- State persistence across suspend/resume cycles
"""
import asyncio
import os
import json

from agents import Runner
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.extensions.sandbox import (
    RunloopSandboxClient,
    RunloopSandboxClientOptions,
    RunloopUserParameters,
)


async def main():
    # Verify API keys
    if not os.environ.get("OPENAI_API_KEY"):
        raise SystemExit("OPENAI_API_KEY must be set")
    if not os.environ.get("RUNLOOP_API_KEY"):
        raise SystemExit("RUNLOOP_API_KEY must be set")

    print("=== Phase 1: Creating sandbox and installing PostgreSQL ===\n")

    # Create a manifest with a workspace marker
    manifest = Manifest(
        root="/root",  # Root user uses /root as home
        entries={
            "workspace_marker.txt": {
                "type": "text",
                "text": "This file was created in the initial workspace setup.\n",
            },
        },
    )

    # Configure agent for database setup
    setup_agent = SandboxAgent(
        name="Database Setup Agent",
        model="gpt-4o",
        instructions=(
            "You are a database administrator with root access. "
            "Install PostgreSQL, create a database, create a table, and insert sample data. "
            "Verify the installation and data insertion by querying the database."
        ),
        default_manifest=manifest,
    )

    # Create Runloop client
    client = RunloopSandboxClient()

    # Configure sandbox with Docker-in-Docker blueprint and root access
    run_config = RunConfig(
        sandbox=SandboxRunConfig(
            client=client,
            options=RunloopSandboxClientOptions(
                # Use Docker-in-Docker capable blueprint
                blueprint_name="runloop/universal-ubuntu-24.04-x86_64-dnd",
                # Enable suspend on exit instead of deletion
                pause_on_exit=True,
                # Launch as root user for package installation
                user_parameters=RunloopUserParameters(username="root", uid=0),
                # Optional: give the sandbox a friendly name
                name="postgres-demo-sandbox",
            ),
        ),
        workflow_name="PostgreSQL Suspend/Resume Demo",
    )

    try:
        # Phase 1: Setup database
        print("Installing PostgreSQL and setting up database...")
        setup_result = await Runner.run(
            setup_agent,
            (
                "Install PostgreSQL using apt. Then as the postgres user, "
                "create a database called 'demo_db', create a table called 'users' "
                "with columns 'id' (serial) and 'name' (text), and insert three rows: "
                "Alice, Bob, and Charlie. Verify by running a SELECT query."
            ),
            run_config=run_config,
        )

        print("\nSetup Result:")
        print(setup_result.final_output)
        print("\n" + "="*60 + "\n")

        # Phase 2: Suspend and serialize
        print("=== Phase 2: Suspending sandbox ===\n")

        # The sandbox is automatically suspended because pause_on_exit=True
        # Get the session state for serialization
        session_state = await client.get_session_state()

        if session_state:
            # Serialize the session state
            state_json = session_state.model_dump_json()
            print(f"Session state serialized ({len(state_json)} bytes)")

            # Save to file (in real use, save to database or storage)
            with open("/tmp/runloop_session_state.json", "w") as f:
                f.write(state_json)
            print("State saved to /tmp/runloop_session_state.json")

        print("\nSandbox is now suspended (not deleted)")
        print("Installed packages, files, and database contents are preserved")
        print("\n" + "="*60 + "\n")

        # Close the client (sandbox remains suspended in the cloud)
        await client.close()

        # Phase 3: Resume from saved state
        print("=== Phase 3: Resuming from suspended state ===\n")

        # In a real application, this could happen hours or days later
        # Load the saved session state
        with open("/tmp/runloop_session_state.json", "r") as f:
            loaded_state_json = f.read()

        print("Loaded session state from file")

        # Deserialize the state
        from agents.extensions.sandbox import RunloopSandboxSessionState
        loaded_state = RunloopSandboxSessionState.model_validate_json(loaded_state_json)

        # Create a new client and resume the session
        resume_client = RunloopSandboxClient()

        # Configure agent for verification
        verify_agent = SandboxAgent(
            name="Database Verification Agent",
            model="gpt-4o",
            instructions=(
                "You are verifying database persistence after resume. "
                "Check if PostgreSQL is still installed, the workspace marker file exists, "
                "and the database table contains the original data."
            ),
            default_manifest=manifest,
        )

        # Resume with the loaded state
        resume_config = RunConfig(
            sandbox=SandboxRunConfig(
                client=resume_client,
                # Pass the loaded state to resume the sandbox
                state=loaded_state,
            ),
            workflow_name="PostgreSQL Resume Verification",
        )

        # Phase 4: Verify persistence
        print("Verifying that state persisted across suspend/resume...")
        verify_result = await Runner.run(
            verify_agent,
            (
                "Verify the following: 1) PostgreSQL is installed (check version), "
                "2) The workspace_marker.txt file exists in /root, "
                "3) Query the demo_db database and list all users from the users table. "
                "Report all findings."
            ),
            run_config=resume_config,
        )

        print("\nVerification Result:")
        print(verify_result.final_output)
        print("\n" + "="*60 + "\n")

        print("✓ Success! The sandbox resumed with all state intact:")
        print("  • Installed packages (PostgreSQL) persisted")
        print("  • Workspace files persisted")
        print("  • Database contents persisted")

    finally:
        # Clean up
        await resume_client.close()


if __name__ == "__main__":
    asyncio.run(main())

uv run python dockerindocker_postgres_example.py

from agents.extensions.sandbox import RunloopSandboxClientOptions

options = RunloopSandboxClientOptions(
    blueprint_name="runloop/universal-ubuntu-24.04-x86_64-dnd",
    pause_on_exit=True,
    user_parameters=RunloopUserParameters(username="root", uid=0),
    name="my-sandbox",
    timeouts=RunloopTimeouts(create_s=600),
    env_vars={"DATABASE_URL": "postgresql://localhost/mydb"},
)

from agents.extensions.sandbox import RunloopUserParameters

# Standard user (default behavior)
user_params = RunloopUserParameters(username="user", uid=1000)

# Root user for package installation
root_params = RunloopUserParameters(username="root", uid=0)

from agents.extensions.sandbox import RunloopTimeouts

timeouts = RunloopTimeouts(
    create_s=600,           # 10 minutes for sandbox creation
    exec_timeout_unbounded_s=7200,  # 2 hours for long-running commands
    suspend_s=180,          # 3 minutes for suspend
    resume_s=600,           # 10 minutes for resume
)

from agents.extensions.sandbox import (
    RunloopSandboxClient,
    RunloopSandboxClientOptions,
    RunloopSandboxSessionState,
)

# Initial run with suspend enabled
client = RunloopSandboxClient()
options = RunloopSandboxClientOptions(pause_on_exit=True)

# ... run agent tasks ...

# Get and serialize state
session_state = await client.get_session_state()
state_json = session_state.model_dump_json()

# Save state_json to database/file/storage
save_to_storage(state_json)

await client.close()  # Sandbox is suspended, not deleted

# --- Later, in a new session ---

# Load and deserialize state
state_json = load_from_storage()
loaded_state = RunloopSandboxSessionState.model_validate_json(state_json)

# Resume with the loaded state
resume_client = RunloopSandboxClient()
resume_config = RunConfig(
    sandbox=SandboxRunConfig(
        client=resume_client,
        state=loaded_state,  # Resume from here
    )
)

# Sandbox resumes with all state intact

from agents.extensions.sandbox import RunloopUserParameters

user_parameters = RunloopUserParameters(username="root", uid=0)

options = RunloopSandboxClientOptions(
    user_parameters=user_parameters,
    # ...
)

from agents.sandbox import Manifest

manifest = Manifest(
    root="/home/user",  # or "/root" for root user
    entries={
        "config.json": {
            "type": "text",
            "text": '{"setting": "value"}\n',
        },
        "scripts/setup.sh": {
            "type": "text",
            "text": "#!/bin/bash\necho 'Setting up...'\n",
        },
        "data/": {
            "type": "directory",
        },
    },
)

# Agent instruction:
"Read the contents of /home/user/config.json and summarize it."

content = await session.read_file("/home/user/output.txt")

# Agent instruction:
"Create a file at /home/user/results.txt with the analysis results."

await session.write_file(
    "/home/user/output.txt",
    "Analysis complete\n",
)

archive_bytes = await session.read_workspace_archive()

# Save locally
with open("workspace.tar", "wb") as f:
    f.write(archive_bytes)

with open("workspace.tar", "rb") as f:
    archive_bytes = f.read()

await session.write_workspace_archive(archive_bytes)

uv sync --extra runloop

uv run python -c "from agents.extensions.sandbox import RunloopSandboxClient; print('OK')"