SDK Examples - Kubiya

Learn by example with these real-world SDK usage patterns for the Control Plane Client.

Getting Started

Basic Client Setup

from kubiya import ControlPlaneClient

# Initialize with explicit API key
client = ControlPlaneClient(api_key="your-api-key")

# Or use environment variable (recommended)
import os
os.environ["KUBIYA_API_KEY"] = "your-api-key"
client = ControlPlaneClient(api_key=os.environ["KUBIYA_API_KEY"])

Health Check

Verify connectivity and service status:

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Check health
status = client.health.check()
print(status)

Output:

{
  "status": "healthy",
  "service": "agent-control-plane",
  "timestamp": "2026-02-05T03:18:20.915580",
  "services": {
    "kubiya_api": "healthy",
    "context_graph": "healthy",
    "cognitive_memory": "healthy"
  }
}

Agent Management

List Agents

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# List all agents
agents = client.agents.list(limit=10)
print(f"Found {len(agents)} agents")

for agent in agents:
    print(f"  - {agent['name']} (ID: {agent['id']})")
    print(f"    Model: {agent['model_id']}")
    print(f"    Status: {agent['status']}")

Output:

Found 4 agents
  - DevOps Assistant (ID: 039f75ea-b90b-46b5-97c8-ed5475468f3c)
    Model: claude-sonnet-4
    Status: running
  - HauKI (ID: ...)
    Model: claude-sonnet-4
    Status: running

Get Agent Details

# Get specific agent
agent = client.agents.get("039f75ea-b90b-46b5-97c8-ed5475468f3c")

print(f"Agent: {agent['name']}")
print(f"Description: {agent['description']}")
print(f"Model: {agent['model_id']}")
print(f"Runtime: {agent['runtime']}")
print(f"Skills: {[s['name'] for s in agent.get('skills', [])]}")

Output:

Agent: DevOps Assistant
Description: Development and operation of software products
Model: claude-sonnet-4
Runtime: default
Skills: ['Shell - Full Access']

Create an Agent

# Create a new agent
agent_data = {
    "name": "Infrastructure Bot",
    "description": "Manages cloud infrastructure",
    "model_id": "claude-sonnet-4",
    "system_prompt": "You are an infrastructure management assistant."
}

new_agent = client.agents.create(agent_data)
print(f"Created agent: {new_agent['name']} (ID: {new_agent['id']})")

Update an Agent

# Update agent configuration
update_data = {
    "description": "Updated description",
    "skill_ids": ["780af43c-fa4f-43da-9aff-74dc8244b967"]
}

updated = client.agents.update(agent_id, update_data)
print(f"Updated agent: {updated['name']}")

Execute an Agent

# Execute an agent task
execution_data = {
    "worker_queue_id": "default",
    "prompt": "List all running pods in the production namespace"
}

result = client.agents.execute(agent_id, execution_data)
print(f"Execution started: {result}")

Models and Runtimes

List Available Models

# Get all available LLM models
models = client.models.list()
print(f"Found {len(models)} models")

for model in models[:10]:
    print(f"  - {model['label']} ({model['id']})")
    print(f"    Provider: {model['provider']}")
    print(f"    Enabled: {model['enabled']}")

Output:

Found 56 models
  - GPT 4 (gpt-4)
    Provider: OpenAI
    Enabled: True
  - Claude Sonnet 4 (claude-sonnet-4)
    Provider: Anthropic
    Enabled: True
  ...

List Runtimes

# Get available runtimes
runtimes = client.runtimes.list()
print(f"Found {len(runtimes)} runtimes")

for runtime in runtimes:
    print(f"  - {runtime['name']}")

Output:

Found 2 runtimes
  - Agno Runtime
  - Claude Code SDK

Skills Management

List Skills

# Get all available skills
skills = client.skills.list()
print(f"Found {len(skills)} skills")

for skill in skills:
    print(f"  - {skill['name']} ({skill['type']})")
    print(f"    Description: {skill['description']}")
    print(f"    Enabled: {skill['enabled']}")

Output:

Found 1 skills
  - Shell - Full Access (shell)
    Description: Complete ability to use the terminal
    Enabled: True

Integrations

List Integrations

# Get all active integrations
integrations = client.integrations.list()
print(f"Found {len(integrations)} integrations")

for integration in integrations:
    print(f"  - {integration.get('name', integration.get('vendor', 'N/A'))}")

