Build an MCP Server That Gives Any Agent a Sandbox

"""
MCP Server: Sandbox Platform

Exposes sandbox operations (create, execute, browse, list, stop, save_file)
to any MCP-compatible agent: Claude Code, Cursor, Gemini CLI, Codex CLI,
JetBrains, and more.

Usage:
    python sandbox_mcp_server.py

Configure in your agent's MCP settings to use this as a stdio server.
"""

import os
import json
import asyncio
from typing import Any

import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
    Resource,
    ResourceTemplate,
    Prompt,
    PromptMessage,
    PromptArgument,
)

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

SANDBOX_API_URL = os.environ.get("SANDBOX_API_URL", "https://sandbox.dev.ab0t.com")
SANDBOX_API_KEY = os.environ.get("SANDBOX_API_KEY", "")

if not SANDBOX_API_KEY:
    raise ValueError("SANDBOX_API_KEY environment variable is required")


def _headers() -> dict[str, str]:
    return {
        "Authorization": f"Bearer {SANDBOX_API_KEY}",
        "Content-Type": "application/json",
    }


async def _api(method: str, path: str, body: dict | None = None) -> dict:
    """Make an API call to the Sandbox Platform."""
    async with httpx.AsyncClient(timeout=120) as client:
        resp = await client.request(
            method,
            f"{SANDBOX_API_URL}{path}",
            headers=_headers(),
            json=body,
        )
        resp.raise_for_status()
        return resp.json()


# ---------------------------------------------------------------------------
# Server
# ---------------------------------------------------------------------------

server = Server("sandbox-platform")


# ---- Tools ----------------------------------------------------------------

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="sandbox_create",
            description="Create a new sandbox. Returns sandbox_id and connection info. "
                        "Use instance_type ab0t.micro for light tasks, ab0t.medium for coding, "
                        "ab0t.large for heavy builds.",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "Human-readable sandbox name"},
                    "instance_type": {
                        "type": "string",
                        "default": "ab0t.medium",
                        "description": "EC2 instance type",
                    },
                    "auto_stop_minutes": {
                        "type": "integer",
                        "default": 60,
                        "description": "Auto-stop after N minutes of inactivity (0 = never)",
                    },
                },
                "required": ["name"],
            },
        ),
        Tool(
            name="sandbox_execute",
            description="Execute a shell command in a running sandbox. "
                        "Returns stdout, stderr, and exit code. "
                        "Use for running code, installing packages, git operations.",
            inputSchema={
                "type": "object",
                "properties": {
                    "sandbox_id": {"type": "string", "description": "Sandbox ID"},
                    "command": {"type": "string", "description": "Shell command to execute"},
                    "timeout": {
                        "type": "integer",
                        "default": 300,
                        "description": "Timeout in seconds",
                    },
                },
                "required": ["sandbox_id", "command"],
            },
        ),
        Tool(
            name="sandbox_browse",
            description="Open a URL in a headless browser running in the sandbox. "
                        "Returns page title, text content, and a screenshot (base64). "
                        "Use for web research, scraping, and testing web apps.",
            inputSchema={
                "type": "object",
                "properties": {
                    "sandbox_id": {"type": "string", "description": "Sandbox ID"},
                    "url": {"type": "string", "description": "URL to navigate to"},
                    "wait_seconds": {
                        "type": "integer",
                        "default": 5,
                        "description": "Wait N seconds after page load for JS rendering",
                    },
                },
                "required": ["sandbox_id", "url"],
            },
        ),
        Tool(
            name="sandbox_list",
            description="List all sandboxes. Returns sandbox_id, name, status, "
                        "instance_type, and uptime for each.",
            inputSchema={
                "type": "object",
                "properties": {},
            },
        ),
        Tool(
            name="sandbox_stop",
            description="Stop a running sandbox. Stops billing. "
                        "The sandbox can be restarted later with its state preserved.",
            inputSchema={
                "type": "object",
                "properties": {
                    "sandbox_id": {"type": "string", "description": "Sandbox ID"},
                },
                "required": ["sandbox_id"],
            },
        ),
        Tool(
            name="sandbox_save_file",
            description="Write a file to the sandbox filesystem. "
                        "Creates parent directories automatically. "
                        "Use for saving code, configs, scripts.",
            inputSchema={
                "type": "object",
                "properties": {
                    "sandbox_id": {"type": "string", "description": "Sandbox ID"},
                    "path": {"type": "string", "description": "Absolute file path in the sandbox"},
                    "content": {"type": "string", "description": "File content"},
                },
                "required": ["sandbox_id", "path", "content"],
            },
        ),
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """Route tool calls to the Sandbox Platform API."""

    if name == "sandbox_create":
        result = await _api("POST", "/api/sandboxes", {
            "name": arguments["name"],
            "instance_type": arguments.get("instance_type", "ab0t.medium"),
            "auto_stop_minutes": arguments.get("auto_stop_minutes", 60),
        })

    elif name == "sandbox_execute":
        result = await _api("POST", f"/api/sandboxes/{arguments['sandbox_id']}/execute", {
            "command": arguments["command"],
            "timeout": arguments.get("timeout", 300),
        })

    elif name == "sandbox_browse":
        result = await _api("POST", f"/api/sandboxes/{arguments['sandbox_id']}/browse", {
            "url": arguments["url"],
            "wait_seconds": arguments.get("wait_seconds", 5),
        })

    elif name == "sandbox_list":
        result = await _api("GET", "/api/sandboxes")

    elif name == "sandbox_stop":
        result = await _api("POST", f"/api/sandboxes/{arguments['sandbox_id']}/stop")

    elif name == "sandbox_save_file":
        result = await _api("POST", f"/api/sandboxes/{arguments['sandbox_id']}/files", {
            "path": arguments["path"],
            "content": arguments["content"],
        })

    else:
        return [TextContent(type="text", text=f"Unknown tool: {name}")]

    return [TextContent(type="text", text=json.dumps(result, indent=2))]


