跳转至

BaseToolkit

AsyncBaseToolkit

Base class for toolkits.

源代码位于： 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
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_mode = self.config.env_mode
        if self.env_mode == "e2b":
            self.e2b_env: E2BEnv = None
            self.e2b_sandbox: AsyncSandbox = None

    def setup_e2b_env(self, env: "E2BEnv") -> None:
        if self.env_mode != "e2b":
            logger.warning(f"Toolkit should not setup e2b sandbox in env_mode {self.env_mode}!")
            return
        self.e2b_env = env
        self.e2b_sandbox = env.sandbox

    @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

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 
56
57
58
59
60
61
62
63
64
65
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 
67
68
69
70
71
72
73
74
75
76
77
78
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 
80
81
82
83
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 
85
86
87
88
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 
90
91
92
93
94
95
96
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)