0
# Tool System
1
2
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.
3
4
## Capabilities
5
6
### Core Tool Interface
7
8
Base classes and protocols for tool implementation.
9
10
```python { .api }
11
from kiln_ai.tools import KilnTool, KilnToolInterface
12
13
class KilnTool:
14
"""
15
Base class for tools that AI agents can use.
16
17
Properties:
18
- id (str): Tool identifier
19
- name (str): Tool name
20
- description (str): Tool description for AI models
21
- input_schema (dict): JSON schema for tool inputs
22
23
Methods:
24
- invoke(): Execute the tool synchronously
25
- async_invoke(): Execute the tool asynchronously
26
"""
27
28
def __init__(
29
self,
30
id: str,
31
name: str,
32
description: str,
33
input_schema: dict
34
):
35
"""
36
Initialize tool.
37
38
Parameters:
39
- id (str): Unique tool identifier
40
- name (str): Human-readable tool name
41
- description (str): Description of what the tool does
42
- input_schema (dict): JSON schema defining expected inputs
43
"""
44
45
def invoke(self, **kwargs) -> dict:
46
"""
47
Execute tool synchronously.
48
49
Parameters:
50
- **kwargs: Tool inputs matching input_schema
51
52
Returns:
53
dict: Tool execution result
54
"""
55
56
async def async_invoke(self, **kwargs) -> dict:
57
"""
58
Execute tool asynchronously.
59
60
Parameters:
61
- **kwargs: Tool inputs matching input_schema
62
63
Returns:
64
dict: Tool execution result
65
"""
66
67
class KilnToolInterface:
68
"""
69
Tool interface protocol.
70
71
Defines the contract that all tools must implement.
72
"""
73
74
def invoke(self, **kwargs) -> dict:
75
"""
76
Execute the tool.
77
78
Parameters:
79
- **kwargs: Tool inputs
80
81
Returns:
82
dict: Execution result
83
"""
84
```
85
86
### Tool Registry
87
88
Registry for managing and discovering tools.
89
90
```python { .api }
91
from kiln_ai.tools.tool_registry import tool_from_id
92
93
def tool_from_id(tool_id: str) -> 'KilnTool':
94
"""
95
Get tool instance by identifier.
96
97
Parameters:
98
- tool_id (str): Tool identifier
99
100
Returns:
101
KilnTool: Tool instance
102
103
Raises:
104
KeyError: If tool not found
105
"""
106
```
107
108
### Built-in Math Tools
109
110
Mathematical operation tools for AI agents.
111
112
```python { .api }
113
from kiln_ai.tools.built_in_tools.math_tools import (
114
AddTool,
115
SubtractTool,
116
MultiplyTool,
117
DivideTool
118
)
119
120
class AddTool(KilnTool):
121
"""
122
Addition operation tool.
123
124
Input schema:
125
{
126
"type": "object",
127
"properties": {
128
"a": {"type": "number"},
129
"b": {"type": "number"}
130
},
131
"required": ["a", "b"]
132
}
133
134
Returns:
135
{"result": number} - Sum of a and b
136
"""
137
138
def invoke(self, a: float, b: float) -> dict:
139
"""
140
Add two numbers.
141
142
Parameters:
143
- a (float): First number
144
- b (float): Second number
145
146
Returns:
147
dict: {"result": a + b}
148
"""
149
150
class SubtractTool(KilnTool):
151
"""
152
Subtraction operation tool.
153
154
Input schema:
155
{
156
"type": "object",
157
"properties": {
158
"a": {"type": "number"},
159
"b": {"type": "number"}
160
},
161
"required": ["a", "b"]
162
}
163
164
Returns:
165
{"result": number} - Difference of a and b
166
"""
167
168
def invoke(self, a: float, b: float) -> dict:
169
"""
170
Subtract b from a.
171
172
Parameters:
173
- a (float): First number
174
- b (float): Second number
175
176
Returns:
177
dict: {"result": a - b}
178
"""
179
180
class MultiplyTool(KilnTool):
181
"""
182
Multiplication operation tool.
183
184
Input schema:
185
{
186
"type": "object",
187
"properties": {
188
"a": {"type": "number"},
189
"b": {"type": "number"}
190
},
191
"required": ["a", "b"]
192
}
193
194
Returns:
195
{"result": number} - Product of a and b
196
"""
197
198
def invoke(self, a: float, b: float) -> dict:
199
"""
200
Multiply two numbers.
201
202
Parameters:
203
- a (float): First number
204
- b (float): Second number
205
206
Returns:
207
dict: {"result": a * b}
208
"""
209
210
class DivideTool(KilnTool):
211
"""
212
Division operation tool.
213
214
Input schema:
215
{
216
"type": "object",
217
"properties": {
218
"a": {"type": "number"},
219
"b": {"type": "number"}
220
},
221
"required": ["a", "b"]
222
}
223
224
Returns:
225
{"result": number} - Quotient of a divided by b
226
227
Raises:
228
ZeroDivisionError: If b is zero
229
"""
230
231
def invoke(self, a: float, b: float) -> dict:
232
"""
233
Divide a by b.
234
235
Parameters:
236
- a (float): Numerator
237
- b (float): Denominator
238
239
Returns:
240
dict: {"result": a / b}
241
242
Raises:
243
ZeroDivisionError: If b is zero
244
"""
245
```
246
247
### External Tool Servers
248
249
MCP (Model Control Protocol) tool server integration.
250
251
```python { .api }
252
from kiln_ai.datamodel import ExternalToolServer
253
254
class ExternalToolServer:
255
"""
256
MCP tool server configuration.
257
258
Properties:
259
- name (str): Server name
260
- server_url (str): Server URL endpoint
261
- api_key (str | None): API key for authentication
262
"""
263
```
264
265
## Usage Examples
266
267
### Using Built-in Tools
268
269
```python
270
from kiln_ai.tools.tool_registry import tool_from_id
271
272
# Get addition tool
273
add_tool = tool_from_id("add")
274
275
# Use the tool
276
result = add_tool.invoke(a=5, b=3)
277
print(result) # {"result": 8}
278
279
# Get multiplication tool
280
multiply_tool = tool_from_id("multiply")
281
result = multiply_tool.invoke(a=4, b=7)
282
print(result) # {"result": 28}
283
```
284
285
### Creating Custom Tool
286
287
```python
288
from kiln_ai.tools import KilnTool
289
290
class WeatherTool(KilnTool):
291
"""Tool for getting weather information."""
292
293
def __init__(self):
294
super().__init__(
295
id="weather_lookup",
296
name="Weather Lookup",
297
description="Get current weather for a location",
298
input_schema={
299
"type": "object",
300
"properties": {
301
"location": {
302
"type": "string",
303
"description": "City name or zip code"
304
},
305
"units": {
306
"type": "string",
307
"enum": ["celsius", "fahrenheit"],
308
"default": "celsius"
309
}
310
},
311
"required": ["location"]
312
}
313
)
314
315
def invoke(self, location: str, units: str = "celsius") -> dict:
316
"""Get weather for location."""
317
# Implementation would call weather API
318
return {
319
"location": location,
320
"temperature": 72,
321
"units": units,
322
"conditions": "sunny"
323
}
324
325
# Use custom tool
326
weather = WeatherTool()
327
result = weather.invoke(location="San Francisco", units="fahrenheit")
328
print(result)
329
```
330
331
### Async Tool Implementation
332
333
```python
334
from kiln_ai.tools import KilnTool
335
import aiohttp
336
337
class AsyncAPITool(KilnTool):
338
"""Tool that makes async API calls."""
339
340
def __init__(self):
341
super().__init__(
342
id="api_caller",
343
name="API Caller",
344
description="Make HTTP API requests",
345
input_schema={
346
"type": "object",
347
"properties": {
348
"url": {"type": "string"},
349
"method": {
350
"type": "string",
351
"enum": ["GET", "POST"]
352
}
353
},
354
"required": ["url", "method"]
355
}
356
)
357
358
async def async_invoke(self, url: str, method: str) -> dict:
359
"""Make async HTTP request."""
360
async with aiohttp.ClientSession() as session:
361
if method == "GET":
362
async with session.get(url) as response:
363
data = await response.json()
364
return {"status": response.status, "data": data}
365
elif method == "POST":
366
async with session.post(url) as response:
367
data = await response.json()
368
return {"status": response.status, "data": data}
369
370
# Use async tool
371
api_tool = AsyncAPITool()
372
result = await api_tool.async_invoke(
373
url="https://api.example.com/data",
374
method="GET"
375
)
376
```
377
378
### Tool with AI Model
379
380
```python
381
from kiln_ai.tools import KilnTool
382
from kiln_ai.tools.tool_registry import tool_from_id
383
from kiln_ai.datamodel import Task
384
from kiln_ai.adapters import adapter_for_task
385
386
# Define available tools
387
tools = [
388
tool_from_id("add"),
389
tool_from_id("multiply"),
390
tool_from_id("divide")
391
]
392
393
# Create task
394
task = Task(
395
name="calculator",
396
instruction="Use the available tools to solve math problems."
397
)
398
399
# Create adapter with tool support
400
adapter = adapter_for_task(
401
task,
402
model_name="gpt_4o",
403
provider="openai",
404
config={"tools": tools}
405
)
406
407
# AI model can now call tools
408
result = await adapter.invoke("What is 15 * 7, then divide by 3?")
409
# Model would call multiply_tool(15, 7) -> 105
410
# Then call divide_tool(105, 3) -> 35
411
```
412
413
### Tool Schema Definition
414
415
```python
416
from kiln_ai.tools import KilnTool
417
418
class SearchTool(KilnTool):
419
"""Tool with complex input schema."""
420
421
def __init__(self):
422
super().__init__(
423
id="search",
424
name="Search",
425
description="Search for information",
426
input_schema={
427
"type": "object",
428
"properties": {
429
"query": {
430
"type": "string",
431
"description": "Search query"
432
},
433
"filters": {
434
"type": "object",
435
"properties": {
436
"date_from": {"type": "string", "format": "date"},
437
"date_to": {"type": "string", "format": "date"},
438
"source": {
439
"type": "array",
440
"items": {"type": "string"}
441
}
442
}
443
},
444
"max_results": {
445
"type": "integer",
446
"minimum": 1,
447
"maximum": 100,
448
"default": 10
449
}
450
},
451
"required": ["query"]
452
}
453
)
454
455
def invoke(
456
self,
457
query: str,
458
filters: dict = None,
459
max_results: int = 10
460
) -> dict:
461
"""Execute search."""
462
# Search implementation
463
return {
464
"query": query,
465
"results": [],
466
"count": 0
467
}
468
```
469
470
### Error Handling in Tools
471
472
```python
473
from kiln_ai.tools import KilnTool
474
475
class SafeDivideTool(KilnTool):
476
"""Division tool with error handling."""
477
478
def __init__(self):
479
super().__init__(
480
id="safe_divide",
481
name="Safe Divide",
482
description="Divide two numbers with error handling",
483
input_schema={
484
"type": "object",
485
"properties": {
486
"a": {"type": "number"},
487
"b": {"type": "number"}
488
},
489
"required": ["a", "b"]
490
}
491
)
492
493
def invoke(self, a: float, b: float) -> dict:
494
"""Divide with error handling."""
495
try:
496
if b == 0:
497
return {
498
"error": "Division by zero",
499
"result": None
500
}
501
return {"result": a / b}
502
except Exception as e:
503
return {
504
"error": str(e),
505
"result": None
506
}
507
508
# Use safe division
509
tool = SafeDivideTool()
510
result = tool.invoke(a=10, b=0)
511
print(result) # {"error": "Division by zero", "result": None}
512
```
513
514
### Tool Registry Pattern
515
516
```python
517
from kiln_ai.tools import KilnTool
518
519
# Custom tool registry
520
CUSTOM_TOOLS = {}
521
522
def register_tool(tool: KilnTool):
523
"""Register a tool."""
524
CUSTOM_TOOLS[tool.id] = tool
525
return tool
526
527
@register_tool
528
class CustomTool1(KilnTool):
529
def __init__(self):
530
super().__init__(
531
id="custom1",
532
name="Custom Tool 1",
533
description="First custom tool",
534
input_schema={"type": "object", "properties": {}}
535
)
536
537
def invoke(self, **kwargs):
538
return {"status": "success"}
539
540
@register_tool
541
class CustomTool2(KilnTool):
542
def __init__(self):
543
super().__init__(
544
id="custom2",
545
name="Custom Tool 2",
546
description="Second custom tool",
547
input_schema={"type": "object", "properties": {}}
548
)
549
550
def invoke(self, **kwargs):
551
return {"status": "success"}
552
553
# Get tool from custom registry
554
tool = CUSTOM_TOOLS["custom1"]
555
```
556
557
### External Tool Server
558
559
```python
560
from kiln_ai.datamodel import ExternalToolServer
561
562
# Configure external MCP tool server
563
tool_server = ExternalToolServer(
564
name="my_tool_server",
565
server_url="https://tools.example.com/mcp",
566
api_key="secret_key_123"
567
)
568
569
# Use in task configuration
570
# (Integration with task execution would be implemented
571
# by the adapter to call the external server)
572
```
573
574
### Tool Chain Execution
575
576
```python
577
from kiln_ai.tools.tool_registry import tool_from_id
578
579
# Get tools
580
add = tool_from_id("add")
581
multiply = tool_from_id("multiply")
582
583
# Chain tool executions
584
def calculate_expression(x, y, z):
585
"""Calculate (x + y) * z"""
586
# Step 1: Add x and y
587
sum_result = add.invoke(a=x, b=y)
588
intermediate = sum_result["result"]
589
590
# Step 2: Multiply result by z
591
final_result = multiply.invoke(a=intermediate, b=z)
592
return final_result["result"]
593
594
result = calculate_expression(5, 3, 4)
595
print(f"(5 + 3) * 4 = {result}") # 32
596
```
597
598
### Tool Input Validation
599
600
```python
601
from kiln_ai.tools import KilnTool
602
import jsonschema
603
604
class ValidatedTool(KilnTool):
605
"""Tool with input validation."""
606
607
def __init__(self):
608
super().__init__(
609
id="validated_tool",
610
name="Validated Tool",
611
description="Tool with strict input validation",
612
input_schema={
613
"type": "object",
614
"properties": {
615
"email": {
616
"type": "string",
617
"format": "email"
618
},
619
"age": {
620
"type": "integer",
621
"minimum": 0,
622
"maximum": 150
623
}
624
},
625
"required": ["email"]
626
}
627
)
628
629
def invoke(self, **kwargs) -> dict:
630
"""Execute with validation."""
631
try:
632
# Validate inputs against schema
633
jsonschema.validate(kwargs, self.input_schema)
634
# Process valid inputs
635
return {"status": "success", "inputs": kwargs}
636
except jsonschema.ValidationError as e:
637
return {"error": str(e), "status": "validation_failed"}
638
639
# Test validation
640
tool = ValidatedTool()
641
642
# Valid input
643
result = tool.invoke(email="user@example.com", age=25)
644
print(result) # Success
645
646
# Invalid input
647
result = tool.invoke(email="invalid", age=25)
648
print(result) # Validation error
649
```
650
651
### Tool with State
652
653
```python
654
from kiln_ai.tools import KilnTool
655
656
class CounterTool(KilnTool):
657
"""Tool that maintains state."""
658
659
def __init__(self):
660
super().__init__(
661
id="counter",
662
name="Counter",
663
description="Increment a counter",
664
input_schema={
665
"type": "object",
666
"properties": {
667
"increment": {
668
"type": "integer",
669
"default": 1
670
}
671
}
672
}
673
)
674
self.count = 0
675
676
def invoke(self, increment: int = 1) -> dict:
677
"""Increment counter."""
678
self.count += increment
679
return {"count": self.count}
680
681
def reset(self):
682
"""Reset counter."""
683
self.count = 0
684
685
# Use stateful tool
686
counter = CounterTool()
687
print(counter.invoke()) # {"count": 1}
688
print(counter.invoke(increment=5)) # {"count": 6}
689
print(counter.invoke()) # {"count": 7}
690
counter.reset()
691
print(counter.invoke()) # {"count": 1}
692
```
693