# ---- Resources ------------------------------------------------------------

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="sandbox://list",
            name="Running Sandboxes",
            description="List of all active sandboxes with status and uptime",
            mimeType="application/json",
        ),
    ]


@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "sandbox://list":
        result = await _api("GET", "/api/sandboxes")
        return json.dumps(result, indent=2)
    raise ValueError(f"Unknown resource: {uri}")


# ---- Prompts --------------------------------------------------------------

@server.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="coding-workflow",
            description="Set up a sandbox, clone a repo, and start a coding session",
            arguments=[
                PromptArgument(
                    name="repo_url",
                    description="Git repository URL to clone",
                    required=True,
                ),
                PromptArgument(
                    name="task",
                    description="What to work on",
                    required=True,
                ),
            ],
        ),
    ]


@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
    if name == "coding-workflow":
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=(
                        f"1. Create a sandbox named 'coding-session'\n"
                        f"2. Execute: git clone {arguments['repo_url']} /workspace/repo\n"
                        f"3. Execute: cd /workspace/repo && ls -la\n"
                        f"4. Now work on: {arguments['task']}\n"
                        f"5. Run tests after each change\n"
                        f"6. When done, commit and stop the sandbox"
                    ),
                ),
            )
        ]
    raise ValueError(f"Unknown prompt: {name}")


# ---- Entrypoint -----------------------------------------------------------

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)


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

{
  "mcpServers": {
    "sandbox": {
      "command": "python",
      "args": ["/absolute/path/to/sandbox_mcp_server.py"],
      "env": {
        "SANDBOX_API_KEY": "ab0t_sk_live_YOUR_KEY_HERE",
        "SANDBOX_API_URL": "https://sandbox.dev.ab0t.com"
      }
    }
  }
}

> Create a sandbox called "experiment", clone https://github.com/fastapi/fastapi,
> run the test suite, and tell me what percentage passes.

{
  "mcpServers": {
    "sandbox": {
      "command": "python",
      "args": ["/absolute/path/to/sandbox_mcp_server.py"],
      "env": {
        "SANDBOX_API_KEY": "ab0t_sk_live_YOUR_KEY_HERE",
        "SANDBOX_API_URL": "https://sandbox.dev.ab0t.com"
      }
    }
  }
}

{
  "mcpServers": {
    "sandbox": {
      "command": "python",
      "args": ["/absolute/path/to/sandbox_mcp_server.py"],
      "env": {
        "SANDBOX_API_KEY": "ab0t_sk_live_YOUR_KEY_HERE",
        "SANDBOX_API_URL": "https://sandbox.dev.ab0t.com"
      }
    }
  }
}

