BashToolkit

--- https://www.anthropic.com/engineering/swe-bench-sonnet --- Run commands in a bash shell

• When invoking this tool, the contents of the "command" parameter does NOT need to be XML-escaped.

• You don't have access to the internet via this tool.

• You do have access to a mirror of common linux and python packages via apt and pip.

• State is persistent across command calls and discussions with the user.

• To inspect a particular line range of a file, e.g. lines 10-25, try 'sed -n 10,25p /path/to/the/file'.

• Please avoid commands that may produce a very large amount of output.

• Please run long lived commands in the background, e.g. 'sleep 10 &' or start a server in the background."

BashToolkit

Bases: AsyncBaseToolkit

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

class BashToolkit(AsyncBaseToolkit):
    def __init__(self, config: ToolkitConfig = None) -> None:
        super().__init__(config)
        self.timeout = self.config.config.get("timeout", 60)

        if self.env_mode == "local":
            from .local_env.bash_pexpect import PexpectBash

            self.bash_runner = PexpectBash(timeout=self.timeout)
            self.setup_workspace(self.config.config.get("workspace_root", "/tmp/"))

    def setup_workspace(self, workspace_root: str):
        if self.env_mode != "local":
            logger.warning(f"BashToolkit should not setup workspace in env_mode {self.env_mode}!")
            return
        workspace_dir = pathlib.Path(workspace_root)
        workspace_dir.mkdir(parents=True, exist_ok=True)
        self.workspace_root = workspace_root
        self.bash_runner.run(f"cd {workspace_root}")

    @register_tool
    async def run_bash(self, command: str) -> str:
        """Execute a bash command in your workspace and return its output.

        Args:
            command: The command to execute
        """
        if self.env_mode == "local":
            return self.bash_runner.run(command)
        else:
            assert self.e2b_sandbox is not None, "E2B sandbox is not set up!"
            from e2b.sandbox.commands.command_handle import CommandExitException

            try:
                result = await self.e2b_sandbox.commands.run(command, timeout=self.timeout)
                return E2BUtils.command_result_to_str(result)
            except CommandExitException as e:
                return E2BUtils.command_exit_exception_to_str(e)

tools_map property

tools_map: dict[str, Callable]

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

run_bash async

run_bash(command: str) -> str

Execute a bash command in your workspace and return its output.

参数：

名称	类型	描述	默认
command	str	The command to execute	必需

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

@register_tool
async def run_bash(self, command: str) -> str:
    """Execute a bash command in your workspace and return its output.

    Args:
        command: The command to execute
    """
    if self.env_mode == "local":
        return self.bash_runner.run(command)
    else:
        assert self.e2b_sandbox is not None, "E2B sandbox is not set up!"
        from e2b.sandbox.commands.command_handle import CommandExitException

        try:
            result = await self.e2b_sandbox.commands.run(command, timeout=self.timeout)
            return E2BUtils.command_result_to_str(result)
        except CommandExitException as e:
            return E2BUtils.command_exit_exception_to_str(e)

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)