Spaces:
Build error
Build error
| """ | |
| Use case for executing tools. | |
| """ | |
| from typing import Dict, Any, Optional, List | |
| from uuid import UUID | |
| import logging | |
| import time | |
| from src.core.entities.tool import Tool, ToolType | |
| from src.core.interfaces.tool_repository import ToolRepository | |
| from src.core.interfaces.tool_executor import ToolExecutor | |
| from src.core.interfaces.logging_service import LoggingService | |
| from src.shared.exceptions import DomainException, ValidationException | |
| class ExecuteToolUseCase: | |
| """ | |
| Use case for executing tools. | |
| This use case handles tool execution, validation, | |
| and result processing. | |
| """ | |
| def __init__( | |
| self, | |
| tool_repository: ToolRepository, | |
| tool_executor: ToolExecutor, | |
| logging_service: LoggingService | |
| ): | |
| self.tool_repository = tool_repository | |
| self.tool_executor = tool_executor | |
| self.logging_service = logging_service | |
| self.logger = logging.getLogger(__name__) | |
| async def execute_tool( | |
| self, | |
| tool_id: UUID, | |
| parameters: Dict[str, Any], | |
| context: Optional[Dict[str, Any]] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Execute a tool with given parameters. | |
| Args: | |
| tool_id: ID of the tool to execute | |
| parameters: Tool execution parameters | |
| context: Optional execution context | |
| Returns: | |
| Dictionary containing the execution result | |
| """ | |
| start_time = time.time() | |
| try: | |
| # Validate input | |
| if not parameters: | |
| raise ValidationException("Tool parameters cannot be empty") | |
| # Find tool | |
| tool = await self.tool_repository.find_by_id(tool_id) | |
| if not tool: | |
| raise DomainException(f"Tool {tool_id} not found") | |
| # Validate tool is available | |
| if not tool.is_available: | |
| raise DomainException(f"Tool {tool_id} is not available") | |
| # Execute tool | |
| result = await self.tool_executor.execute(tool, parameters, context) | |
| execution_time = time.time() - start_time | |
| # Log execution | |
| await self.logging_service.log_info( | |
| "tool_executed", | |
| f"Executed tool {tool_id} successfully", | |
| { | |
| "tool_id": str(tool_id), | |
| "tool_name": tool.name, | |
| "execution_time": execution_time, | |
| "success": result.get("success", False) | |
| } | |
| ) | |
| return { | |
| "success": True, | |
| "tool_id": str(tool_id), | |
| "tool_name": tool.name, | |
| "result": result.get("result"), | |
| "execution_time": execution_time, | |
| "metadata": result.get("metadata", {}) | |
| } | |
| except Exception as e: | |
| execution_time = time.time() - start_time | |
| self.logger.error(f"Tool execution failed: {str(e)}") | |
| await self.logging_service.log_error( | |
| "tool_execution_failed", | |
| str(e), | |
| { | |
| "tool_id": str(tool_id), | |
| "execution_time": execution_time | |
| } | |
| ) | |
| return { | |
| "success": False, | |
| "error": str(e), | |
| "execution_time": execution_time | |
| } | |
| async def execute_tool_by_name( | |
| self, | |
| tool_name: str, | |
| parameters: Dict[str, Any], | |
| context: Optional[Dict[str, Any]] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Execute a tool by name. | |
| Args: | |
| tool_name: Name of the tool to execute | |
| parameters: Tool execution parameters | |
| context: Optional execution context | |
| Returns: | |
| Dictionary containing the execution result | |
| """ | |
| try: | |
| # Find tool by name | |
| tool = await self.tool_repository.find_by_name(tool_name) | |
| if not tool: | |
| raise DomainException(f"Tool '{tool_name}' not found") | |
| # Execute using the found tool | |
| return await self.execute_tool(tool.id, parameters, context) | |
| except Exception as e: | |
| self.logger.error(f"Failed to execute tool by name '{tool_name}': {str(e)}") | |
| return {"success": False, "error": str(e)} | |
| async def validate_tool_parameters( | |
| self, | |
| tool_id: UUID, | |
| parameters: Dict[str, Any] | |
| ) -> Dict[str, Any]: | |
| """ | |
| Validate tool parameters before execution. | |
| Args: | |
| tool_id: ID of the tool to validate | |
| parameters: Parameters to validate | |
| Returns: | |
| Dictionary containing validation result | |
| """ | |
| try: | |
| # Find tool | |
| tool = await self.tool_repository.find_by_id(tool_id) | |
| if not tool: | |
| return {"success": False, "error": f"Tool {tool_id} not found"} | |
| # Validate parameters | |
| validation_result = await self.tool_executor.validate_parameters(tool, parameters) | |
| return { | |
| "success": True, | |
| "valid": validation_result.get("valid", False), | |
| "errors": validation_result.get("errors", []), | |
| "warnings": validation_result.get("warnings", []) | |
| } | |
| except Exception as e: | |
| self.logger.error(f"Parameter validation failed: {str(e)}") | |
| return {"success": False, "error": str(e)} | |
| async def get_tool_info(self, tool_id: UUID) -> Dict[str, Any]: | |
| """ | |
| Get tool information. | |
| Args: | |
| tool_id: ID of the tool to retrieve | |
| Returns: | |
| Dictionary containing tool information | |
| """ | |
| try: | |
| tool = await self.tool_repository.find_by_id(tool_id) | |
| if not tool: | |
| return {"success": False, "error": f"Tool {tool_id} not found"} | |
| return { | |
| "success": True, | |
| "tool": { | |
| "id": str(tool.id), | |
| "name": tool.name, | |
| "description": tool.description, | |
| "tool_type": tool.tool_type.value, | |
| "is_available": tool.is_available, | |
| "parameters_schema": tool.parameters_schema, | |
| "created_at": tool.created_at.isoformat() if tool.created_at else None, | |
| "updated_at": tool.updated_at.isoformat() if tool.updated_at else None | |
| } | |
| } | |
| except Exception as e: | |
| self.logger.error(f"Failed to get tool info {tool_id}: {str(e)}") | |
| return {"success": False, "error": str(e)} | |
| async def list_available_tools(self, tool_type: Optional[ToolType] = None) -> Dict[str, Any]: | |
| """ | |
| List available tools, optionally filtered by type. | |
| Args: | |
| tool_type: Optional tool type filter | |
| Returns: | |
| Dictionary containing the list of tools | |
| """ | |
| try: | |
| if tool_type: | |
| tools = await self.tool_repository.find_by_type(tool_type) | |
| else: | |
| tools = await self.tool_repository.find_available() | |
| tool_list = [] | |
| for tool in tools: | |
| tool_list.append({ | |
| "id": str(tool.id), | |
| "name": tool.name, | |
| "description": tool.description, | |
| "tool_type": tool.tool_type.value, | |
| "is_available": tool.is_available | |
| }) | |
| return { | |
| "success": True, | |
| "tools": tool_list, | |
| "count": len(tool_list) | |
| } | |
| except Exception as e: | |
| self.logger.error(f"Failed to list tools: {str(e)}") | |
| return {"success": False, "error": str(e)} | |
| async def get_tool_statistics(self) -> Dict[str, Any]: | |
| """ | |
| Get tool repository statistics. | |
| Returns: | |
| Dictionary containing tool statistics | |
| """ | |
| try: | |
| stats = await self.tool_repository.get_statistics() | |
| return {"success": True, "statistics": stats} | |
| except Exception as e: | |
| self.logger.error(f"Failed to get tool statistics: {str(e)}") | |
| return {"success": False, "error": str(e)} |