Output:

Found 1 integrations
  - aws

Get Integration Credentials

# Get credentials for an integration
try:
    creds = client.integrations.get_integration_credentials(
        vendor="aws",
        id="aws"
    )
    print(f"Got AWS credentials")
except Exception as e:
    print(f"Error: {e}")

Workers

List Workers

# Get all workers
workers = client.workers.list()
print(f"Found {len(workers)} workers")

for worker in workers:
    print(f"  - Queue: {worker.get('queue_id', 'N/A')}")
    print(f"    Status: {worker.get('status', 'N/A')}")

Context Graph

Query the Context Graph

# List nodes in the context graph
try:
    nodes = client.graph.list_nodes(limit=10)
    print(f"Total nodes: {nodes.get('count', 0)}")

    for node in nodes.get('nodes', []):
        name = node.get('properties', {}).get('name', 'N/A')
        labels = node.get('labels', [])
        print(f"  - {name} (labels: {labels})")
except Exception as e:
    print(f"Error: {e}")

Get Graph Statistics

# Get context graph statistics
try:
    stats = client.graph.get_stats()
    print(f"Graph Statistics: {stats}")
except Exception as e:
    print(f"Error: {e}")

Search Nodes

# Search nodes by text
try:
    results = client.graph.search_nodes_by_text({
        "property_name": "name",
        "search_text": "production"
    })
    print(f"Found {len(results.get('nodes', []))} matching nodes")
except Exception as e:
    print(f"Error: {e}")

Task Planning

Generate a Task Plan

# Use AI to plan a task
plan = client.task_planning.plan(
    task_description="Deploy a new microservice to production",
    context={"environment": "production", "service": "user-api"}
)
print(f"Generated plan: {plan}")

Stream Task Planning

# Stream task planning updates
for update in client.task_planning.plan_stream(
    task_description="Set up monitoring for the payment service"
):
    print(update)

Complete Workflow Example

DevOps Automation Script

from kubiya import ControlPlaneClient
import json

def devops_automation():
    """Complete DevOps automation example using Control Plane SDK"""

    client = ControlPlaneClient(api_key="your-api-key")

    # 1. Verify service health
    print("=== Checking Service Health ===")
    health = client.health.check()
    if health['status'] != 'healthy':
        print("Warning: Service not healthy!")
        return
    print(f"Status: {health['status']}")

    # 2. List available agents
    print("\n=== Available Agents ===")
    agents = client.agents.list()
    for agent in agents:
        print(f"  - {agent['name']} ({agent['status']})")

    # 3. Find a DevOps agent
    devops_agent = next(
        (a for a in agents if 'devops' in a['name'].lower()),
        None
    )

    if not devops_agent:
        print("No DevOps agent found, creating one...")
        devops_agent = client.agents.create({
            "name": "DevOps Bot",
            "description": "Automated DevOps operations",
            "model_id": "claude-sonnet-4"
        })

    print(f"\n=== Using Agent: {devops_agent['name']} ===")

    # 4. Check available models
    print("\n=== Available Models ===")
    models = client.models.list()
    claude_models = [m for m in models if 'claude' in m.get('id', '').lower()]
    for model in claude_models[:3]:
        print(f"  - {model['label']}")

    # 5. Check runtimes
    print("\n=== Available Runtimes ===")
    runtimes = client.runtimes.list()
    for runtime in runtimes:
        print(f"  - {runtime['name']}")

    # 6. List integrations
    print("\n=== Active Integrations ===")
    integrations = client.integrations.list()
    for integration in integrations:
        print(f"  - {integration.get('vendor', 'Unknown')}")

    print("\n=== DevOps Automation Complete ===")

if __name__ == "__main__":
    devops_automation()

Infrastructure Management Script

from kubiya import ControlPlaneClient

def manage_infrastructure():
    """Infrastructure management with Control Plane SDK"""

    client = ControlPlaneClient(api_key="your-api-key")

    # List all agents with their skills
    print("=== Agents and Their Capabilities ===")
    agents = client.agents.list()

    for agent in agents:
        print(f"\nAgent: {agent['name']}")
        print(f"  Model: {agent['model_id']}")
        print(f"  Runtime: {agent.get('runtime', 'default')}")

        skills = agent.get('skills', [])
        if skills:
            print("  Skills:")
            for skill in skills:
                print(f"    - {skill['name']} ({skill['type']})")

    # Find agents with specific skills
    print("\n=== Agents with Shell Access ===")
    shell_agents = [
        a for a in agents
        if any(s['type'] == 'shell' for s in a.get('skills', []))
    ]

    for agent in shell_agents:
        print(f"  - {agent['name']}")

