Skip to content

Tool Building

Tools are the fundamental building blocks in the Kubiya ecosystem. This guide explains how to build effective Docker-based tools that can be consumed by teammates (AI agents) and other LLM applications.

What Makes Kubiya Tools Special

Kubiya tools are:

  1. Docker-Backed: Every tool is powered by a Docker image
  2. Stateless: Built for reliable, repeatable execution
  3. Consumable: Can be used by various AI applications
  4. Infrastructure-Agnostic: Run anywhere Docker is supported

Basic Tool Structure

A Kubiya tool consists of:

  • A function decorated with @tool
  • Docker image specification
  • Input/output schemas
  • Optional configuration requirements

Here's a simple example:

Python
from kubiya_sdk import tool

@tool(image="python:3.12-slim")
def capitalize_text(text: str) -> str:
    """Capitalize the input text"""
    return text.upper()

Leveraging Existing Docker Images

The power of Kubiya comes from using existing Docker images rather than writing everything from scratch:

Python
from kubiya_sdk import tool

@tool(
    image="aquasec/trivy:latest",
    command=["image", "--format", "json", "${IMAGE_TO_SCAN}"],
    environment={
        "IMAGE_TO_SCAN": "${image_name}"  # Maps to input parameter
    }
)
def scan_container_image(image_name: str) -> dict:
    """Scan a container image for vulnerabilities"""
    # No code needed - execution happens directly in the container
    pass

Including Code in Your Tools

For more complex behavior, you can implement your logic within the tool function:

Python
from kubiya_sdk import tool

@tool(
    image="python:3.12-slim",
    requirements=["requests", "beautifulsoup4"]
)
def web_scraper(url: str, selector: str = "body") -> dict:
    """Scrape content from a web page"""
    import requests
    from bs4 import BeautifulSoup

    # Fetch the webpage
    response = requests.get(url)
    response.raise_for_status()

    # Parse the HTML
    soup = BeautifulSoup(response.text, "html.parser")

    # Extract content based on the selector
    elements = soup.select(selector)

    # Return results
    return {
        "url": url,
        "selector": selector,
        "count": len(elements),
        "content": [element.text.strip() for element in elements]
    }

Adding Input Validation with Type Hints

Kubiya uses Python type hints to validate inputs:

Python
from typing import List, Dict, Optional
from kubiya_sdk import tool

@tool(image="python:3.12-slim")
def process_data(
    items: List[Dict[str, str]],
    operation: str = "summarize",
    limit: Optional[int] = None
) -> Dict[str, any]:
    """Process a list of items"""
    if limit:
        items = items[:limit]

    if operation == "summarize":
        return {"count": len(items), "fields": list(items[0].keys()) if items else []}
    elif operation == "extract":
        return {"items": items}
    else:
        raise ValueError(f"Unknown operation: {operation}")

Using Configuration Values

For API keys and other configuration values, use dynamic configuration:

Python
from kubiya_sdk import tool, config_model
from pydantic import BaseModel

@config_model(name="github_config", description="GitHub API configuration")
class GitHubConfig(BaseModel):
    api_key: str
    organization: str = ""

@tool(
    image="python:3.12-slim",
    requirements=["requests"],
    required_configs=["github_config"]
)
def list_repositories(config=None) -> dict:
    """List repositories for the configured GitHub organization"""
    import requests

    if not config or "github_config" not in config:
        raise ValueError("GitHub configuration is required")

    github_config = config["github_config"]
    api_key = github_config["api_key"]
    org = github_config["organization"]

    headers = {
        "Authorization": f"token {api_key}",
        "Accept": "application/vnd.github.v3+json"
    }

    url = f"https://api.github.com/orgs/{org}/repos" if org else "https://api.github.com/user/repos"

    response = requests.get(url, headers=headers)
    response.raise_for_status()

    repos = response.json()
    return {
        "count": len(repos),
        "repositories": [{"name": repo["name"], "url": repo["html_url"]} for repo in repos]
    }

Running Tools in Different Environments

Kubiya tools can run in various environments:

Python
from kubiya_sdk import tool

# Define the tool as usual
@tool(image="python:3.12-slim")
def simple_tool(param: str) -> str:
    """A simple tool that runs in different environments"""
    return f"Processed: {param}"

# Set runtime environment options in your execution logic
if use_kubernetes:
    from kubiya_sdk.execution import execute_tool_in_kubernetes

    result = execute_tool_in_kubernetes(
        "simple_tool", 
        {"param": "test"},
        namespace="kubiya-tools",
        service_account="tool-runner"
    )
else:
    # Run locally
    result = simple_tool("test")

Including Files in Your Tools

You can bundle files with your tools:

Python
from kubiya_sdk import tool, FileSpec

@tool(
    image="python:3.12-slim",
    with_files=[
        FileSpec(
            destination="/app/config.json",
            content="""
{
    "settings": {
        "max_results": 100,
        "timeout": 30
    }
}
"""
        ),
        FileSpec(
            destination="/app/helper.py",
            content="""
def process_data(data):
    return {k.upper(): v for k, v in data.items()}
"""
        )
    ]
)
def tool_with_files(input_data: dict) -> dict:
    """Tool that uses bundled files"""
    import json
    import sys

    # Load the config file
    with open("/app/config.json", "r") as f:
        config = json.load(f)

    # Import the helper module
    sys.path.append("/app")
    from helper import process_data

    # Process the data using the helper
    processed = process_data(input_data)

    # Apply config settings
    max_results = config["settings"]["max_results"]

    # Return the result
    return {
        "processed": processed,
        "config": config["settings"],
        "limited": dict(list(processed.items())[:max_results])
    }

Next Steps