|
import sys
|
|
import io
|
|
import contextlib
|
|
from typing import List
|
|
|
|
from pyr.tools.base import BaseTool, ToolParameter
|
|
|
|
|
|
class PythonExecuteTool(BaseTool):
|
|
@property
|
|
def name(self) -> str:
|
|
return "python_execute"
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return "Execute Python code and return the output"
|
|
|
|
@property
|
|
def parameters(self) -> List[ToolParameter]:
|
|
return [
|
|
ToolParameter(
|
|
name="source_code",
|
|
type="string",
|
|
description="Python source code to execute"
|
|
)
|
|
]
|
|
|
|
async def execute(self, source_code: str) -> str:
|
|
try:
|
|
output_buffer = io.StringIO()
|
|
error_buffer = io.StringIO()
|
|
|
|
namespace = {
|
|
'__name__': '__main__',
|
|
'__builtins__': __builtins__,
|
|
}
|
|
|
|
with contextlib.redirect_stdout(output_buffer), \
|
|
contextlib.redirect_stderr(error_buffer):
|
|
|
|
try:
|
|
exec(source_code, namespace)
|
|
except Exception as e:
|
|
error_buffer.write(f"Execution error: {e}\\n")
|
|
|
|
stdout_content = output_buffer.getvalue()
|
|
stderr_content = error_buffer.getvalue()
|
|
|
|
result_parts = []
|
|
|
|
if stdout_content.strip():
|
|
result_parts.append(f"Output:\\n{stdout_content}")
|
|
|
|
if stderr_content.strip():
|
|
result_parts.append(f"Errors:\\n{stderr_content}")
|
|
|
|
if not result_parts:
|
|
result_parts.append("Code executed successfully (no output)")
|
|
|
|
return "\\n".join(result_parts)
|
|
|
|
except Exception as e:
|
|
return f"Error executing Python code: {e}"
|