metabuilder/workflow/executor/python/tool_calls_handler.py

"""Handle tool calls from LLM responses."""
import json


def handle_tool_calls(resp_msg, tool_map: dict, msgs: dict, args, policies: dict, logger) -> list:
    """Execute tool calls and return tool result messages."""
    if not resp_msg.tool_calls:
        return []

    modifying_tools = set(policies.get("modifying_tools", []))
    tool_results = []

    for tool_call in resp_msg.tool_calls:
        function_name = tool_call.function.name
        call_id = tool_call.id
        payload = json.loads(tool_call.function.arguments)
        logger.trace("Tool call %s payload: %s", function_name, payload)

        handler = tool_map.get(function_name)
        if not handler:
            msg_template = msgs.get(
                "error_tool_not_found",
                "Tool {name} not found or unavailable."
            )
            msg = msg_template.format(name=function_name)
            logger.error(msg)
            tool_results.append({
                "tool_call_id": call_id,
                "role": "tool",
                "name": function_name,
                "content": msg,
            })
            continue

        if not args.yolo:
            confirm = input(
                msgs.get(
                    "confirm_tool_execution",
                    "Do you want to execute {name} with {args}? [y/N]: "
                ).format(name=function_name, args=payload)
            )
            if confirm.lower() != "y":
                skipped_template = msgs.get("info_tool_skipped", "Skipping tool: {name}")
                logger.info(skipped_template.format(name=function_name))
                tool_results.append({
                    "tool_call_id": call_id,
                    "role": "tool",
                    "name": function_name,
                    "content": "Skipped by user.",
                })
                continue

        if args.dry_run and function_name in modifying_tools:
            logger.info(
                msgs.get(
                    "info_dry_run_skipping",
                    "DRY RUN: Skipping state-modifying tool {name}"
                ).format(name=function_name)
            )
            tool_results.append({
                "tool_call_id": call_id,
                "role": "tool",
                "name": function_name,
                "content": "Skipped due to dry-run.",
            })
            continue

        exec_template = msgs.get("info_executing_tool", "Executing tool: {name}")
        logger.info(exec_template.format(name=function_name))
        try:
            result = handler(**payload)
            content = str(result) if result is not None else "Success"
            if hasattr(result, "__iter__") and not isinstance(result, str):
                items = list(result)[:5]
                content = "\n".join([f"- {item}" for item in items])
                logger.info(content)
            elif result is not None:
                logger.info(result)

            tool_results.append({
                "tool_call_id": call_id,
                "role": "tool",
                "name": function_name,
                "content": content,
            })
        except Exception as error:  # pylint: disable=broad-exception-caught
            error_msg = f"Error executing {function_name}: {error}"
            logger.error(error_msg)
            tool_results.append({
                "tool_call_id": call_id,
                "role": "tool",
                "name": function_name,
                "content": error_msg,
            })

    return tool_results