0
# Configuration and Utilities
1

2
Configuration management, formatting utilities, and async lock management for the Kiln AI system. Provides centralized configuration storage, helper functions, and concurrency utilities.
3

4
## Capabilities
5

6
### Configuration Management
7

8
Centralized configuration for Kiln AI with singleton access pattern.
9

10
```python { .api }
11
from kiln_ai.utils.config import Config, ConfigProperty, MCP_SECRETS_KEY
12

13
class Config:
14
    """
15
    Configuration management for Kiln AI.
16

17
    Stores configuration in ~/.kiln_settings/config.yaml
18

19
    Properties:
20
    - custom_models (list[str]): Custom model identifiers in format "provider::model"
21
    - openai_compatible_providers (list[dict]): OpenAI-compatible provider configs
22
    """
23

24
    @classmethod
25
    def shared(cls) -> 'Config':
26
        """
27
        Get singleton configuration instance.
28

29
        Returns:
30
        Config: Shared configuration instance
31
        """
32

33
    def save(self) -> None:
34
        """
35
        Save configuration to disk.
36

37
        Writes to ~/.kiln_settings/config.yaml
38
        """
39

40
    def load(self) -> None:
41
        """
42
        Load configuration from disk.
43

44
        Reads from ~/.kiln_settings/config.yaml
45
        """
46

47
    def get(self, key: str, default=None):
48
        """
49
        Get configuration value.
50

51
        Parameters:
52
        - key (str): Configuration key
53
        - default: Default value if key not found
54

55
        Returns:
56
        Any: Configuration value or default
57
        """
58

59
    def set(self, key: str, value) -> None:
60
        """
61
        Set configuration value.
62

63
        Parameters:
64
        - key (str): Configuration key
65
        - value: Value to set
66
        """
67

68
class ConfigProperty:
69
    """
70
    Configuration property definition.
71

72
    Properties:
73
    - key (str): Property key
74
    - default: Default value
75
    - description (str): Property description
76
    """
77

78
# MCP secrets configuration key
79
MCP_SECRETS_KEY = "mcp_secrets"
80
```
81

82
### Async Lock Management
83

84
Manage async locks for concurrent operations.
85

86
```python { .api }
87
from kiln_ai.utils.lock import AsyncLockManager, shared_async_lock_manager
88

89
class AsyncLockManager:
90
    """
91
    Manage async locks for concurrency control.
92

93
    Methods:
94
    - acquire(): Acquire a lock
95
    - release(): Release a lock
96
    - with_lock(): Context manager for lock
97
    """
98

99
    async def acquire(self, lock_id: str) -> None:
100
        """
101
        Acquire a lock.
102

103
        Parameters:
104
        - lock_id (str): Lock identifier
105

106
        Blocks until lock is available.
107
        """
108

109
    async def release(self, lock_id: str) -> None:
110
        """
111
        Release a lock.
112

113
        Parameters:
114
        - lock_id (str): Lock identifier
115
        """
116

117
    async def with_lock(self, lock_id: str):
118
        """
119
        Context manager for lock acquisition.
120

121
        Parameters:
122
        - lock_id (str): Lock identifier
123

124
        Usage:
125
        async with lock_manager.with_lock("my_lock"):
126
            # Critical section
127
            pass
128
        """
129

130
# Shared lock manager singleton
131
shared_async_lock_manager = AsyncLockManager()
132
```
133

134
### Formatting Utilities
135

136
String formatting and conversion utilities.
137

138
```python { .api }
139
from kiln_ai.utils.formatting import snake_case
140

141
def snake_case(text: str) -> str:
142
    """
143
    Convert string to snake_case.
144

145
    Parameters:
146
    - text (str): Input text (can be camelCase, PascalCase, or mixed)
147

148
    Returns:
149
    str: snake_case formatted string
150

151
    Examples:
152
    - "HelloWorld" -> "hello_world"
153
    - "camelCase" -> "camel_case"
154
    - "already_snake" -> "already_snake"
155
    """
156
```
157

