tessl/pypi-kiln-ai
Kiln AI is a comprehensive platform for building, evaluating, and deploying AI systems with dataset management, model fine-tuning, RAG, and evaluation capabilities.
tools.mddocs/
Tool System
Tool system enabling AI agents to use external functions and APIs. Provides a structured interface for defining tools, managing tool registries, and integrating with AI models that support function calling.
Capabilities
Core Tool Interface
Base classes and protocols for tool implementation.
from kiln_ai.tools import KilnTool, KilnToolInterface
class KilnTool:
"""
Base class for tools that AI agents can use.
Properties:
- id (str): Tool identifier
- name (str): Tool name
- description (str): Tool description for AI models
- input_schema (dict): JSON schema for tool inputs
Methods:
- invoke(): Execute the tool synchronously
- async_invoke(): Execute the tool asynchronously
"""
def __init__(
self,
id: str,
name: str,
description: str,
input_schema: dict
):
"""
Initialize tool.
Parameters:
- id (str): Unique tool identifier
- name (str): Human-readable tool name
- description (str): Description of what the tool does
- input_schema (dict): JSON schema defining expected inputs
"""
def invoke(self, **kwargs) -> dict:
"""
Execute tool synchronously.
Parameters:
- **kwargs: Tool inputs matching input_schema
Returns:
dict: Tool execution result
"""
async def async_invoke(self, **kwargs) -> dict:
"""
Execute tool asynchronously.
Parameters:
- **kwargs: Tool inputs matching input_schema
Returns:
dict: Tool execution result
"""
class KilnToolInterface:
"""
Tool interface protocol.
Defines the contract that all tools must implement.
"""
def invoke(self, **kwargs) -> dict:
"""
Execute the tool.
Parameters:
- **kwargs: Tool inputs
Returns:
dict: Execution result
"""Tool Registry
Registry for managing and discovering tools.
from kiln_ai.tools.tool_registry import tool_from_id
def tool_from_id(tool_id: str) -> 'KilnTool':
"""
Get tool instance by identifier.
Parameters:
- tool_id (str): Tool identifier
Returns:
KilnTool: Tool instance
Raises:
KeyError: If tool not found
"""Built-in Math Tools
Mathematical operation tools for AI agents.
from kiln_ai.tools.built_in_tools.math_tools import (
AddTool,
SubtractTool,
MultiplyTool,
DivideTool
)
class AddTool(KilnTool):
"""
Addition operation tool.
Input schema:
{
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
Returns:
{"result": number} - Sum of a and b
"""
def invoke(self, a: float, b: float) -> dict:
"""
Add two numbers.
Parameters:
- a (float): First number
- b (float): Second number
Returns:
dict: {"result": a + b}
"""
class SubtractTool(KilnTool):
"""
Subtraction operation tool.
Input schema:
{
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
Returns:
{"result": number} - Difference of a and b
"""
def invoke(self, a: float, b: float) -> dict:
"""
Subtract b from a.
Parameters:
- a (float): First number
- b (float): Second number
Returns:
dict: {"result": a - b}
"""
class MultiplyTool(KilnTool):
"""
Multiplication operation tool.
Input schema:
{
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
Returns:
{"result": number} - Product of a and b
"""
def invoke(self, a: float, b: float) -> dict:
"""
Multiply two numbers.
Parameters:
- a (float): First number
- b (float): Second number
Returns:
dict: {"result": a * b}
"""
class DivideTool(KilnTool):
"""
Division operation tool.
Input schema:
{
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
Returns:
{"result": number} - Quotient of a divided by b
Raises:
ZeroDivisionError: If b is zero
"""
def invoke(self, a: float, b: float) -> dict:
"""
Divide a by b.
Parameters:
- a (float): Numerator
- b (float): Denominator
Returns:
dict: {"result": a / b}
Raises:
ZeroDivisionError: If b is zero
"""External Tool Servers
MCP (Model Control Protocol) tool server integration.
from kiln_ai.datamodel import ExternalToolServer
class ExternalToolServer:
"""
MCP tool server configuration.
Properties:
- name (str): Server name
- server_url (str): Server URL endpoint
- api_key (str | None): API key for authentication
"""Usage Examples
Using Built-in Tools
from kiln_ai.tools.tool_registry import tool_from_id
# Get addition tool
add_tool = tool_from_id("add")
# Use the tool
result = add_tool.invoke(a=5, b=3)
print(result) # {"result": 8}
# Get multiplication tool
multiply_tool = tool_from_id("multiply")
result = multiply_tool.invoke(a=4, b=7)
print(result) # {"result": 28}Creating Custom Tool
from kiln_ai.tools import KilnTool
class WeatherTool(KilnTool):
"""Tool for getting weather information."""
def __init__(self):
super().__init__(
id="weather_lookup",
name="Weather Lookup",
description="Get current weather for a location",
input_schema={
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name or zip code"
},
"units": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius"
}
},
"required": ["location"]
}
)
def invoke(self, location: str, units: str = "celsius") -> dict:
"""Get weather for location."""
# Implementation would call weather API
return {
"location": location,
"temperature": 72,
"units": units,
"conditions": "sunny"
}
# Use custom tool
weather = WeatherTool()
result = weather.invoke(location="San Francisco", units="fahrenheit")
print(result)Async Tool Implementation
from kiln_ai.tools import KilnTool
import aiohttp
class AsyncAPITool(KilnTool):
"""Tool that makes async API calls."""
def __init__(self):
super().__init__(
id="api_caller",
name="API Caller",
description="Make HTTP API requests",
input_schema={
"type": "object",
"properties": {
"url": {"type": "string"},
"method": {
"type": "string",
"enum": ["GET", "POST"]
}
},
"required": ["url", "method"]
}
)
async def async_invoke(self, url: str, method: str) -> dict:
"""Make async HTTP request."""
async with aiohttp.ClientSession() as session:
if method == "GET":
async with session.get(url) as response:
data = await response.json()
return {"status": response.status, "data": data}
elif method == "POST":
async with session.post(url) as response:
data = await response.json()
return {"status": response.status, "data": data}
# Use async tool
api_tool = AsyncAPITool()
result = await api_tool.async_invoke(
url="https://api.example.com/data",
method="GET"
)Tool with AI Model
from kiln_ai.tools import KilnTool
from kiln_ai.tools.tool_registry import tool_from_id
from kiln_ai.datamodel import Task
from kiln_ai.adapters import adapter_for_task
# Define available tools
tools = [
tool_from_id("add"),
tool_from_id("multiply"),
tool_from_id("divide")
]
# Create task
task = Task(
name="calculator",
instruction="Use the available tools to solve math problems."
)
# Create adapter with tool support
adapter = adapter_for_task(
task,
model_name="gpt_4o",
provider="openai",
config={"tools": tools}
)
# AI model can now call tools
result = await adapter.invoke("What is 15 * 7, then divide by 3?")
# Model would call multiply_tool(15, 7) -> 105
# Then call divide_tool(105, 3) -> 35Tool Schema Definition
from kiln_ai.tools import KilnTool
class SearchTool(KilnTool):
"""Tool with complex input schema."""
def __init__(self):
super().__init__(
id="search",
name="Search",
description="Search for information",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
},
"filters": {
"type": "object",
"properties": {
"date_from": {"type": "string", "format": "date"},
"date_to": {"type": "string", "format": "date"},
"source": {
"type": "array",
"items": {"type": "string"}
}
}
},
"max_results": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"default": 10
}
},
"required": ["query"]
}
)
def invoke(
self,
query: str,
filters: dict = None,
max_results: int = 10
) -> dict:
"""Execute search."""
# Search implementation
return {
"query": query,
"results": [],
"count": 0
}Error Handling in Tools
from kiln_ai.tools import KilnTool
class SafeDivideTool(KilnTool):
"""Division tool with error handling."""
def __init__(self):
super().__init__(
id="safe_divide",
name="Safe Divide",
description="Divide two numbers with error handling",
input_schema={
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
)
def invoke(self, a: float, b: float) -> dict:
"""Divide with error handling."""
try:
if b == 0:
return {
"error": "Division by zero",
"result": None
}
return {"result": a / b}
except Exception as e:
return {
"error": str(e),
"result": None
}
# Use safe division
tool = SafeDivideTool()
result = tool.invoke(a=10, b=0)
print(result) # {"error": "Division by zero", "result": None}Tool Registry Pattern
from kiln_ai.tools import KilnTool
# Custom tool registry
CUSTOM_TOOLS = {}
def register_tool(tool: KilnTool):
"""Register a tool."""
CUSTOM_TOOLS[tool.id] = tool
return tool
@register_tool
class CustomTool1(KilnTool):
def __init__(self):
super().__init__(
id="custom1",
name="Custom Tool 1",
description="First custom tool",
input_schema={"type": "object", "properties": {}}
)
def invoke(self, **kwargs):
return {"status": "success"}
@register_tool
class CustomTool2(KilnTool):
def __init__(self):
super().__init__(
id="custom2",
name="Custom Tool 2",
description="Second custom tool",
input_schema={"type": "object", "properties": {}}
)
def invoke(self, **kwargs):
return {"status": "success"}
# Get tool from custom registry
tool = CUSTOM_TOOLS["custom1"]External Tool Server
from kiln_ai.datamodel import ExternalToolServer
# Configure external MCP tool server
tool_server = ExternalToolServer(
name="my_tool_server",
server_url="https://tools.example.com/mcp",
api_key="secret_key_123"
)
# Use in task configuration
# (Integration with task execution would be implemented
# by the adapter to call the external server)Tool Chain Execution
from kiln_ai.tools.tool_registry import tool_from_id
# Get tools
add = tool_from_id("add")
multiply = tool_from_id("multiply")
# Chain tool executions
def calculate_expression(x, y, z):
"""Calculate (x + y) * z"""
# Step 1: Add x and y
sum_result = add.invoke(a=x, b=y)
intermediate = sum_result["result"]
# Step 2: Multiply result by z
final_result = multiply.invoke(a=intermediate, b=z)
return final_result["result"]
result = calculate_expression(5, 3, 4)
print(f"(5 + 3) * 4 = {result}") # 32Tool Input Validation
from kiln_ai.tools import KilnTool
import jsonschema
class ValidatedTool(KilnTool):
"""Tool with input validation."""
def __init__(self):
super().__init__(
id="validated_tool",
name="Validated Tool",
description="Tool with strict input validation",
input_schema={
"type": "object",
"properties": {
"email": {
"type": "string",
"format": "email"
},
"age": {
"type": "integer",
"minimum": 0,
"maximum": 150
}
},
"required": ["email"]
}
)
def invoke(self, **kwargs) -> dict:
"""Execute with validation."""
try:
# Validate inputs against schema
jsonschema.validate(kwargs, self.input_schema)
# Process valid inputs
return {"status": "success", "inputs": kwargs}
except jsonschema.ValidationError as e:
return {"error": str(e), "status": "validation_failed"}
# Test validation
tool = ValidatedTool()
# Valid input
result = tool.invoke(email="user@example.com", age=25)
print(result) # Success
# Invalid input
result = tool.invoke(email="invalid", age=25)
print(result) # Validation errorTool with State
from kiln_ai.tools import KilnTool
class CounterTool(KilnTool):
"""Tool that maintains state."""
def __init__(self):
super().__init__(
id="counter",
name="Counter",
description="Increment a counter",
input_schema={
"type": "object",
"properties": {
"increment": {
"type": "integer",
"default": 1
}
}
}
)
self.count = 0
def invoke(self, increment: int = 1) -> dict:
"""Increment counter."""
self.count += increment
return {"count": self.count}
def reset(self):
"""Reset counter."""
self.count = 0
# Use stateful tool
counter = CounterTool()
print(counter.invoke()) # {"count": 1}
print(counter.invoke(increment=5)) # {"count": 6}
print(counter.invoke()) # {"count": 7}
counter.reset()
print(counter.invoke()) # {"count": 1}Install with Tessl CLI
npx tessl i tessl/pypi-kiln-ai