Spaces:
Runtime error
Runtime error
File size: 1,977 Bytes
0e02b97 a8ee91f 7bd7366 ecc9e42 c8f7ad8 2bae1d8 a8ee91f 2bae1d8 7bd7366 aee9883 2afcf85 94490a0 59f95d3 7bd7366 d39d8db ecc9e42 aee9883 7a7b1d3 2bae1d8 d39d8db 2bae1d8 7bd7366 d39d8db ecc9e42 d39d8db c8f7ad8 d39d8db ecc9e42 d39d8db ecc9e42 a8ee91f |
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 57 58 59 60 61 62 63 64 65 |
from __future__ import annotations
__all__ = ["execute_terminal", "execute_terminal_async", "set_vm"]
import subprocess
import os
from typing import Optional
import asyncio
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 Ubuntu terminal.
Use this tool to inspect uploaded documents under ``/data``, fetch web
content with utilities like ``curl`` or ``wget`` and run other commands.
The assistant must call this tool to search the internet whenever unsure
about any detail.
The command is executed with network access enabled. Output from
``stdout`` and ``stderr`` is captured when the command completes.
Execution happens asynchronously so the assistant can continue
responding while the command runs.
"""
if not command:
return "No command provided."
if _VM:
try:
return _VM.execute(command, timeout=None)
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,
env=os.environ.copy(),
timeout=None,
)
output = completed.stdout
if completed.stderr:
output = f"{output}\n{completed.stderr}" if output else completed.stderr
return output.strip()
except Exception as exc: # pragma: no cover - unforeseen errors
return f"Failed to execute command: {exc}"
async def execute_terminal_async(command: str) -> str:
"""Asynchronously execute a shell command."""
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, execute_terminal, command)
|