Function execute_python_script

src/sandbox_vm/tools/code/python.py:146–225

Execute an existing Python script

(
    ctx: Context,
    script_path: str,
    args: Optional[List[str]] = None,
    timeout: int = 30,
    *,
    cwd_path: Optional[str] = None,  # defaults to script's parent
)

async def execute_python_script(
    ctx: Context,
    script_path: str,
    args: Optional[List[str]] = None,
    timeout: int = 30,
    *,
    cwd_path: Optional[str] = None,  # defaults to script's parent
) -> dict:
    """
    Execute an existing Python script
    """
    sid = getattr(ctx, "session_id", None)
    if not sid:
        return _err("session_id_REQUIRED", message="ctx.session_id is missing")

    rt = MANAGER.get(sid)
    env = rt.minimal_env()
    tout = _clamp_timeout(timeout)

    try:
        sp = (rt.root / script_path.lstrip("/")).resolve()
        if not sp.exists():
            return _err(
                "FILE_NOT_FOUND", message=f"Script does not exist: {script_path}"
            )
        if sp.suffix.lower() != ".py":
            return _err(
                "INVALID_ARGUMENT", message=f"File must be a .py script: {sp.name}"
            )

        cwd = (rt.root / cwd_path.lstrip("/")).resolve() if cwd_path else sp.parent

        cmd = ["python3", str(sp)]
        if args:
            cmd.extend([str(a) for a in args])

        result = {
            "stdout": "",
            "stderr": "",
            "return_code": -1,
            "execution_time": 0.0,
            "script_path": script_path,
        }

        start = time.time()
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=str(cwd),
            env=env,
        )

        try:
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=tout)
            result["execution_time"] = time.time() - start
            result["return_code"] = proc.returncode