Repository: ErikBjare/gptme
File: gptme/tools/__init__.py (Python)

Code before:
    """Loads a tool."""
    if tool in loaded_tools:
        logger.warning(f"Tool '{tool.name}' already loaded")
        return

    # tool init happens in init_tools to check that spec is available
    if tool.functions:
        for func in tool.functions:
            register_function(func)
    loaded_tools.append(tool)


def execute_msg(msg: Message, confirm: ConfirmFunc) -> Generator[Message, None, None]:
    """Uses any tools called in a message and returns the response."""
    assert msg.role == "assistant", "Only assistant messages can be executed"

    for tooluse in ToolUse.iter_from_content(msg.content):
        if tooluse.is_runnable:
            try:
                for tool_response in tooluse.execute(confirm):
                    yield tool_response.replace(call_id=tooluse.call_id)
            except KeyboardInterrupt:
                clear_interruptible()
                yield Message(
                    "system",
                    "User hit Ctrl-c to interrupt the process",
                    call_id=tooluse.call_id,
                )
                break


# Called often when checking streaming output for executable blocks,
# so we cache the result.
@lru_cache
def get_tool_for_langtag(lang: str) -> ToolSpec | None:
    block_type = lang.split(" ")[0]
    for tool in loaded_tools:
        if block_type in tool.block_types:
            return tool
    is_filename = "." in lang or "/" in lang
    if is_filename:
        # NOTE: special case
        return tool_save
    return None


def is_supported_langtag(lang: str) -> bool:
    return bool(get_tool_for_langtag(lang))


def get_tool(tool_name: str) -> ToolSpec | None:

Review comment (suggestion):
I kinda want to make these shorter

```suggestion
                    "Interrupted by user",
```

Code after:
        _loaded_tools.append(tool)

    for tool_name in allowlist or []:
        if not has_tool(tool_name):
            raise ValueError(f"Tool '{tool_name}' not found")


def execute_msg(msg: Message, confirm: ConfirmFunc) -> Generator[Message, None, None]:
    """Uses any tools called in a message and returns the response."""
    assert msg.role == "assistant", "Only assistant messages can be executed"

    for tooluse in ToolUse.iter_from_content(msg.content):
        if tooluse.is_runnable:
            try:
                for tool_response in tooluse.execute(confirm):
                    yield tool_response.replace(call_id=tooluse.call_id)
            except KeyboardInterrupt:
                clear_interruptible()
                yield Message(
                    "system",
                    INTERRUPT_CONTENT,
                    call_id=tooluse.call_id,
                )
                break


# Called often when checking streaming output for executable blocks,
# so we cache the result.
@lru_cache
def get_tool_for_langtag(lang: str) -> ToolSpec | None:
    block_type = lang.split(" ")[0]
    for tool in _loaded_tools:
        if block_type in tool.block_types:
            return tool
    return None


def is_supported_langtag(lang: str) -> bool:
    return bool(get_tool_for_langtag(lang))


def get_available_tools() -> list[ToolSpec]:
    global _available_tools

    if _available_tools is None:
        # We need to load tools first
        config = get_config()

        tool_modules: frozenset[str] = frozenset()
        env_tool_modules = config.get_env("TOOL_MODULES", "gptme.tools")