45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
"""Python code execution tool (sandboxed via subprocess)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from . import tool
|
|
|
|
|
|
@tool("run_python", "Execute Python code and return the output", category="code")
|
|
def run_python(code: str, timeout: int = 30) -> str:
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, encoding="utf-8") as f:
|
|
f.write(code)
|
|
f.flush()
|
|
tmp_path = f.name
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
[sys.executable, tmp_path],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=min(timeout, 60),
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
output = ""
|
|
if result.stdout:
|
|
output += result.stdout
|
|
if result.stderr:
|
|
output += f"\n[stderr]\n{result.stderr}"
|
|
if result.returncode != 0:
|
|
output += f"\n[exit code: {result.returncode}]"
|
|
if len(output) > 10000:
|
|
output = output[:10000] + "\n... (truncated)"
|
|
return output.strip() or "(no output)"
|
|
except subprocess.TimeoutExpired:
|
|
return f"Execution timed out after {timeout}s"
|
|
except Exception as e:
|
|
return f"Execution error: {e}"
|
|
finally:
|
|
Path(tmp_path).unlink(missing_ok=True)
|