​
Tools Overview

Kubiya tools are the fundamental building blocks of workflows, enabling both AI agents and developers to perform specific tasks and operations. Every tool runs in its own isolated environment, providing deterministic and reliable execution.

​
What Are Tools?

Tools in Kubiya are containerized, reusable components that:
  • Execute specific functions or operations in isolation
  • Accept typed parameters and return structured results
  • Can be chained together in complex workflows
  • Are discoverable and executable by AI agents
  • Support authentication, secrets, and environment configuration

​
Installation

Install the tools functionality with specific extras: 
# Install with tools support
pip install kubiya-workflow-sdk[tools]

# Install with all features (recommended)
pip install kubiya-workflow-sdk[all]

# For development with all tools features
pip install kubiya-workflow-sdk[dev,tools]

​
Quick Start

Get started with tools in 5 minutes: 
# 1. Install the SDK with tools support
# pip install kubiya-workflow-sdk[tools]

# 2. Create your first tool using the function_tool decorator
from kubiya_workflow_sdk.tools import function_tool

@function_tool(
    description="Simple greeting tool",
    requirements=[]
)
def greet(name: "str", greeting: "str" = "Hello") -> "str":
    """Generate a personalized greeting."""
    return f"{greeting}, {name}!"

# 3. The tool is automatically registered and ready to use
from kubiya_workflow_sdk.tools import tool_registry

# List registered tools
tools = tool_registry.list_tools("default")
print(f"Available tools: {[tool.name for tool in tools]}")

# Get the tool
greeting_tool = tool_registry.get_tool("default", "greet")
if greeting_tool:
    print(f"Tool description: {greeting_tool.description}")
else:
    print("Tool not found")

​
Tool Types

​
Function Tools

Create tools from Python functions using the @function_tool decorator: 
from kubiya_workflow_sdk.tools import function_tool

# Simple function tool
@function_tool(
    description="Calculate mathematical operations",
    requirements=["numpy"]
)
def calculate(operation: "str", x: "float", y: "float") -> "float":
    """Perform mathematical calculations."""
    import numpy as np
    
    if operation == "add":
        return x + y
    elif operation == "multiply":
        return x * y
    elif operation == "power":
        return np.power(x, y)
    else:
        raise ValueError(f"Unknown operation: {operation}")

# Advanced function tool with environment variables
@function_tool(
    description="Process CSV data and return statistics",
    requirements=["pandas>=1.3.0", "numpy"],
    env=["DATA_PATH"],
    secrets=["api_key"]
)
def process_csv(file_path: "str", delimiter: "str" = ",") -> "dict":
    """Process a CSV file and return basic statistics."""
    import pandas as pd
    import os
    
    # Use environment variables
    base_path = os.environ.get("DATA_PATH", "/tmp")
    full_path = os.path.join(base_path, file_path)
    
    df = pd.read_csv(full_path, delimiter=delimiter)
    return {
        "rows": len(df),
        "columns": len(df.columns),
        "memory_usage": df.memory_usage(deep=True).sum(),
        "file_processed": full_path
    }

​
Docker Tools

Create containerized tools that run in Docker with full system access: 
from kubiya_workflow_sdk.tools import Tool, Arg

# Basic Docker tool
kubectl_tool = Tool(
    name="kubectl_manager",
    type="docker",
    image="bitnami/kubectl:latest",
    description="Kubernetes cluster management tool",
    args=[
        Arg(
            name="command",
            description="Kubectl command to execute",
            required=True
        ),
        Arg(
            name="namespace",
            description="Kubernetes namespace",
            default="default"
        )
    ],
    env=["KUBECONFIG"],
    secrets=["k8s_token"],
    content="""
#!/bin/bash
kubectl --namespace={{ .namespace }} {{ .command }}
    """
)

# Import Volume and FileSpec for the example
from kubiya_workflow_sdk.tools import Volume, FileSpec

# Advanced Docker tool with file mounts and services
database_tool = Tool(
    name="postgres_backup",
    type="docker",
    image="postgres:13",
    description="Create and manage database backups",
    args=[
        Arg(name="action", description="Action: backup, restore, list", required=True),
        Arg(name="database", description="Database name", required=True),
        Arg(name="backup_name", description="Backup file name", default="")
    ],
    env=["POSTGRES_USER", "POSTGRES_PASSWORD", "POSTGRES_HOST"],
    with_volumes=[
        Volume(name="/host/backups", path="/backups")
    ],
    with_files=[
        FileSpec(
            destination="/backup-script.sh",
            content="""#!/bin/bash
case $1 in
  backup)
    pg_dump -h $POSTGRES_HOST -U $POSTGRES_USER $2 > /backups/$3.sql
    gzip /backups/$3.sql
    ;;
  restore)
    gunzip -c /backups/$3.sql.gz | psql -h $POSTGRES_HOST -U $POSTGRES_USER $2
    ;;
  list)
    ls -la /backups/
    ;;
esac
            """
        )
    ],
    content="""
chmod +x /backup-script.sh
/backup-script.sh {{ .action }} {{ .database }} {{ .backup_name }}
    """
)

