AI Deploy

pip install e2b==2.21.1 e2b-code-interpreter==2.4.1 python-dotenv

E2B_API_KEY=<key>
E2B_DOMAIN=<domain>
E2B_API_URL=http://agent-sandbox.<domain>/e2b/v1

1. generate project  →  2. build  →  3. push to git
  →  4. create/reuse sandbox  →  5. get the code onto it
  →  6. install deps  →  7. start  →  8. snapshot
  →  9. save deploy state  →  10. push to git

import json
import os
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
from dotenv import load_dotenv
from e2b import CommandExitException, SandboxNotFoundException
from e2b_code_interpreter import Sandbox

load_dotenv()
idle_timeout = 600  # seconds
WORKDIR = "/workspace"    # sandbox inner persistent mount; always address it explicitly

# Absolute path — a relative one breaks "reuse the existing sandbox"
# silently if this script ever runs from a different cwd.
PROJECT_DIR = Path("/path/to/the/project").resolve()  # the project you just wrote
STATE_FILE = PROJECT_DIR / ".ai-deploy-state.json"  # sandbox_id + start_cmd per service


def public_base_url():
    # Force https regardless of E2B_API_URL's own scheme.
    parts = urlsplit(os.environ["E2B_API_URL"])
    return urlunsplit(("https", parts.netloc, "", "", ""))


def load_state():
    return json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else {}


def save_state(state):
    STATE_FILE.write_text(json.dumps(state, indent=2))


def get_or_create_sandbox(state, key, template, idle_timeout, envs=None):
    """Reuse the recorded sandbox if it still exists, else create one.
    Returns (sandbox, service_state_entry, created)."""
    entry = state.setdefault(key, {})
    sandbox_id = entry.get("sandbox_id")
    if sandbox_id:
        try:
            return Sandbox.connect(sandbox_id), entry, False  # reused
        except SandboxNotFoundException:
            pass  # deleted/expired — fall through and create a new one

    sbx = Sandbox.create(
        template=template,
        timeout=-1,  # no hard lifetime; idle timeout owns reclamation
        metadata={"idleTimeout": str(idle_timeout)},
        lifecycle={"on_timeout": "pause", "auto_resume": True},
        envs=envs
    )
    entry["sandbox_id"] = sbx.sandbox_id
    return sbx, entry, True  # freshly created

def upload_dir(sbx, local_dir, workdir, skip=()):
    for local in sorted(local_dir.rglob("*")):
        if not local.is_file():
            continue
        rel = local.relative_to(local_dir).as_posix()
        if rel in skip:
            continue
        with open(local, "rb") as file:
            sbx.files.write(f"{workdir}/{rel}", file)

def sync_via_git(sbx, workdir, repo_url, branch=None, username=None, password=None):
    if sbx.files.exists(f"{workdir}/.git"):
        sbx.git.pull(workdir, branch=branch, username=username, password=password)
    else:
        sbx.git.clone(repo_url, path=workdir, branch=branch,
                       username=username, password=password)

def install_deps(sbx, workdir, cmd, timeout=300):
    sbx.commands.run(cmd, cwd=workdir, timeout=timeout)

# install_deps(sbx, WORKDIR, "pip install -r requirements.txt")
# install_deps(sbx, WORKDIR, "npm install --omit=dev")