Spaces:
Build error
Build error
File size: 2,610 Bytes
8a682b5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
"""
Tool executor interface defining the contract for tool execution.
"""
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from uuid import UUID
from src.core.entities.tool import Tool
class ToolExecutor(ABC):
"""
Abstract interface for tool execution.
This interface defines the contract that all tool executor
implementations must follow, ensuring consistency across
different execution strategies.
"""
@abstractmethod
async def execute(self, tool: Tool, parameters: Dict[str, Any], context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Execute a tool with given parameters.
Args:
tool: The tool to execute
parameters: Tool execution parameters
context: Optional execution context
Returns:
Dictionary containing the execution result
Raises:
DomainException: If execution fails
"""
pass
@abstractmethod
async def validate_parameters(self, tool: Tool, parameters: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate tool parameters before execution.
Args:
tool: The tool to validate parameters for
parameters: Parameters to validate
Returns:
Dictionary containing validation result
"""
pass
@abstractmethod
async def get_tool_info(self, tool: Tool) -> Dict[str, Any]:
"""
Get tool information and capabilities.
Args:
tool: The tool to query
Returns:
Dictionary containing tool information
"""
pass
@abstractmethod
async def execute_batch(self, tools: list[Tool], parameters_list: list[Dict[str, Any]], context: Optional[Dict[str, Any]] = None) -> list[Dict[str, Any]]:
"""
Execute multiple tools in batch.
Args:
tools: List of tools to execute
parameters_list: List of parameter dictionaries for each tool
context: Optional execution context
Returns:
List of execution results
"""
pass
@abstractmethod
async def cancel_execution(self, execution_id: UUID) -> bool:
"""
Cancel a running tool execution.
Args:
execution_id: The execution to cancel
Returns:
True if cancellation was successful, False otherwise
"""
pass |