if __name__ == "__main__":
    manage_infrastructure()

Error Handling

Robust Error Handling Example

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import (
    ControlPlaneError,
    AgentError,
    GraphError
)

def safe_operations():
    """Example with proper error handling"""

    client = ControlPlaneClient(api_key="your-api-key")

    # Safe agent listing
    try:
        agents = client.agents.list()
        print(f"Found {len(agents)} agents")
    except AgentError as e:
        print(f"Agent operation failed: {e}")
    except ControlPlaneError as e:
        print(f"Control plane error: {e}")

    # Safe graph query
    try:
        nodes = client.graph.list_nodes(limit=10)
        print(f"Found {nodes.get('count', 0)} nodes")
    except GraphError as e:
        print(f"Graph query failed: {e}")
    except ControlPlaneError as e:
        print(f"Control plane error: {e}")

    # General exception handling
    try:
        # Any operation
        result = client.health.check()
    except Exception as e:
        print(f"Unexpected error: {e}")

if __name__ == "__main__":
    safe_operations()

Best Practices

Environment Configuration

import os
from kubiya import ControlPlaneClient

# Best practice: Use environment variables
API_KEY = os.environ.get("KUBIYA_API_KEY")
if not API_KEY:
    raise ValueError("KUBIYA_API_KEY environment variable not set")

client = ControlPlaneClient(api_key=API_KEY)

Pagination

# Handle large result sets with pagination
def get_all_agents(client, page_size=100):
    """Get all agents using pagination"""
    all_agents = []
    skip = 0

    while True:
        agents = client.agents.list(skip=skip, limit=page_size)
        if not agents:
            break
        all_agents.extend(agents)
        skip += page_size

    return all_agents

agents = get_all_agents(client)
print(f"Total agents: {len(agents)}")

Caching Results

from functools import lru_cache
import time

class CachedClient:
    def __init__(self, api_key):
        self.client = ControlPlaneClient(api_key=api_key)
        self._models_cache = None
        self._models_cache_time = 0
        self._cache_ttl = 300  # 5 minutes

    def get_models(self):
        """Get models with caching"""
        now = time.time()
        if self._models_cache is None or (now - self._models_cache_time) > self._cache_ttl:
            self._models_cache = self.client.models.list()
            self._models_cache_time = now
        return self._models_cache

cached = CachedClient(api_key="your-api-key")
models = cached.get_models()  # Cached after first call

Next Steps

Control Plane Client

Deep dive into the Control Plane Client

Context Graph

Query entities and relationships

API Reference

Complete API documentation

Installation

Get started with the SDK

​Getting Started

​Basic Client Setup

​Health Check

​Agent Management

​List Agents

​Get Agent Details

​Create an Agent

​Update an Agent

​Execute an Agent

​Models and Runtimes

​List Available Models

​List Runtimes

​Skills Management

​List Skills

​Integrations

​List Integrations

​Get Integration Credentials

​Workers

​List Workers

​Context Graph

​Query the Context Graph

​Get Graph Statistics

​Search Nodes

​Task Planning

​Generate a Task Plan

​Stream Task Planning

​Complete Workflow Example

​DevOps Automation Script

​Infrastructure Management Script

​Error Handling

​Robust Error Handling Example

​Best Practices

​Environment Configuration

​Pagination

​Caching Results

​Next Steps

Control Plane Client

Context Graph

API Reference

Installation

Getting Started

Basic Client Setup

Health Check

Agent Management

List Agents

Get Agent Details

Create an Agent

Update an Agent

Execute an Agent

Models and Runtimes

List Available Models

List Runtimes

Skills Management

List Skills

Integrations

List Integrations

Get Integration Credentials

Workers

List Workers

Context Graph

Query the Context Graph

Get Graph Statistics

Search Nodes

Task Planning

Generate a Task Plan

Stream Task Planning

Complete Workflow Example

DevOps Automation Script

Infrastructure Management Script

Error Handling

Robust Error Handling Example

Best Practices

Environment Configuration

Pagination

Caching Results

Next Steps