Skip to content

BaseToolkit

AsyncBaseToolkit

Base class for toolkits.

Source code in utu/tools/base.py 
 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
class AsyncBaseToolkit:
    """Base class for toolkits."""

    def __init__(self, config: ToolkitConfig | dict | None = None):
        if not isinstance(config, ToolkitConfig):
            config = config or {}
            config = ToolkitConfig(config=config, name=self.__class__.__name__)

        self.config: ToolkitConfig = config
        self._tools_map: dict[str, Callable] = None

        self.env: _BaseEnv = None
        self.env_mode = self.config.env_mode  # TODO: deprecate it
        if self.env_mode == "e2b":
            self.e2b_sandbox: AsyncSandbox = None

    def setup_env(self, env: "_BaseEnv") -> None:
        """Setup env and workspace."""
        self.env = env
        if self.env_mode == "e2b":  # assert is E2BEnv
            self.e2b_sandbox = env.sandbox
        self.setup_workspace()

    def setup_workspace(self, workspace_root: str = None):
        """Setup workspace. Implemented inside specific toolkits."""
        pass

    async def build(self) -> None:
        """Build/initialize the toolkit. Override in subclasses that need async initialization."""
        pass

    async def cleanup(self) -> None:
        """Cleanup toolkit resources. Override in subclasses that need cleanup."""
        pass

    @property
    def tools_map(self) -> dict[str, Callable]:
        """Lazy loading of tools map.
        - collect tools registered by @register_tool
        """
        if self._tools_map is None:
            self._tools_map = {}
            # Iterate through all methods of the class and register @tool
            for attr_name in dir(self):
                attr = getattr(self, attr_name)
                if callable(attr) and getattr(attr, "_is_tool", False):
                    self._tools_map[attr._tool_name] = attr
        return self._tools_map

    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

    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

    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]

    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]

    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)

tools_map property 

tools_map: dict[str, Callable]

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

setup_env 

setup_env(env: _BaseEnv) -> None

Setup env and workspace.

Source code in utu/tools/base.py 
35
36
37
38
39
40
def setup_env(self, env: "_BaseEnv") -> None:
    """Setup env and workspace."""
    self.env = env
    if self.env_mode == "e2b":  # assert is E2BEnv
        self.e2b_sandbox = env.sandbox
    self.setup_workspace()

setup_workspace 

setup_workspace(workspace_root: str = None)

Setup workspace. Implemented inside specific toolkits.

Source code in utu/tools/base.py 
42
43
44
def setup_workspace(self, workspace_root: str = None):
    """Setup workspace. Implemented inside specific toolkits."""
    pass

build async 

build() -> None

Build/initialize the toolkit. Override in subclasses that need async initialization.

Source code in utu/tools/base.py 
46
47
48
async def build(self) -> None:
    """Build/initialize the toolkit. Override in subclasses that need async initialization."""
    pass

cleanup async 

cleanup() -> None

Cleanup toolkit resources. Override in subclasses that need cleanup.

Source code in utu/tools/base.py 
50
51
52
async def cleanup(self) -> None:
    """Cleanup toolkit resources. Override in subclasses that need cleanup."""
    pass

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.

Source code in utu/tools/base.py 
68
69
70
71
72
73
74
75
76
77
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.

Source code in utu/tools/base.py 
79
80
81
82
83
84
85
86
87
88
89
90
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.

Source code in utu/tools/base.py 
92
93
94
95
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.

Source code in utu/tools/base.py 
 97
 98
 99
100
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.

Source code in utu/tools/base.py 
102
103
104
105
106
107
108
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)