Spaces:
Runtime error
Runtime error
File size: 1,561 Bytes
0e02b97 2bae1d8 7bd7366 ecc9e42 2bae1d8 7bd7366 aee9883 2afcf85 04ad610 7a7b1d3 7bd7366 ecc9e42 aee9883 ecc9e42 aee9883 7a7b1d3 2bae1d8 7bd7366 ecc9e42 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 |
from __future__ import annotations
__all__ = ["execute_terminal", "set_vm"]
import subprocess
from typing import Optional
from .vm import LinuxVM
_VM: Optional[LinuxVM] = None
def set_vm(vm: LinuxVM | None) -> None:
"""Register the VM instance used for command execution."""
global _VM
_VM = vm
def execute_terminal(command: str) -> str:
"""
Execute a shell command in a Linux terminal.
Use this tool to inspect uploaded documents under ``/data`` or run other commands.
The command is executed with network access enabled. Output from both
``stdout`` and ``stderr`` is captured and returned. Commands are killed if
they exceed 2 seconds.
"""
timeout = 2
if not command:
return "No command provided."
if _VM:
try:
return _VM.execute(command, timeout=timeout)
except Exception as exc: # pragma: no cover - unforeseen errors
return f"Failed to execute command in VM: {exc}"
try:
completed = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
)
except subprocess.TimeoutExpired as exc:
return f"Command timed out after {timeout}s: {exc.cmd}"
except Exception as exc: # pragma: no cover - unforeseen errors
return f"Failed to execute command: {exc}"
output = completed.stdout
if completed.stderr:
output = f"{output}\n{completed.stderr}" if output else completed.stderr
return output.strip()
|