​
Python Tools

Create native Python tools using the Tool class directly: 
from kubiya_workflow_sdk.tools import Tool, Arg, FunctionTool, Source

# Using FunctionTool class
def data_processor(input_data: str, format_type: str = "json") -> dict:
    """Process input data and return formatted output.
    
    Args:
        input_data: JSON string containing the data to process
        format_type: Output format type
    """
    import json
    
    # Parse the JSON string to get the dictionary
    data_dict = json.loads(input_data)
    
    processed = {
        "original_keys": list(data_dict.keys()),
        "processed_at": "2024-01-01T00:00:00Z",
        "format": format_type,
        "item_count": len(data_dict)
    }
    return processed

# Create tool from function with explicit args
processor_tool = Tool(
    name="data_processor",
    description="Process and format data",
    type="python",
    args=[
        Arg(name="input_data", type="str", description="JSON string containing the data to process", required=True),
        Arg(name="format_type", type="str", description="Output format type", required=False, default="json")
    ],
    source=Source(id="local"),
    function=data_processor
)

# Register the tool
from kubiya_workflow_sdk.tools import tool_registry
tool_registry.register(processor_tool)

​
Tool Registration & Discovery

Tools are automatically registered when using decorators, or can be registered manually: 
from kubiya_workflow_sdk.tools import tool_registry, function_tool, Tool, Arg

# Automatic registration with decorators
@function_tool(description="Auto-registered tool")
def my_function(x: "int") -> "int":
    return x * 2

# Manual registration
manual_tool = Tool(
    name="manual_tool",
    type="docker",
    image="alpine:latest",
    description="Manually registered tool",
    args=[Arg(name="message", description="Message to echo")],
    content="echo '{{ .message }}'"
)
tool_registry.register(manual_tool)

# Discovery
tools = tool_registry.list_tools("default")
print(f"Available tools: {[tool.name for tool in tools]}")

# Get specific tool
my_tool = tool_registry.get_tool("default", "my_function")
if my_tool:
    print(f"Found tool: {my_tool.description}")

​
Tool Arguments & Validation

Define typed arguments with validation for your tools: 
from kubiya_workflow_sdk.tools import Tool, Arg

# Tool with comprehensive argument definitions
advanced_tool = Tool(
    name="data_analyzer",
    type="docker",
    image="python:3.11-slim",
    description="Analyze data with configurable parameters",
    args=[
        Arg(
            name="input_file",
            description="Path to input file",
            required=True
        ),
        Arg(
            name="threshold",
            type="float",
            description="Analysis threshold",
            default=0.5,
            required=False
        ),
        Arg(
            name="filters",
            type="array",
            description="List of filters to apply",
            default=[]
        ),
        Arg(
            name="enable_debug",
            type="bool",
            description="Enable debug mode",
            default=True,
            required=False
        ),
        Arg(
            name="max_records",
            type="int",
            description="Maximum records to process",
            default=1000
        )
    ],
    content="""
python -c "
import json
import sys

# Access arguments
input_file = '{{ .input_file }}'
threshold = {{ .threshold }}
filters = {{ .filters | toJson }}
debug = {{ .enable_debug }}
max_records = {{ .max_records }}

print(f'Processing {input_file} with threshold {threshold}')
if debug:
    print(f'Debug mode enabled, filters: {filters}')
    print(f'Max records: {max_records}')
"
    """
)

# Validate inputs before execution
inputs = {
    "input_file": "data.csv",
    "threshold": 0.8,
    "enable_debug": True
}

validated_inputs = advanced_tool.validate_inputs(inputs)
print(f"Validated inputs: {validated_inputs}")

​
Environment Configuration

​
Environment Variables & Secrets

Tools can access environment variables and secrets for configuration: 
# Function tool with environment access
from kubiya_workflow_sdk.tools import function_tool, Tool, Arg

@function_tool(
    description="GitHub API client",
    env=["GITHUB_TOKEN", "GITHUB_ORG"],
    secrets=["github_webhook_secret"],
    requirements=["requests"]
)
def create_github_issue(title: "str", body: "str", repo: "str") -> "dict":
    """Create a GitHub issue using API."""
    import os
    import requests
    
    token = os.environ["GITHUB_TOKEN"]
    org = os.environ["GITHUB_ORG"]
    
    response = requests.post(
        f"https://api.github.com/repos/{org}/{repo}/issues",
        headers={"Authorization": f"token {token}"},
        json={"title": title, "body": body}
    )
    
    return response.json()

