base_agent/tools/code_executor.py

60 lines
2.4 KiB
Python
Raw Normal View History

2026-02-28 08:21:35 +00:00
"""代码执行工具"""
# ════════════════════════════════════════════════════════════════
# tools/code_executor.py — 代码执行工具
# ════════════════════════════════════════════════════════════════
"""
tools/code_executor.py
沙箱代码执行工具在受限环境中运行 Python 代码片段
"""
import io
import contextlib
import time
from tools.base_tool import BaseTool, ToolResult
class CodeExecutorTool(BaseTool):
name = "code_executor"
description = "在沙箱环境中执行 Python 代码片段,返回标准输出"
parameters = {
"code": {
"type": "string",
"description": "要执行的 Python 代码",
},
"timeout": {
"type": "integer",
"description": "超时时间(秒),默认 5",
},
}
# 沙箱:仅允许安全的内置函数
_SAFE_BUILTINS = {
"print": print, "range": range, "len": len,
"int": int, "float": float, "str": str, "list": list,
"dict": dict, "tuple": tuple, "set": set, "bool": bool,
"abs": abs, "max": max, "min": min, "sum": sum,
"enumerate": enumerate, "zip": zip, "map": map,
"sorted": sorted, "reversed": reversed,
}
def execute(self, code: str, timeout: int = 5, **_) -> ToolResult:
stdout_buf = io.StringIO()
start_time = time.perf_counter()
try:
# 重定向 stdout捕获 print 输出
with contextlib.redirect_stdout(stdout_buf):
exec( # noqa: S102
compile(code, "<agent_sandbox>", "exec"),
{"__builtins__": self._SAFE_BUILTINS},
)
elapsed = (time.perf_counter() - start_time) * 1000
output = stdout_buf.getvalue() or "(无输出)"
return ToolResult(
success=True,
output=f"执行成功 ({elapsed:.1f}ms):\n{output}",
metadata={"elapsed_ms": elapsed},
)
except Exception as exc:
return ToolResult(success=False, output=f"执行错误: {type(exc).__name__}: {exc}")