Skip to content

BaseToolkit

AsyncBaseToolkit

Bases: ABC

Base class for toolkits.

Source code in utu/tools/base.py 
 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
109
110
111
112
113
114
115
116
117
118
119
120
121
class AsyncBaseToolkit(abc.ABC):
    """Base class for toolkits."""

    config: ToolkitConfig
    tools_map: dict[str, Callable] = None

    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 = config
        self._built = False

    async def __aenter__(self):
        await self.build()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.cleanup()

    async def build(self):
        if self._built:
            return
        self._built = True

    async def cleanup(self):
        self._built = False

    @abc.abstractmethod
    async def get_tools_map(self) -> dict[str, Callable]:
        """Abstract method to get tools map.

        Returns:
            dict[str, Callable]: A dictionary of tool names to their corresponding functions.
        """
        pass

    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

    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

    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]

    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]

    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)

    # -------------------------------------------------------------------------------------------------------------
    def get_tools_map_sync(self) -> dict[str, Callable]:
        if self.tools_map is None:
            loop = get_event_loop()
            if not self._built:
                loop.run_until_complete(self.build())
            self.tools_map = loop.run_until_complete(self.get_tools_map_func())
        return self.tools_map

    def get_tools_in_agents_sync(self) -> list[FunctionTool]:
        tools_map = self.get_tools_map_sync()
        tools = []
        for _, tool in tools_map.items():
            tools.append(
                function_tool(
                    tool,
                    strict_mode=False,  # turn off strict mode
                )
            )
        return tools

get_tools_map abstractmethod async 

get_tools_map() -> dict[str, Callable]

Abstract method to get tools map.

Returns:

Type Description
dict[str, Callable]

dict[str, Callable]: A dictionary of tool names to their corresponding functions.
Source code in utu/tools/base.py 
49
50
51
52
53
54
55
56
@abc.abstractmethod
async def get_tools_map(self) -> dict[str, Callable]:
    """Abstract method to get tools map.

    Returns:
        dict[str, Callable]: A dictionary of tool names to their corresponding functions.
    """
    pass

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 
58
59
60
61
62
63
64
65
66
67
68
69
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 
71
72
73
74
75
76
77
78
79
80
81
82
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 
84
85
86
87
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 
89
90
91
92
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 
 94
 95
 96
 97
 98
 99
100
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)