158
## Usage Examples
159

160
### Basic Configuration
161

162
```python
163
from kiln_ai.utils.config import Config
164

165
# Get shared configuration instance
166
config = Config.shared()
167

168
# Access configuration values
169
custom_models = config.custom_models or []
170
print(f"Custom models: {custom_models}")
171

172
providers = config.openai_compatible_providers or []
173
print(f"Custom providers: {len(providers)}")
174

175
# Save configuration
176
config.save()
177
```
178

179
### Adding Custom Models
180

181
```python
182
from kiln_ai.utils.config import Config
183

184
config = Config.shared()
185

186
# Add custom model
187
new_model = "openai::gpt-3.5-turbo-custom"
188
custom_models = config.custom_models or []
189

190
if new_model not in custom_models:
191
    custom_models.append(new_model)
192
    config.custom_models = custom_models
193
    config.save()
194
    print(f"Added custom model: {new_model}")
195

196
# List all custom models
197
print("\nCustom models:")
198
for model in config.custom_models:
199
    print(f"  - {model}")
200
```
201

202
### Adding OpenAI Compatible Provider
203

204
```python
205
from kiln_ai.utils.config import Config
206

207
config = Config.shared()
208

209
# Add custom provider
210
provider_config = {
211
    "name": "CustomOllama",
212
    "base_url": "http://localhost:11434/v1",
213
    "api_key": "ollama"
214
}
215

216
providers = config.openai_compatible_providers or []
217

218
# Check if provider already exists
219
existing = next((p for p in providers if p["name"] == provider_config["name"]), None)
220

221
if not existing:
222
    providers.append(provider_config)
223
    config.openai_compatible_providers = providers
224
    config.save()
225
    print(f"Added provider: {provider_config['name']}")
226
else:
227
    print(f"Provider {provider_config['name']} already exists")
228

229
# List all providers
230
print("\nConfigured providers:")
231
for provider in config.openai_compatible_providers:
232
    print(f"  - {provider['name']}: {provider['base_url']}")
233
```
234

235
### Configuration with Environment Variables
236

237
```python
238
from kiln_ai.utils.config import Config
239
import os
240

241
config = Config.shared()
242

243
# Get API keys from environment
244
openai_key = os.getenv("OPENAI_API_KEY")
245
anthropic_key = os.getenv("ANTHROPIC_API_KEY")
246

247
# Store in config if not already present
248
if openai_key and not config.get("openai_api_key"):
249
    config.set("openai_api_key", openai_key)
250

251
if anthropic_key and not config.get("anthropic_api_key"):
252
    config.set("anthropic_api_key", anthropic_key)
253

254
config.save()
255
```
256

257
### MCP Secrets Management
258

259
```python
260
from kiln_ai.utils.config import Config, MCP_SECRETS_KEY
261

262
config = Config.shared()
263

264
# Store MCP secrets
265
secrets = config.get(MCP_SECRETS_KEY, {})
266
secrets["my_mcp_server"] = {
267
    "api_key": "secret_key_123",
268
    "endpoint": "https://mcp.example.com"
269
}
270
config.set(MCP_SECRETS_KEY, secrets)
271
config.save()
272

273
# Retrieve MCP secrets
274
retrieved_secrets = config.get(MCP_SECRETS_KEY, {})
275
server_secret = retrieved_secrets.get("my_mcp_server")
276
print(f"MCP Server endpoint: {server_secret['endpoint']}")
277
```
278

279
### Async Lock Usage
280