{
  "mcp_servers": {
    "sandbox": {
      "command": "python",
      "args": ["/absolute/path/to/sandbox_mcp_server.py"],
      "env": {
        "SANDBOX_API_KEY": "ab0t_sk_live_YOUR_KEY_HERE",
        "SANDBOX_API_URL": "https://sandbox.dev.ab0t.com"
      }
    }
  }
}

{
  "mcpServers": {
    "sandbox": {
      "command": "python",
      "args": ["/absolute/path/to/sandbox_mcp_server.py"],
      "env": {
        "SANDBOX_API_KEY": "ab0t_sk_live_YOUR_KEY_HERE",
        "SANDBOX_API_URL": "https://sandbox.dev.ab0t.com"
      }
    }
  }
}

npx @modelcontextprotocol/inspector

# Set the API key in your environment first
export SANDBOX_API_KEY="ab0t_sk_live_YOUR_KEY_HERE"
export SANDBOX_API_URL="https://sandbox.dev.ab0t.com"

# Run the inspector, pointing it at your server
npx @modelcontextprotocol/inspector python sandbox_mcp_server.py

@server.list_resource_templates()
async def list_resource_templates() -> list[ResourceTemplate]:
    return [
        ResourceTemplate(
            uriTemplate="sandbox://{sandbox_id}/status",
            name="Sandbox Status",
            description="Detailed status of a specific sandbox including "
                        "uptime, CPU/memory usage, and running processes",
            mimeType="application/json",
        ),
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "sandbox://list":
        result = await _api("GET", "/api/sandboxes")
        return json.dumps(result, indent=2)

    # Match sandbox://{id}/status
    if uri.startswith("sandbox://") and uri.endswith("/status"):
        sandbox_id = uri.replace("sandbox://", "").replace("/status", "")
        result = await _api("GET", f"/api/sandboxes/{sandbox_id}")
        return json.dumps(result, indent=2)

    raise ValueError(f"Unknown resource: {uri}")

Prompt(
    name="web-research",
    description="Create a sandbox with a browser and research a topic. "
                "Browses multiple URLs and compiles findings.",
    arguments=[
        PromptArgument(
            name="topic",
            description="Topic to research",
            required=True,
        ),
        PromptArgument(
            name="urls",
            description="Comma-separated URLs to start from (optional)",
            required=False,
        ),
    ],
),
Prompt(
    name="test-matrix",
    description="Create N sandboxes and run a test suite across different configurations",
    arguments=[
        PromptArgument(
            name="repo_url",
            description="Git repository URL",
            required=True,
        ),
        PromptArgument(
            name="configurations",
            description="Comma-separated configs to test (e.g., python3.10,python3.11,python3.12)",
            required=True,
        ),
    ],
),

elif name == "web-research":
    urls_hint = ""
    if arguments.get("urls"):
        urls_hint = f"\nStart by browsing these URLs: {arguments['urls']}"
    return [
        PromptMessage(
            role="user",
            content=TextContent(
                type="text",
                text=(
                    f"1. Create a sandbox named 'research'\n"
                    f"2. Research this topic: {arguments['topic']}\n"
                    f"{urls_hint}\n"
                    f"3. Browse at least 5 relevant URLs using sandbox_browse\n"
                    f"4. Compile findings into a structured markdown report\n"
                    f"5. Save the report to /workspace/research-report.md\n"
                    f"6. Stop the sandbox when done"
                ),
            ),
        )
    ]

elif name == "test-matrix":
    configs = [c.strip() for c in arguments["configurations"].split(",")]
    steps = f"1. Create {len(configs)} sandboxes, one for each configuration\n"
    for i, cfg in enumerate(configs, 1):
        steps += f"   - Sandbox {i}: '{cfg}'\n"
    steps += (
        f"2. In each sandbox, clone {arguments['repo_url']}\n"
        f"3. Set up the specific configuration (install deps, set env vars)\n"
        f"4. Run the test suite in each sandbox\n"
        f"5. Collect and compare results across all configurations\n"
        f"6. Stop all sandboxes when done"
    )
    return [
        PromptMessage(
            role="user",
            content=TextContent(type="text", text=steps),
        )
    ]

# You tell Claude Code:
> Create a sandbox and refactor the auth module to use async/await.
> Clone git@github.com:myorg/backend.git, make the changes,
> run the tests, and commit.

# Claude Code calls:
# 1. sandbox_create(name="auth-refactor", instance_type="ab0t.medium")
# 2. sandbox_execute(sandbox_id="...", command="git clone ... && cd backend")
# 3. sandbox_execute(sandbox_id="...", command="pip install -r requirements.txt")
# 4. sandbox_execute(sandbox_id="...", command="python -m pytest tests/")
# 5. [makes changes, runs tests, iterates until green]
# 6. sandbox_execute(sandbox_id="...", command="git push origin auth-async")
# 7. sandbox_stop(sandbox_id="...")

# You tell Cursor:
> Spin up a sandbox, deploy the frontend build, and browse to
> localhost:3000. Check if the new login form renders correctly.

# Cursor calls:
# 1. sandbox_create(name="frontend-test")
# 2. sandbox_save_file(path="/workspace/dist/index.html", content="...")
# 3. sandbox_execute(command="cd /workspace/dist && python -m http.server 3000 &")
# 4. sandbox_browse(url="http://localhost:3000")
# 5. [analyzes the page content and reports back]
# 6. sandbox_stop(sandbox_id="...")

# You tell Gemini CLI:
> Research the top 5 Python web frameworks in 2026.
> Browse their official sites and GitHub repos.
> Compile a comparison table with stars, downloads, and key features.

# Gemini CLI calls:
# 1. sandbox_create(name="research")
# 2. sandbox_browse(url="https://github.com/topics/python-web-framework")
# 3. sandbox_browse(url="https://fastapi.tiangolo.com")
# 4. sandbox_browse(url="https://www.djangoproject.com")
# 5. [browses more URLs, compiles findings]
# 6. sandbox_save_file(path="/workspace/report.md", content="...")
# 7. sandbox_stop(sandbox_id="...")

# Create a dedicated virtualenv
python -m venv ~/.mcp-sandbox-env
~/.mcp-sandbox-env/bin/pip install mcp httpx

# Point your MCP config at this Python
# In .claude/settings.json:
#   "command": "/Users/you/.mcp-sandbox-env/bin/python"

# In call_tool(), wrap the API call:
try:
    result = await _api("POST", path, body)
except httpx.HTTPStatusError as e:
    error_detail = {
        "error": True,
        "status_code": e.response.status_code,
        "message": e.response.text,
        "hint": "Check if the sandbox is running (sandbox_list) "
                "and the sandbox_id is correct.",
    }
    return [TextContent(type="text", text=json.dumps(error_detail, indent=2))]

import sys

def log(msg: str):
    """Write to stderr for debugging. Does not interfere with MCP protocol."""
    print(f"[sandbox-mcp] {msg}", file=sys.stderr)

# In call_tool():
log(f"Tool call: {name} with args {json.dumps(arguments)}")

Tool(
    name="sandbox_ping",
    description="Test connectivity to the Sandbox Platform API. "
                "Call this first to verify the MCP server is configured correctly.",
    inputSchema={"type": "object", "properties": {}},
),

# In call_tool():
elif name == "sandbox_ping":
    result = await _api("GET", "/health")
    result["mcp_server"] = "connected"

if name == "sandbox_create":
    result = await _api("POST", "/api/sandboxes", {
        "name": arguments["name"],
        "instance_type": arguments.get("instance_type", "ab0t.medium"),
        "auto_stop_minutes": arguments.get("auto_stop_minutes", 60),
    })
    sandbox_id = result["sandbox_id"]

    # Wait for sandbox to be running (up to 90 seconds)
    for _ in range(18):
        await asyncio.sleep(5)
        status = await _api("GET", f"/api/sandboxes/{sandbox_id}")
        if status.get("status") == "running":
            result["status"] = "running"
            result["ready"] = True
            break
    else:
        result["ready"] = False
        result["hint"] = "Sandbox is still booting. Wait a moment before executing."

async def _api_with_retry(method, path, body=None, retries=3):
    for attempt in range(retries):
        try:
            return await _api(method, path, body)
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429 and attempt < retries - 1:
                wait = 2 ** attempt
                log(f"Rate limited. Retrying in {wait}s...")
                await asyncio.sleep(wait)
            else:
                raise