PythonExecutorToolkit

PythonExecutorToolkit

Bases: AsyncBaseToolkit

A tool for executing Python code in a sandboxed environment.

源代码位于： 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)

        if self.env_mode == "local":
            self.setup_workspace(self.config.config.get("workspace_root", None))
        elif self.env_mode == "e2b":
            pass
        else:
            raise ValueError(f"Unsupported env_mode {self.env_mode} for PythonExecutorToolkit!")

    def setup_workspace(self, workspace_root: str = None):
        if self.env_mode != "local":
            logger.warning(f"PythonExecutorToolkit should not setup workspace in env_mode {self.env_mode}!")
            return
        if workspace_root is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            unique_id = str(uuid.uuid4())[:8]
            workspace_root = f"/tmp/utu/python_executor/{timestamp}_{unique_id}"
        workspace_dir = pathlib.Path(workspace_root)
        workspace_dir.mkdir(parents=True, exist_ok=True)
        self.workspace_root = str(workspace_root)

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

        Args:
            code (str): The Python code to execute.
            timeout (int): The execution timeout in seconds. Defaults to 30.

        Returns:
            dict: A dictionary containing the execution results.
        """
        if self.env_mode == "local":
            return await execute_python_code_async(code, self.workspace_root, timeout=timeout)
        else:
            assert self.e2b_sandbox is not None, "E2B sandbox is not set up!"
            result = await self.e2b_sandbox.run_code(code, language="python", timeout=timeout)
            return E2BUtils.execution_to_str(result)

tools_map property

tools_map: dict[str, Callable]

Lazy loading of tools map. - collect tools registered by @register_tool

execute_python_code async

execute_python_code(code: str, timeout: int = 30) -> dict

Executes Python code and returns the output.

参数：

名称	类型	描述	默认
code	str	The Python code to execute.	必需
timeout	int	The execution timeout in seconds. Defaults to 30.	30

返回：

名称	类型	描述
dict	dict	A dictionary containing the execution results.

源代码位于： utu/tools/python_executor_toolkit.py

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

    Args:
        code (str): The Python code to execute.
        timeout (int): The execution timeout in seconds. Defaults to 30.

    Returns:
        dict: A dictionary containing the execution results.
    """
    if self.env_mode == "local":
        return await execute_python_code_async(code, self.workspace_root, timeout=timeout)
    else:
        assert self.e2b_sandbox is not None, "E2B sandbox is not set up!"
        result = await self.e2b_sandbox.run_code(code, language="python", timeout=timeout)
        return E2BUtils.execution_to_str(result)

get_tools_map_func

get_tools_map_func() -> dict[str, Callable]

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

源代码位于： utu/tools/base.py

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.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

get_tools_in_agents() -> list[FunctionTool]

Get tools in openai-agents format.

源代码位于： utu/tools/base.py

def get_tools_in_agents(self) -> list[FunctionTool]:
    """Get tools in openai-agents format."""
    tools_map = 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

get_tools_in_openai() -> list[dict]

Get tools in OpenAI format.

源代码位于： utu/tools/base.py

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

get_tools_in_mcp

get_tools_in_mcp() -> list[Tool]

Get tools in MCP format.

源代码位于： utu/tools/base.py

def get_tools_in_mcp(self) -> list[types.Tool]:
    """Get tools in MCP format."""
    tools = 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.

源代码位于： utu/tools/base.py

async def call_tool(self, name: str, arguments: dict) -> str:
    """Call a tool by its name."""
    tools_map = 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)