kiln_ai.tools

1from kiln_ai.tools.base_tool import KilnTool, KilnToolInterface
2from kiln_ai.tools.tool_registry import tool_from_id
3
4__all__ = [
5    "KilnTool",
6    "KilnToolInterface",
7    "tool_from_id",
8]
class KilnTool(kiln_ai.tools.KilnToolInterface):
41class KilnTool(KilnToolInterface):
42    """
43    Base helper class that provides common functionality for tool implementations.
44    Subclasses only need to implement run() and provide tool configuration.
45    """
46
47    def __init__(
48        self,
49        tool_id: KilnBuiltInToolId,
50        name: str,
51        description: str,
52        parameters_schema: Dict[str, Any],
53    ):
54        self._id = tool_id
55        self._name = name
56        self._description = description
57        validate_schema_dict(parameters_schema)
58        self._parameters_schema = parameters_schema
59
60    async def id(self) -> KilnBuiltInToolId:
61        return self._id
62
63    async def name(self) -> str:
64        return self._name
65
66    async def description(self) -> str:
67        return self._description
68
69    async def toolcall_definition(self) -> Dict[str, Any]:
70        """Generate OpenAI-compatible tool definition."""
71        return {
72            "type": "function",
73            "function": {
74                "name": await self.name(),
75                "description": await self.description(),
76                "parameters": self._parameters_schema,
77            },
78        }
79
80    @abstractmethod
81    async def run(self, **kwargs) -> str:
82        """Subclasses must implement the actual tool logic."""
83        pass

Base helper class that provides common functionality for tool implementations. Subclasses only need to implement run() and provide tool configuration.

async def id(self) -> kiln_ai.datamodel.tool_id.KilnBuiltInToolId:
60    async def id(self) -> KilnBuiltInToolId:
61        return self._id

Return a unique identifier for this tool.

async def name(self) -> str:
63    async def name(self) -> str:
64        return self._name

Return the tool name (function name) of this tool.

async def description(self) -> str:
66    async def description(self) -> str:
67        return self._description

Return a description of what this tool does.

async def toolcall_definition(self) -> Dict[str, Any]:
69    async def toolcall_definition(self) -> Dict[str, Any]:
70        """Generate OpenAI-compatible tool definition."""
71        return {
72            "type": "function",
73            "function": {
74                "name": await self.name(),
75                "description": await self.description(),
76                "parameters": self._parameters_schema,
77            },
78        }

Generate OpenAI-compatible tool definition.

@abstractmethod
async def run(self, **kwargs) -> str:
80    @abstractmethod
81    async def run(self, **kwargs) -> str:
82        """Subclasses must implement the actual tool logic."""
83        pass

Subclasses must implement the actual tool logic.

class KilnToolInterface(abc.ABC):
 9class KilnToolInterface(ABC):
10    """
11    Abstract interface defining the core API that all Kiln tools must implement.
12    This ensures consistency across all tool implementations.
13    """
14
15    @abstractmethod
16    async def run(self, **kwargs) -> Any:
17        """Execute the tool with the given parameters."""
18        pass
19
20    @abstractmethod
21    async def toolcall_definition(self) -> Dict[str, Any]:
22        """Return the OpenAI-compatible tool definition for this tool."""
23        pass
24
25    @abstractmethod
26    async def id(self) -> ToolId:
27        """Return a unique identifier for this tool."""
28        pass
29
30    @abstractmethod
31    async def name(self) -> str:
32        """Return the tool name (function name) of this tool."""
33        pass
34
35    @abstractmethod
36    async def description(self) -> str:
37        """Return a description of what this tool does."""
38        pass

Abstract interface defining the core API that all Kiln tools must implement. This ensures consistency across all tool implementations.

@abstractmethod
async def run(self, **kwargs) -> Any:
15    @abstractmethod
16    async def run(self, **kwargs) -> Any:
17        """Execute the tool with the given parameters."""
18        pass

Execute the tool with the given parameters.

@abstractmethod
async def toolcall_definition(self) -> Dict[str, Any]:
20    @abstractmethod
21    async def toolcall_definition(self) -> Dict[str, Any]:
22        """Return the OpenAI-compatible tool definition for this tool."""
23        pass

Return the OpenAI-compatible tool definition for this tool.

@abstractmethod
async def id( self) -> Annotated[str, AfterValidator(func=<function <lambda> at 0x7f29ba9d65c0>)]:
25    @abstractmethod
26    async def id(self) -> ToolId:
27        """Return a unique identifier for this tool."""
28        pass

Return a unique identifier for this tool.

@abstractmethod
async def name(self) -> str:
30    @abstractmethod
31    async def name(self) -> str:
32        """Return the tool name (function name) of this tool."""
33        pass

Return the tool name (function name) of this tool.

@abstractmethod
async def description(self) -> str:
35    @abstractmethod
36    async def description(self) -> str:
37        """Return a description of what this tool does."""
38        pass

Return a description of what this tool does.

def tool_from_id( tool_id: str, task: kiln_ai.datamodel.Task | None = None) -> KilnToolInterface:
20def tool_from_id(tool_id: str, task: Task | None = None) -> KilnToolInterface:
21    """
22    Get a tool from its ID.
23    """
24    # Check built-in tools
25    if tool_id in [member.value for member in KilnBuiltInToolId]:
26        typed_tool_id = KilnBuiltInToolId(tool_id)
27        match typed_tool_id:
28            case KilnBuiltInToolId.ADD_NUMBERS:
29                return AddTool()
30            case KilnBuiltInToolId.SUBTRACT_NUMBERS:
31                return SubtractTool()
32            case KilnBuiltInToolId.MULTIPLY_NUMBERS:
33                return MultiplyTool()
34            case KilnBuiltInToolId.DIVIDE_NUMBERS:
35                return DivideTool()
36            case _:
37                raise_exhaustive_enum_error(typed_tool_id)
38
39    # Check MCP Server Tools
40    if tool_id.startswith((MCP_REMOTE_TOOL_ID_PREFIX, MCP_LOCAL_TOOL_ID_PREFIX)):
41        project = task.parent_project() if task is not None else None
42        if project is None:
43            raise ValueError(
44                f"Unable to resolve tool from id: {tool_id}. Requires a parent project/task."
45            )
46
47        # Get the tool server ID and tool name from the ID
48        tool_server_id, tool_name = mcp_server_and_tool_name_from_id(tool_id)
49
50        server = next(
51            (
52                server
53                for server in project.external_tool_servers()
54                if server.id == tool_server_id
55            ),
56            None,
57        )
58        if server is None:
59            raise ValueError(
60                f"External tool server not found: {tool_server_id} in project ID {project.id}"
61            )
62
63        return MCPServerTool(server, tool_name)
64
65    raise ValueError(f"Tool ID {tool_id} not found in tool registry")

Get a tool from its ID.