Dynamic Configuration¶
Kubiya SDK provides a powerful dynamic configuration system that allows tools to receive configuration data at runtime. This is especially useful for:
- Connecting to external services (databases, cloud providers, APIs)
- Storing authentication credentials
- Setting environment-specific parameters
Configuration Methods¶
Kubiya SDK supports two ways to define configurations:
1. Pydantic Models (Recommended)¶
Using Pydantic models provides type safety and validation:
Python
from kubiya_sdk import config_model
from pydantic import BaseModel, Field
from typing import Optional
@config_model(name="aws_config", description="AWS Configuration")
class AWSConfig(BaseModel):
"""AWS Configuration Schema"""
region: str = "us-east-1"
access_key_id: Optional[str] = None
secret_access_key: Optional[str] = None
use_iam_role: bool = True
s3_bucket: Optional[str] = None
2. Dictionary Schema¶
For more flexibility, you can use dictionary-based schemas:
Python
from kubiya_sdk import config_dict
config_dict(
name="database_config",
description="Database connection settings",
config_dict={
"host": {
"type": "string",
"description": "Database host",
"required": True,
},
"port": {
"type": "integer",
"description": "Database port",
"default": 5432,
},
"username": {
"type": "string",
"description": "Database username",
"required": True,
},
"password": {
"type": "string",
"description": "Database password",
"required": True,
},
"database": {
"type": "string",
"description": "Database name",
"required": True,
},
"ssl_mode": {
"type": "string",
"description": "SSL mode for connection",
"options": ["disable", "require", "verify-ca", "verify-full"],
"default": "require",
},
},
)
Using Configuration in Tools¶
Requiring Configuration¶
You can specify required and optional configurations for a tool:
Python
from kubiya_sdk.tools import function_tool
from kubiya_sdk import with_config
@function_tool(
name="list_s3_buckets",
description="List all S3 buckets",
required_configs=["aws_config"] # This tool requires aws_config
)
@with_config("aws_config")
def list_s3_buckets(config, limit: int = 10) -> list:
"""List S3 buckets using AWS configuration"""
# Access configuration values
region = config.get("region", "us-east-1")
access_key = config.get("access_key_id")
secret_key = config.get("secret_access_key")
# In a real tool, use these to configure boto3
# Here we just return an example
return [f"example-bucket-{i}-{region}" for i in range(limit)]
Optional Configuration¶
Tools can also specify optional configurations:
Python
@function_tool(
name="multi_cloud_upload",
description="Upload a file to multiple cloud providers",
required_configs=["aws_config"], # Required config
optional_configs=["azure_config", "gcp_config"] # Optional configs
)
@with_config("aws_config") # Primary config for the tool
def multi_cloud_upload(config, file_path: str, object_key: str) -> dict:
"""Upload a file to multiple cloud providers"""
# Get additional configurations if available
from kubiya_sdk import tool_registry
azure_config = tool_registry.get_config("azure_config")
gcp_config = tool_registry.get_config("gcp_config")
# Use configurations to implement your logic
result = {
"aws": f"Uploaded to AWS S3 in {config.get('region', 'us-east-1')}",
"providers_used": ["aws"]
}
if azure_config:
result["azure"] = "Uploaded to Azure Blob Storage"
result["providers_used"].append("azure")
if gcp_config:
result["gcp"] = "Uploaded to Google Cloud Storage"
result["providers_used"].append("gcp")
return result
Setting Configuration Values¶
Configuration values can be set in several ways:
1. Using the Tool Registry¶
Python
from kubiya_sdk import tool_registry
# Set the dynamic configuration
tool_registry.set_dynamic_config({
"aws_config": {
"region": "us-west-2",
"access_key_id": "EXAMPLE_KEY_ID",
"secret_access_key": "EXAMPLE_SECRET_KEY",
"s3_bucket": "example-bucket",
},
"database_config": {
"host": "db.example.com",
"port": 5432,
"username": "dbuser",
"password": "dbpassword",
"database": "exampledb",
}
})
2. Using Environment Variables¶
Configuration can also be set using environment variables:
Bash
# Set AWS configuration
export KUBIYA_CONFIG_aws_config='{"region":"us-west-2","access_key_id":"EXAMPLE_KEY_ID","secret_access_key":"EXAMPLE_SECRET_KEY"}'
# Set database configuration
export KUBIYA_CONFIG_database_config='{"host":"db.example.com","port":5432,"username":"dbuser","password":"dbpassword","database":"exampledb"}'
3. Using Configuration Files¶
You can store configuration in a JSON or YAML file:
JSON
{
"aws_config": {
"region": "us-west-2",
"access_key_id": "EXAMPLE_KEY_ID",
"secret_access_key": "EXAMPLE_SECRET_KEY",
"s3_bucket": "example-bucket"
},
"database_config": {
"host": "db.example.com",
"port": 5432,
"username": "dbuser",
"password": "dbpassword",
"database": "exampledb"
}
}
And load it in your code:
Python
import json
from kubiya_sdk import tool_registry
# Load configuration from a file
with open("config.json", "r") as f:
config_data = json.load(f)
# Set the dynamic configuration
tool_registry.set_dynamic_config(config_data)
Real-World Example¶
Here's a complete example of using dynamic configuration with a database tool:
Python
from kubiya_sdk import config_model, with_config
from kubiya_sdk.tools import function_tool
from pydantic import BaseModel
from typing import List, Dict, Any
# Define database configuration
@config_model(name="postgres_config", description="PostgreSQL Configuration")
class PostgresConfig(BaseModel):
"""PostgreSQL Configuration Schema"""
host: str
port: int = 5432
username: str
password: str
database: str
ssl_mode: str = "require"
# Create a tool that uses this configuration
@function_tool(
name="query_database",
description="Run a SQL query against a PostgreSQL database",
required_configs=["postgres_config"],
requirements=["psycopg2-binary"]
)
@with_config("postgres_config")
def query_database(config, query: str) -> List[Dict[str, Any]]:
"""Run a query against a PostgreSQL database"""
import psycopg2
import psycopg2.extras
# Use configuration to connect to the database
conn = psycopg2.connect(
host=config.get("host"),
port=config.get("port", 5432),
user=config.get("username"),
password=config.get("password"),
dbname=config.get("database"),
sslmode=config.get("ssl_mode", "require")
)
# Execute query and return results
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(query)
results = cur.fetchall()
return [dict(row) for row in results]
# Usage example
if __name__ == "__main__":
from kubiya_sdk import tool_registry
# Set configuration
tool_registry.set_dynamic_config({
"postgres_config": {
"host": "db.example.com",
"port": 5432,
"username": "dbuser",
"password": "dbpassword",
"database": "exampledb"
}
})
# Use the tool
results = query_database("SELECT * FROM users LIMIT 10")
print(f"Found {len(results)} users")
This example is directly based on the implementation in examples/dynamic_config.py in the SDK.