Source code for praval.tools

"""
Tool decorator and utilities for Praval Framework.

This module provides the @tool decorator for creating tools that can be
registered and used by agents. Tools are automatically registered in the
global tool registry and can be associated with specific agents.
"""

import inspect
import logging
import importlib
import importlib.util
import glob
import hashlib
from typing import Optional, List, Callable, Union, Any
from functools import wraps

from .core.tool_registry import Tool, ToolMetadata, get_tool_registry, ToolRegistry
from .core.exceptions import ToolError




def tool(
    tool_name: Optional[str] = None,
    owned_by: Optional[str] = None,
    description: Optional[str] = None,
    category: str = "general",
    shared: bool = False,
    version: str = "1.0.0",
    author: str = "",
    tags: Optional[List[str]] = None,
    requires_approval: bool = False,
    risk_level: str = "low",
    approval_reason: str = "",
) -> Callable:
    """
    Decorator to register a function as a tool in the Praval framework.

    The @tool decorator automatically registers functions as tools that can
    be used by agents. Tools can be owned by specific agents, shared across
    all agents, or organized by category.

    Args:
        tool_name: Name of the tool (defaults to function name)
        owned_by: Agent that owns this tool
        description: Description of what the tool does (defaults to docstring)
        category: Category for organizing tools
        shared: Whether this tool is available to all agents
        version: Version of the tool
        author: Author of the tool
        tags: Tags for tool discovery
        requires_approval: Whether this tool requires human approval before execution
        risk_level: Risk category for operator visibility (low/medium/high/critical)
        approval_reason: Optional reason shown to the human approver

    Returns:
        Decorated function with tool metadata attached

    Raises:
        ToolError: If tool registration fails or validation errors occur

    Examples::

        # Basic tool owned by a specific agent
        @tool("add_numbers", owned_by="calculator")
        def add(x: float, y: float) -> float:
            # Add two numbers together.
            return x + y

        # Shared tool available to all agents
        @tool("logger", shared=True, category="utility")
        def log_message(level: str, message: str) -> str:
            # Log a message at the specified level.
            import logging
            logger = logging.getLogger("praval.tools")
            getattr(logger, level.lower())(message)
            return f"Logged: {message}"

        # Tool with metadata
        @tool(
            "validate_email",
            owned_by="data_processor",
            category="validation",
            tags=["email", "validation", "data"],
            version="2.0.0",
            author="Praval Team"
        )
        def validate_email(email: str) -> bool:
            # Validate email address format.
            import re
            pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'
            return bool(re.match(pattern, email))
    """

    def decorator(func: Callable) -> Callable:
        # Auto-generate tool name from function name if not provided
        actual_tool_name = tool_name or func.__name__

        # Auto-generate description from docstring if not provided
        actual_description = description
        if not actual_description and func.__doc__:
            actual_description = func.__doc__.strip()

        # Prepare tags list
        actual_tags = tags or []
        normalized_risk = str(risk_level or "low").strip().lower()
        if normalized_risk not in {"low", "medium", "high", "critical"}:
            normalized_risk = "low"

        # Create tool metadata
        metadata = ToolMetadata(
            tool_name=actual_tool_name,
            owned_by=owned_by,
            description=actual_description or "",
            category=category,
            shared=shared,
            version=version,
            author=author,
            tags=actual_tags,
            requires_approval=requires_approval,
            risk_level=normalized_risk,
            approval_reason=approval_reason or "",
        )

        # Create tool instance
        try:
            tool_instance = Tool(func, metadata)
        except Exception as e:
            raise ToolError(
                f"Failed to create tool '{actual_tool_name}': {str(e)}"
            ) from e

        # Register the tool in the global registry
        registry = get_tool_registry()
        try:
            registry.register_tool(tool_instance)
        except Exception as e:
            raise ToolError(
                f"Failed to register tool '{actual_tool_name}': {str(e)}"
            ) from e

        # Add tool metadata to the function for introspection
        func._praval_tool = tool_instance
        func._praval_tool_name = actual_tool_name
        func._praval_tool_metadata = metadata

        # Add utility methods to the function
        func.get_metadata = lambda: metadata
        func.get_tool_info = lambda: tool_instance.to_dict()
        func.execute_as_tool = tool_instance.execute

        return func

    return decorator





def get_tool_info(tool_func: Callable) -> dict:
    """
    Get information about a @tool decorated function.

    Args:
        tool_func: Function decorated with @tool

    Returns:
        Dictionary with tool metadata

    Raises:
        ValueError: If function is not decorated with @tool
    """
    if not hasattr(tool_func, "_praval_tool"):
        raise ValueError("Function is not decorated with @tool")

    return tool_func._praval_tool.to_dict()





