kiln_ai.tools
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:
Return a unique identifier for this tool.
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.
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.
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.