Dynamic Configuration¶

Dynamic configuration in Kubiya SDK allows tools to access runtime configuration values, such as API keys, service endpoints, and other settings that may change between environments or executions.

Overview¶

Dynamic configuration provides several benefits:

Security: Sensitive values like API keys never need to be hardcoded
Environment Management: Different configurations for dev, staging, production
Sharing: Tools can be shared without exposing sensitive configuration
Runtime Updates: Configuration can be updated without changing code

Defining Configuration Models¶

Kubiya uses Pydantic models to define configuration schemas:

Python

from kubiya_sdk import config_model
from pydantic import BaseModel

@config_model(name="database_config", description="Database connection configuration")
class DatabaseConfig(BaseModel):
    """Configuration for database connections"""
    host: str
    port: int = 5432
    username: str
    password: str
    database: str
    ssl: bool = True

Using Configuration in Tools¶

Once you've defined a configuration model, you can use it in your tools:

Python

from kubiya_sdk import kubiya
from kubiya_sdk.tools.registry import tool_registry

@kubiya.tool(
    name="database-query",
    description="Execute a database query",
    image="python:3.12-slim",
    requirements=["psycopg2-binary"],
    required_configs=["database_config"]
)
def execute_query(query: str, config=None) -> dict:
    """Execute a SQL query against a database"""
    import psycopg2
    import psycopg2.extras

    if not config:
        raise ValueError("Database configuration is required")

    conn_string = f"host={config['host']} port={config['port']} user={config['username']} password={config['password']} dbname={config['database']}"
    if config.get('ssl', True):
        conn_string += " sslmode=require"

    conn = psycopg2.connect(conn_string)
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

    try:
        cursor.execute(query)
        if query.strip().upper().startswith("SELECT"):
            results = cursor.fetchall()
            return {"rows": results, "count": len(results)}
        else:
            conn.commit()
            return {"affected_rows": cursor.rowcount}
    finally:
        cursor.close()
        conn.close()

Setting Configuration Values¶

Configuration values can be set using the tool registry:

Python

from kubiya_sdk.tools.registry import tool_registry

# Set configuration values
tool_registry.set_dynamic_config({
    "database_config": {
        "host": "db.example.com",
        "port": 5432,
        "username": "app_user",
        "password": "secr3t_p@ssw0rd",
        "database": "application_db",
        "ssl": True
    }
})

# Now any tool that requires "database_config" will receive these values

Configuration Inheritance¶

Configuration can be hierarchical, allowing for shared settings:

Python

from kubiya_sdk import config_model
from pydantic import BaseModel

@config_model(name="aws_base_config", description="Base AWS configuration")
class AWSBaseConfig(BaseModel):
    """Base configuration for AWS services"""
    region: str = "us-east-1"
    access_key_id: str
    secret_access_key: str

@config_model(name="s3_config", description="S3 configuration")
class S3Config(AWSBaseConfig):
    """Configuration for S3 bucket operations"""
    bucket: str
    prefix: str = ""

Environment Variables¶

Configuration values can also be loaded from environment variables:

Python

import os
from kubiya_sdk.tools.registry import tool_registry

# Load configuration from environment variables
tool_registry.set_dynamic_config({
    "api_config": {
        "url": os.environ.get("API_URL", "https://api.example.com"),
        "api_key": os.environ.get("API_KEY", ""),
        "timeout": int(os.environ.get("API_TIMEOUT", "30"))
    }
})

Configuration Management Best Practices¶

Never hardcode sensitive values - Always use dynamic configuration for secrets
Use descriptive names - Configuration models should have clear, descriptive names
Set sensible defaults - Provide default values where appropriate
Validate configuration - Use Pydantic validators to ensure configuration is valid
Separate configurations - Create separate configurations for different services
Document requirements - Clearly document required configurations

Example: Multi-Service Configuration¶

Python

from kubiya_sdk import config_model, kubiya
from pydantic import BaseModel, field_validator
import re

@config_model(name="slack_config", description="Slack API configuration")
class SlackConfig(BaseModel):
    """Configuration for Slack integration"""
    api_token: str
    channel: str

    @field_validator('api_token')
    def validate_token(cls, v):
        if not v.startswith('xoxb-'):
            raise ValueError('Slack API token must start with xoxb-')
        return v

@config_model(name="jira_config", description="Jira API configuration")
class JiraConfig(BaseModel):
    """Configuration for Jira integration"""
    url: str
    username: str
    api_token: str
    project_key: str

    @field_validator('url')
    def validate_url(cls, v):
        if not re.match(r'^https?://.*\.atlassian\.net/.*$', v):
            raise ValueError('Jira URL must be a valid Atlassian domain')
        return v

@kubiya.tool(
    name="create-issue-and-notify",
    description="Create a Jira issue and notify in Slack",
    image="python:3.12-slim",
    requirements=["requests", "slack-sdk", "jira"],
    required_configs=["jira_config", "slack_config"]
)
def create_issue_and_notify(title: str, description: str, config=None) -> dict:
    """Create a Jira issue and send a notification to Slack"""
    from jira import JIRA
    from slack_sdk import WebClient
    from slack_sdk.errors import SlackApiError

    if not config:
        raise ValueError("Configuration is required")

    # Extract configurations
    jira_config = config.get("jira_config", {})
    slack_config = config.get("slack_config", {})

    # Create Jira issue
    jira = JIRA(
        server=jira_config["url"],
        basic_auth=(jira_config["username"], jira_config["api_token"])
    )

    issue_dict = {
        'project': {'key': jira_config["project_key"]},
        'summary': title,
        'description': description,
        'issuetype': {'name': 'Task'},
    }

    issue = jira.create_issue(fields=issue_dict)

    # Send Slack notification
    client = WebClient(token=slack_config["api_token"])
    try:
        response = client.chat_postMessage(
            channel=slack_config["channel"],
            text=f"New issue created: {issue.key} - {title}\n{jira_config['url']}/browse/{issue.key}"
        )
        message_ts = response["ts"]
    except SlackApiError as e:
        return {"error": f"Error sending slack message: {e.response['error']}"}

    return {
        "issue_key": issue.key,
        "issue_url": f"{jira_config['url']}/browse/{issue.key}",
        "slack_message_ts": message_ts
    }