src/hermes/agents/base/tool_executor.py

"""Tool executor for agent actions."""

from __future__ import annotations

import logging
import time
from typing import Any

from hermes.core.exceptions import ToolError, ToolNotFoundError
from hermes.core.types import ToolCall, ToolResult
from hermes.tools.base.registry import tool_registry

logger = logging.getLogger(__name__)


class ToolExecutor:
    """Executes tool calls with error handling and retries."""

    def __init__(self, allowed_tools: list[str] | None = None) -> None:
        self.allowed_tools = allowed_tools

    async def execute(self, tool_call: ToolCall) -> ToolResult:
        """Execute a tool call and return the result."""
        start_time = time.monotonic()

        try:
            tool = tool_registry.get(tool_call.tool_name)
            if tool is None:
                raise ToolNotFoundError(tool_call.tool_name)

            if self.allowed_tools and tool_call.tool_name not in self.allowed_tools:
                raise ToolError(
                    f"Tool '{tool_call.tool_name}' not in allowed tools",
                    {"allowed": self.allowed_tools},
                )

            output = await tool.execute(**tool_call.arguments)
            execution_time = (time.monotonic() - start_time) * 1000

            from hermes.observability.metrics import metrics as m
            m.record_tool_call(tool_call.tool_name, True, execution_time)

            return ToolResult(
                tool_call_id=tool_call.id,
                tool_name=tool_call.tool_name,
                success=True,
                output=output,
                execution_time_ms=execution_time,
            )

        except (ToolNotFoundError, ToolError) as e:
            execution_time = (time.monotonic() - start_time) * 1000
            logger.error(f"Tool error: {e}")
            from hermes.observability.metrics import metrics as m
            m.record_tool_call(tool_call.tool_name, False, execution_time)
            return ToolResult(
                tool_call_id=tool_call.id,
                tool_name=tool_call.tool_name,
                success=False,
                error=str(e),
                execution_time_ms=execution_time,
            )

        except TimeoutError:
            execution_time = (time.monotonic() - start_time) * 1000
            return ToolResult(
                tool_call_id=tool_call.id,
                tool_name=tool_call.tool_name,
                success=False,
                error=f"Tool '{tool_call.tool_name}' timed out",
                execution_time_ms=execution_time,
            )

        except Exception as e:
            execution_time = (time.monotonic() - start_time) * 1000
            logger.exception(f"Unexpected error executing tool: {e}")
            return ToolResult(
                tool_call_id=tool_call.id,
                tool_name=tool_call.tool_name,
                success=False,
                error=f"Unexpected error: {str(e)}",
                execution_time_ms=execution_time,
            )

    async def execute_raw(self, tool_name: str, arguments: dict[str, Any]) -> ToolResult:
        """Execute a tool by name with arguments."""
        tool_call = ToolCall(tool_name=tool_name, arguments=arguments)
        return await self.execute(tool_call)