281
```python
282
from kiln_ai.utils.lock import shared_async_lock_manager
283

284
# Use shared lock manager
285
async def update_shared_resource(resource_id: str):
286
    """Update resource with lock protection."""
287
    lock_id = f"resource_{resource_id}"
288

289
    async with shared_async_lock_manager.with_lock(lock_id):
290
        # Critical section - only one task can execute this at a time
291
        print(f"Updating resource {resource_id}")
292
        # Perform update...
293
        await asyncio.sleep(1)
294
        print(f"Completed update {resource_id}")
295

296
# Multiple concurrent calls will be serialized
297
await asyncio.gather(
298
    update_shared_resource("abc"),
299
    update_shared_resource("abc"),
300
    update_shared_resource("abc")
301
)
302
```
303

304
### Lock for File Operations
305

306
```python
307
from kiln_ai.utils.lock import shared_async_lock_manager
308
import asyncio
309

310
async def safe_file_write(file_path: str, content: str):
311
    """Write to file with lock protection."""
312
    lock_id = f"file_{file_path}"
313

314
    async with shared_async_lock_manager.with_lock(lock_id):
315
        # Only one task can write to this file at a time
316
        with open(file_path, "w") as f:
317
            f.write(content)
318
        await asyncio.sleep(0.1)  # Simulate processing
319

320
# Safe concurrent writes
321
await asyncio.gather(
322
    safe_file_write("/tmp/data.txt", "content1"),
323
    safe_file_write("/tmp/data.txt", "content2"),
324
    safe_file_write("/tmp/other.txt", "content3")  # Different file, can run in parallel
325
)
326
```
327

328
### String Formatting
329

330
```python
331
from kiln_ai.utils.formatting import snake_case
332

333
# Convert various formats to snake_case
334
test_strings = [
335
    "HelloWorld",
336
    "camelCase",
337
    "PascalCase",
338
    "already_snake_case",
339
    "SCREAMING_SNAKE_CASE",
340
    "Mixed-Format_String"
341
]
342

343
print("String conversions:")
344
for s in test_strings:
345
    converted = snake_case(s)
346
    print(f"  {s} -> {converted}")
347

348
# Use for generating IDs
349
class_name = "MyCustomModel"
350
model_id = snake_case(class_name)
351
print(f"\nModel ID: {model_id}")  # my_custom_model
352
```
353

354
### Configuration Validation
355

356
```python
357
from kiln_ai.utils.config import Config
358

359
def validate_config():
360
    """Validate configuration has required values."""
361
    config = Config.shared()
362

363
    errors = []
364

365
    # Check for required providers
366
    providers = config.openai_compatible_providers or []
367
    if not providers:
368
        errors.append("No OpenAI compatible providers configured")
369

370
    # Check custom models format
371
    custom_models = config.custom_models or []
372
    for model in custom_models:
373
        if "::" not in model:
374
            errors.append(f"Invalid model format: {model} (expected 'provider::model')")
375

376
    if errors:
377
        print("Configuration errors:")
378
        for error in errors:
379
            print(f"  - {error}")
380
        return False
381
    else:
382
        print("Configuration valid")
383
        return True
384

385
# Validate
386
validate_config()
387
```
388

389
### Configuration Backup
390

391
```python
392
from kiln_ai.utils.config import Config
393
import json
394
import shutil
395
from pathlib import Path
396

397
def backup_config():
398
    """Backup configuration to JSON."""
399
    config = Config.shared()
400

401
    backup_data = {
402
        "custom_models": config.custom_models,
403
        "openai_compatible_providers": config.openai_compatible_providers
404
    }
405

406
    backup_path = Path.home() / ".kiln_settings" / "config_backup.json"
407
    with open(backup_path, "w") as f:
408
        json.dump(backup_data, f, indent=2)
409

410
    print(f"Config backed up to {backup_path}")
411

412
def restore_config():
413
    """Restore configuration from JSON backup."""
414
    backup_path = Path.home() / ".kiln_settings" / "config_backup.json"
415

416
    if not backup_path.exists():
417
        print("No backup found")
418
        return
419

420
    with open(backup_path, "r") as f:
421
        backup_data = json.load(f)
422

423
    config = Config.shared()
424
    config.custom_models = backup_data.get("custom_models", [])
425
    config.openai_compatible_providers = backup_data.get("openai_compatible_providers", [])
426
    config.save()
427

428
    print("Config restored from backup")
429

430
# Backup before changes
431
backup_config()
432
```
433

