PythonExecutorToolkit

PythonExecutorToolkit

Bases: AsyncBaseToolkit

A tool for executing Python code in a sandboxed environment.

Source code in utu/tools/python_executor_toolkit.py

class PythonExecutorToolkit(AsyncBaseToolkit):
    """
    A tool for executing Python code in a sandboxed environment.
    """

    def __init__(self, config: ToolkitConfig | dict | None = None):
        super().__init__(config)

    async def get_tools_map(self) -> dict[str, callable]:
        return {
            "execute_python_code": self.execute_python_code,
        }

    async def execute_python_code(self, code: str, workdir: str = "./run_workdir", timeout: int = 30) -> dict:
        """
        Executes Python code and returns the output.

        Args:
            code (str): The Python code to execute.
            workdir (str): The working directory for the execution. Defaults to "./run_workdir".
            timeout (int): The execution timeout in seconds. Defaults to 30.

        Returns:
            dict: A dictionary containing the execution results.
        """
        loop = asyncio.get_running_loop()
        try:
            return await asyncio.wait_for(
                loop.run_in_executor(
                    None,  # Use the default thread pool executor
                    _execute_python_code_sync,
                    code,
                    workdir,
                ),
                timeout=timeout,
            )
        except TimeoutError:
            return {
                "success": False,
                "message": f"Code execution timed out ({timeout} seconds)",
                "stdout": "",
                "stderr": "",
                "status": False,
                "output": "",
                "files": [],
                "error": f"Code execution timed out ({timeout} seconds)",
            }

execute_python_code async

execute_python_code(
    code: str,
    workdir: str = "./run_workdir",
    timeout: int = 30,
) -> dict

Executes Python code and returns the output.

Parameters:

Name	Type	Description	Default
code	str	The Python code to execute.	required
workdir	str	The working directory for the execution. Defaults to "./run_workdir".	'./run_workdir'
timeout	int	The execution timeout in seconds. Defaults to 30.	30

Returns:

Name	Type	Description
dict	dict	A dictionary containing the execution results.

Source code in utu/tools/python_executor_toolkit.py

async def execute_python_code(self, code: str, workdir: str = "./run_workdir", timeout: int = 30) -> dict:
    """
    Executes Python code and returns the output.

    Args:
        code (str): The Python code to execute.
        workdir (str): The working directory for the execution. Defaults to "./run_workdir".
        timeout (int): The execution timeout in seconds. Defaults to 30.

    Returns:
        dict: A dictionary containing the execution results.
    """
    loop = asyncio.get_running_loop()
    try:
        return await asyncio.wait_for(
            loop.run_in_executor(
                None,  # Use the default thread pool executor
                _execute_python_code_sync,
                code,
                workdir,
            ),
            timeout=timeout,
        )
    except TimeoutError:
        return {
            "success": False,
            "message": f"Code execution timed out ({timeout} seconds)",
            "stdout": "",
            "stderr": "",
            "status": False,
            "output": "",
            "files": [],
            "error": f"Code execution timed out ({timeout} seconds)",
        }

get_tools_map_func async

get_tools_map_func() -> dict[str, Callable]

Get tools map. It will filter tools by config.activated_tools if it is not None.

Source code in utu/tools/base.py

async def get_tools_map_func(self) -> dict[str, Callable]:
    """Get tools map. It will filter tools by config.activated_tools if it is not None."""
    if self.tools_map is None:
        self.tools_map = await self.get_tools_map()
    if self.config.activated_tools:
        assert all(tool_name in self.tools_map for tool_name in self.config.activated_tools), (
            f"Error config activated tools: {self.config.activated_tools}! available tools: {self.tools_map.keys()}"
        )
        tools_map = {tool_name: self.tools_map[tool_name] for tool_name in self.config.activated_tools}
    else:
        tools_map = self.tools_map
    return tools_map

get_tools_in_agents async

get_tools_in_agents() -> list[FunctionTool]

Get tools in openai-agents format.

Source code in utu/tools/base.py

async def get_tools_in_agents(self) -> list[FunctionTool]:
    """Get tools in openai-agents format."""
    tools_map = await self.get_tools_map_func()
    tools = []
    for _, tool in tools_map.items():
        tools.append(
            function_tool(
                tool,
                strict_mode=False,  # turn off strict mode
            )
        )
    return tools

get_tools_in_openai async

get_tools_in_openai() -> list[dict]

Get tools in OpenAI format.

Source code in utu/tools/base.py

async def get_tools_in_openai(self) -> list[dict]:
    """Get tools in OpenAI format."""
    tools = await self.get_tools_in_agents()
    return [ChatCompletionConverter.tool_to_openai(tool) for tool in tools]

get_tools_in_mcp async

get_tools_in_mcp() -> list[Tool]

Get tools in MCP format.

Source code in utu/tools/base.py

async def get_tools_in_mcp(self) -> list[types.Tool]:
    """Get tools in MCP format."""
    tools = await self.get_tools_in_agents()
    return [MCPConverter.function_tool_to_mcp(tool) for tool in tools]

call_tool async

call_tool(name: str, arguments: dict) -> str

Call a tool by its name.

Source code in utu/tools/base.py

async def call_tool(self, name: str, arguments: dict) -> str:
    """Call a tool by its name."""
    tools_map = await self.get_tools_map_func()
    if name not in tools_map:
        raise ValueError(f"Tool {name} not found")
    tool = tools_map[name]
    return await tool(**arguments)