Spaces:
Sleeping
Sleeping
File size: 4,173 Bytes
19ccffa 8c46507 19ccffa 8c46507 19ccffa 8c46507 19ccffa | 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 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | import json
import subprocess
import tempfile
import os
from google.genai import types
def execute_python_code(code: str, timeout: int = 30) -> dict:
"""Execute Python code in a sandboxed subprocess and return output."""
result = {"stdout": "", "stderr": "", "exit_code": -1, "success": False}
# Security: block dangerous operations
blocked = ["os.system", "subprocess", "eval(", "exec(", "import os", "__import__"]
for b in blocked:
if b in code:
result["stderr"] = f"Blocked: '{b}' is not allowed for security reasons."
return result
try:
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(code)
tmp_path = f.name
proc = subprocess.run(
["python3", tmp_path],
capture_output=True,
text=True,
timeout=timeout,
cwd=tempfile.gettempdir(),
env={**os.environ, "HOME": tempfile.gettempdir()},
)
result["stdout"] = proc.stdout
result["stderr"] = proc.stderr
result["exit_code"] = proc.returncode
result["success"] = proc.returncode == 0
except subprocess.TimeoutExpired:
result["stderr"] = f"Code execution timed out after {timeout}s"
except Exception as e:
result["stderr"] = str(e)
finally:
try:
os.unlink(tmp_path)
except Exception:
pass
return result
def execute_javascript_code(code: str, timeout: int = 10) -> dict:
"""Execute JavaScript code using Node.js."""
result = {"stdout": "", "stderr": "", "exit_code": -1, "success": False}
try:
with tempfile.NamedTemporaryFile(mode="w", suffix=".js", delete=False) as f:
f.write(code)
tmp_path = f.name
proc = subprocess.run(
["node", tmp_path],
capture_output=True,
text=True,
timeout=timeout,
)
result["stdout"] = proc.stdout
result["stderr"] = proc.stderr
result["exit_code"] = proc.returncode
result["success"] = proc.returncode == 0
except FileNotFoundError:
result["stderr"] = "Node.js is not available on this server."
except subprocess.TimeoutExpired:
result["stderr"] = f"JS execution timed out after {timeout}s"
except Exception as e:
result["stderr"] = str(e)
finally:
try:
os.unlink(tmp_path)
except Exception:
pass
return result
# Tool declarations for google-genai function calling
code_tool_declarations = [
types.FunctionDeclaration(
name="execute_code",
description="Execute Python or JavaScript code and return output. Use to calculate, process data, or test algorithms. Never use for system access or network requests.",
parameters=types.Schema(
type="OBJECT",
properties={
"language": types.Schema(
type="STRING",
enum=["python", "javascript"],
description="Programming language",
),
"code": types.Schema(type="STRING", description="The code to execute"),
},
required=["language", "code"],
),
)
]
def handle_code_execution(args: dict) -> str:
"""Handle code execution tool call."""
language = args.get("language", "python")
code = args.get("code", "")
if language == "python":
result = execute_python_code(code)
elif language == "javascript":
result = execute_javascript_code(code)
else:
return f"Unsupported language: {language}"
output = ""
if result["success"]:
if result["stdout"]:
output += result["stdout"]
if result["stderr"]:
output += f"\n[stderr]: {result['stderr']}"
if not output:
output = "Code executed successfully (no output)."
else:
output = f"Error (exit {result['exit_code']}): {result['stderr']}"
if result["stdout"]:
output = f"{result['stdout']}\n{output}"
return output.strip()
|