SDK Examples

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"])

from kubiya import ControlPlaneClient

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

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

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

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']}")

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 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', [])]}")

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

# 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 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 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}")

# 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']}")

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

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

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

Found 2 runtimes
  - Agno Runtime
  - Claude Code SDK

# 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']}")

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

# 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'))}")

Found 1 integrations
  - aws

# 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}")

# 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')}")

# 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 context graph statistics
try:
    stats = client.graph.get_stats()
    print(f"Graph Statistics: {stats}")
except Exception as e:
    print(f"Error: {e}")

# 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}")

# 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 updates
for update in client.task_planning.plan_stream(
    task_description="Set up monitoring for the payment service"
):
    print(update)

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()

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()

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()

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)

# 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)}")

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