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

Code before:
            if not allowlist or tool.name not in allowlist:
                continue
        _load_tool(tool)

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


def _load_tool(tool: ToolSpec) -> None:
    """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",
                    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
    is_filename = "." in lang or "/" in lang
    if is_filename:

Review comment (suggestion):
I'm not sure this belongs so deep into the tools module, should probably be kept in `chat.py` if possible

Code after:
        if not tool.available:
            continue
        if tool.disabled_by_default:
            if not allowlist or tool.name not in allowlist:
                continue
        if tool.init:
            tool = tool.init()

        _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