def is_tool(func: Callable) -> bool:
    """
    Check if a function is decorated with @tool.

    Args:
        func: Function to check

    Returns:
        True if function is a tool, False otherwise
    """
    return hasattr(func, "_praval_tool")





def discover_tools(
    module: Optional[str] = None,
    pattern: Optional[str] = None,
    category: Optional[str] = None,
) -> List[Tool]:
    """
    Discover tools based on various criteria.

    Args:
        module: Module name to import (tools register on import)
        pattern: File pattern to search (e.g., ``**/*_tool.py``)
        category: Category to filter by

    Returns:
        List of discovered Tool instances
    """
    registry = get_tool_registry()
    logger = logging.getLogger(__name__)

    if category:
        return registry.get_tools_by_category(category)

    if module:
        try:
            importlib.import_module(module)
        except Exception as e:
            logger.debug("Tool discovery failed to import module '%s': %s", module, e)

    if pattern:
        for file_path in glob.glob(pattern, recursive=True):
            if not file_path.endswith(".py"):
                continue
            try:
                module_id = hashlib.md5(file_path.encode("utf-8")).hexdigest()
                module_name = f"_praval_toolscan_{module_id}"
                spec = importlib.util.spec_from_file_location(module_name, file_path)
                if spec and spec.loader:
                    module_obj = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(module_obj)
            except Exception as e:
                logger.debug("Tool discovery failed to load '%s': %s", file_path, e)

    return registry.list_all_tools()





def list_tools(
    agent_name: Optional[str] = None,
    category: Optional[str] = None,
    shared_only: bool = False,
) -> List[dict]:
    """
    List tools with optional filtering.

    Args:
        agent_name: Filter by agent owner
        category: Filter by category
        shared_only: Only show shared tools

    Returns:
        List of tool information dictionaries
    """
    registry = get_tool_registry()

    if agent_name:
        tools = registry.get_tools_for_agent(agent_name)
    elif category:
        tools = registry.get_tools_by_category(category)
    elif shared_only:
        tools = registry.get_shared_tools()
    else:
        tools = registry.list_all_tools()

    return [tool.to_dict() for tool in tools]





def register_tool_with_agent(tool_name: str, agent_name: str) -> bool:
    """
    Register an existing tool with an agent at runtime.

    Args:
        tool_name: Name of the tool to register
        agent_name: Name of the agent to register with

    Returns:
        True if registration successful, False otherwise
    """
    registry = get_tool_registry()
    return registry.assign_tool_to_agent(tool_name, agent_name)





def unregister_tool_from_agent(tool_name: str, agent_name: str) -> bool:
    """
    Unregister a tool from an agent at runtime.

    Args:
        tool_name: Name of the tool to unregister
        agent_name: Name of the agent to unregister from

    Returns:
        True if unregistration successful, False otherwise
    """
    registry = get_tool_registry()
    return registry.remove_tool_from_agent(tool_name, agent_name)





class ToolCollection:
    """
    A collection of related tools that can be managed as a group.

    Useful for organizing tools by functionality or creating tool suites
    that can be easily assigned to agents.
    """



    def __init__(self, name: str, description: str = ""):
        """
        Initialize a tool collection.

        Args:
            name: Name of the collection
            description: Description of the collection
        """
        self.name = name
        self.description = description
        self.tools: List[str] = []




    def add_tool(self, tool_name: str) -> None:
        """
        Add a tool to the collection.

        Args:
            tool_name: Name of the tool to add

        Raises:
            ToolError: If tool doesn't exist
        """
        registry = get_tool_registry()
        if not registry.get_tool(tool_name):
            raise ToolError(f"Tool '{tool_name}' not found in registry")

        if tool_name not in self.tools:
            self.tools.append(tool_name)




    def remove_tool(self, tool_name: str) -> bool:
        """
        Remove a tool from the collection.

        Args:
            tool_name: Name of the tool to remove

        Returns:
            True if removal successful, False if tool wasn't in collection
        """
        if tool_name in self.tools:
            self.tools.remove(tool_name)
            return True
        return False




    def assign_to_agent(self, agent_name: str) -> int:
        """
        Assign all tools in the collection to an agent.

        Args:
            agent_name: Name of the agent to assign tools to

        Returns:
            Number of tools successfully assigned
        """
        registry = get_tool_registry()
        successful = 0

        for tool_name in self.tools:
            if registry.assign_tool_to_agent(tool_name, agent_name):
                successful += 1

        return successful




    def get_tools(self) -> List[Tool]:
        """
        Get all tools in the collection.

        Returns:
            List of Tool instances in the collection
        """
        registry = get_tool_registry()
        result = []

        for tool_name in self.tools:
            tool = registry.get_tool(tool_name)
            if tool:
                result.append(tool)

        return result