434
### Per-Task Configuration
435

436
```python
437
from kiln_ai.utils.config import Config
438
from kiln_ai.datamodel import Task
439

440
class TaskConfig:
441
    """Task-specific configuration wrapper."""
442

443
    def __init__(self, task: Task):
444
        self.task = task
445
        self.global_config = Config.shared()
446

447
    def get_model_config(self, model_name: str) -> dict:
448
        """Get configuration for specific model."""
449
        # Could store task-specific overrides
450
        return {
451
            "temperature": 0.7,
452
            "max_tokens": 1000
453
        }
454

455
    def get_custom_models(self) -> list:
456
        """Get custom models including global and task-specific."""
457
        global_models = self.global_config.custom_models or []
458
        # Could add task-specific models
459
        return global_models
460

461
# Use with task
462
task = Task.load_from_file("path/to/task.kiln")
463
task_config = TaskConfig(task)
464
models = task_config.get_custom_models()
465
```
466

467
### Configuration Change Listener
468

469
```python
470
from kiln_ai.utils.config import Config
471

472
class ConfigWatcher:
473
    """Watch for configuration changes."""
474

475
    def __init__(self):
476
        self.config = Config.shared()
477
        self.last_models = list(self.config.custom_models or [])
478

479
    def check_changes(self) -> dict:
480
        """Check for configuration changes."""
481
        self.config.load()  # Reload from disk
482

483
        current_models = self.config.custom_models or []
484
        changes = {}
485

486
        # Check for new models
487
        new_models = set(current_models) - set(self.last_models)
488
        if new_models:
489
            changes["added_models"] = list(new_models)
490

491
        # Check for removed models
492
        removed_models = set(self.last_models) - set(current_models)
493
        if removed_models:
494
            changes["removed_models"] = list(removed_models)
495

496
        self.last_models = list(current_models)
497
        return changes
498

499
# Use watcher
500
watcher = ConfigWatcher()
501

502
# Later...
503
changes = watcher.check_changes()
504
if changes:
505
    print("Configuration changes detected:")
506
    print(changes)
507
```
508

509
### Thread-Safe Configuration Access
510

511
```python
512
from kiln_ai.utils.config import Config
513
from kiln_ai.utils.lock import shared_async_lock_manager
514
import asyncio
515

516
async def safe_config_update(key: str, value):
517
    """Thread-safe configuration update."""
518
    async with shared_async_lock_manager.with_lock("config_lock"):
519
        config = Config.shared()
520
        config.set(key, value)
521
        config.save()
522
        print(f"Updated {key} = {value}")
523

524
# Safe concurrent updates
525
await asyncio.gather(
526
    safe_config_update("setting1", "value1"),
527
    safe_config_update("setting2", "value2"),
528
    safe_config_update("setting3", "value3")
529
)
530
```
531

532
### Configuration Migration
533

534
```python
535
from kiln_ai.utils.config import Config
536

537
def migrate_config_v1_to_v2():
538
    """Migrate configuration from v1 to v2 format."""
539
    config = Config.shared()
540

541
    # Old format: list of model strings
542
    old_models = config.get("models", [])
543

544
    # New format: list of dicts with metadata
545
    if old_models and isinstance(old_models[0], str):
546
        new_models = []
547
        for model_str in old_models:
548
            provider, model = model_str.split("::", 1)
549
            new_models.append({
550
                "provider": provider,
551
                "model": model,
552
                "enabled": True
553
            })
554

555
        config.set("models_v2", new_models)
556
        config.save()
557
        print("Migrated configuration to v2")
558

559
# Run migration
560
migrate_config_v1_to_v2()
561
```
562