# Docker tool with environment configuration
api_tool = Tool(
    name="api_client",
    type="docker",
    image="curlimages/curl:latest",
    description="Make API calls with authentication",
    args=[
        Arg(name="endpoint", description="API endpoint", required=True),
        Arg(name="method", description="HTTP method", default="GET")
    ],
    env=["API_TOKEN", "API_BASE_URL"],
    secrets=["api_secret"],
    content="""
curl -X {{ .method }} \
  -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  "$API_BASE_URL/{{ .endpoint }}"
    """
)

​
Working with Tool Registry

The tool registry manages all tools in your application: 
from kubiya_workflow_sdk.tools import tool_registry, function_tool, Tool, Arg

# Check registered tools
print("Available tools:")
for tool in tool_registry.list_tools("default"):
    print(f"- {tool.name}: {tool.description}")

# Get specific tool
my_tool = tool_registry.get_tool("default", "greet")
if my_tool:
    print(f"Tool found: {my_tool.name}")
    print(f"Arguments: {[arg.name for arg in my_tool.args]}")

# Register tool manually
custom_tool = Tool(
    name="echo_tool",
    type="docker",
    image="alpine:latest",
    description="Echo a message",
    args=[Arg(name="message", description="Message to echo")],
    content="echo '{{ .message }}'"
)
tool_registry.register(custom_tool)

​
Type System & Validation

Supported argument types and validation: 
from kubiya_workflow_sdk.tools import Tool, Arg

tool = Tool(
    name="type_demo",
    description="Demonstrates all supported types",
    args=[
        # String type
        Arg(name="name", description="User name", required=True),
        
        # Numeric types
        Arg(name="age", type="int", description="Age in years", default=0),
        Arg(name="score", type="float", description="Score percentage", default=0.0),
        
        # Boolean type
        Arg(name="active", type="bool", description="Is active", default=True),
        
        # Array type
        Arg(name="tags", type="array", description="List of tags", default=[]),
        
        # Optional with choices
        Arg(name="level", description="Difficulty level", 
            default="medium", options=["easy", "medium", "hard"])
    ]
)

# Validate inputs
inputs = {
    "name": "John",
    "age": 25,
    "score": 95.5,
    "active": True,
    "tags": ["python", "tools"],
    "level": "hard"
}

validated = tool.validate_inputs(inputs)
print(f"Validation passed: {validated}")

​
Best Practices

​
Tool Design

  1. Single Purpose: Each tool should do one thing well
  2. Clear Documentation: Provide descriptive names and documentation
  3. Type Safety: Use proper argument types and validation
  4. Error Handling: Handle errors gracefully with clear messages
  5. Resource Management: Clean up resources properly

​
Performance Tips

  1. Efficient Images: Use minimal Docker base images
  2. Layer Caching: Structure Dockerfiles for optimal caching
  3. Resource Limits: Set appropriate CPU and memory limits
  4. Stateless Design: Keep tools stateless for better scalability

​
Security Guidelines

  1. Secrets Management: Use the secrets parameter for sensitive data
  2. Input Validation: Always validate and sanitize inputs
  3. Least Privilege: Run with minimal required permissions
  4. Network Isolation: Restrict network access when possible

​
Common Patterns

​
Data Processing Tool

from kubiya_workflow_sdk.tools import function_tool

@function_tool(
    description="Process data files",
    requirements=["pandas", "numpy"],
    env=["DATA_DIR"]
)
def process_data(file_path: "str", output_format: "str" = "csv") -> "dict":
    """Process data and return summary statistics."""
    import pandas as pd
    import os
    
    full_path = os.path.join(os.environ.get("DATA_DIR", "."), file_path)
    df = pd.read_csv(full_path)
    
    # Process data
    summary = {
        "rows": len(df),
        "columns": len(df.columns),
        "output_file": f"processed_{file_path}"
    }
    
    # Save processed data
    if output_format == "csv":
        df.to_csv(summary["output_file"], index=False)
    elif output_format == "json":
        df.to_json(summary["output_file"], orient="records")
    
    return summary

​
API Integration Tool

from kubiya_workflow_sdk.tools import function_tool

@function_tool(
    description="Slack notification sender",
    requirements=["requests"],
    secrets=["slack_webhook"]
)
def send_slack_message(channel: "str", message: "str", username: "str" = "Bot") -> "dict":
    """Send message to Slack channel."""
    import requests
    import os
    
    webhook_url = os.environ["SLACK_WEBHOOK"]
    
    payload = {
        "channel": channel,
        "username": username,
        "text": message
    }
    
    response = requests.post(webhook_url, json=payload)
    
    return {
        "status": "sent" if response.ok else "failed",
        "status_code": response.status_code,
        "message": message
    }