> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kubiya.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Control Plane Client SDK
> Python client for accessing Kubiya Control Plane services and APIs
The **Control Plane Client SDK** provides a comprehensive Python interface to Kubiya's Control Plane services, enabling you to programmatically manage models, runtimes, skills, policies, agents, workers, and the context graph.
The Control Plane Client (`ControlPlaneClient`) is the modern SDK for accessing Kubiya's Control Plane services. It provides service-oriented access to all platform capabilities including graph operations, cognitive memory, and infrastructure management.
## Key Features
* **π§ Context Graph Operations**: Intelligent search, semantic search, and graph queries
* **πΎ Cognitive Memory**: Store and recall context across interactions
* **π₯ Multi-Agent Teams**: Create and orchestrate collaborative agent teams
* **π Project Organization**: Organize agents and teams into structured projects
* **β° Job Scheduling**: Schedule recurring tasks with cron or webhook triggers
* **π Environment Management**: Manage isolated runtime environments
* **πΊοΈ Task Planning**: AI-powered task decomposition and planning
* **π Context Resolution**: Entity context with inheritance across layers
* **π€ Agent & Model Management**: Access and configure LLMs and agents
* **π οΈ Skills & Policies**: Configure tools, capabilities, and access controls
* **π Infrastructure Management**: Monitor workers and execution infrastructure
## Quick Start
```python theme={null}
from kubiya import ControlPlaneClient
# Initialize the client
client = ControlPlaneClient(api_key="your-api-key")
# Use intelligent search
result = client.graph.intelligent_search(
keywords="Find all production databases with high availability"
)
print(result["answer"]) # Natural language answer
for node in result["nodes"]:
print(f"- {node['properties']['name']}")
```
## Available Control Plane Services
The Control Plane Client provides access to these services:
| Service | Purpose | Key Methods |
| ------------------- | ------------------------------- | ------------------------------------------------------------------------ |
| **`health`** | Health checks | `check` |
| **`graph`** | Context Graph operations | `intelligent_search`, `semantic_search`, `store_memory`, `recall_memory` |
| **`datasets`** | Dataset management | `create_dataset`, `list_datasets`, `get_dataset`, `delete_dataset` |
| **`ingestion`** | Graph data ingestion | `ingest_node`, `ingest_nodes_batch`, `ingest_relationship` |
| **`models`** | LLM model management | `list`, `get`, `get_default` |
| **`runtimes`** | Agent runtime configs | `list`, `get`, `create` |
| **`context`** | Context resolution & management | `get`, `update`, `delete`, `resolve` |
| **`skills`** | Tool sets and capabilities | `list`, `get`, `create` |
| **`policies`** | Policy management | `list`, `get`, `create` |
| **`task_planning`** | AI-powered task planning | `plan`, `plan_stream` |
| **`agents`** | Agent management | `list`, `get`, `create`, `update`, `delete` |
| **`teams`** | Multi-agent teams | `list`, `get`, `create`, `update`, `delete`, `execute`, `execute_stream` |
| **`projects`** | Project organization | `list`, `get`, `create`, `update`, `delete`, `add_agent`, `add_team` |
| **`environments`** | Runtime environments | `list`, `get`, `create`, `update`, `delete`, `get_worker_command` |
| **`jobs`** | Scheduled & webhook jobs | `list`, `get`, `create`, `update`, `trigger`, `enable`, `disable` |
| **`workers`** | Worker management | `list`, `get` |
| **`secrets`** | Secrets management | `list`, `get`, `create` |
| **`integrations`** | Integration configs | `list`, `get` |
## Client Initialization
### Basic Initialization
```python theme={null}
from kubiya import ControlPlaneClient
# Basic initialization with API key
client = ControlPlaneClient(api_key="your-api-key")
```
### Using Environment Variables
The recommended approach for secure credential management:
```bash theme={null}
# Set environment variable
export KUBIYA_API_KEY="your-api-key"
```
```python theme={null}
from kubiya import ControlPlaneClient
# Reads KUBIYA_API_KEY from environment
client = ControlPlaneClient()
```
### Advanced Configuration
```python theme={null}
from kubiya import ControlPlaneClient
client = ControlPlaneClient(
api_key="your-api-key",
base_url="https://control-plane.kubiya.ai", # Control Plane API base URL (default)
timeout=300, # Request timeout in seconds (default: 300)
max_retries=3, # Number of retry attempts (default: 3)
org_name="my-organization" # Optional organization name
)
```
## Core Services
### Context Graph Service
Query and explore your knowledge graph with AI-powered search:
```python theme={null}
# AI-powered intelligent search
result = client.graph.intelligent_search(
keywords="Find all EC2 instances in us-east-1",
integration="AWS"
)
print(result["answer"])
for node in result["nodes"]:
print(f"- {node['properties']['name']}")
```
AI-powered graph search with Claude Agent
Vector-based natural language search
Store and recall context
Complete Context Graph documentation
### Dataset Management
Manage cognitive datasets for memory and knowledge:
```python theme={null}
# Create a dataset
dataset = client.datasets.create_dataset(
name="production-knowledge",
description="Production environment context",
scope="org"
)
# List datasets
datasets = client.datasets.list_datasets()
for ds in datasets:
print(f"Dataset: {ds['name']}")
```
[Learn more about Dataset Management β](/sdk/context-graph-datasets)
### Graph Ingestion
Import nodes and relationships into the context graph:
```python theme={null}
# Ingest nodes in batch
client.ingestion.ingest_nodes_batch(
nodes=[
{
"id": "server-1",
"labels": ["Server", "Production"],
"properties": {"name": "prod-server-01", "region": "us-east-1"}
},
{
"id": "db-1",
"labels": ["Database", "PostgreSQL"],
"properties": {"name": "prod-db-01", "size": "large"}
}
],
duplicate_handling="skip"
)
# Ingest relationships
client.ingestion.ingest_relationships_batch(
relationships=[
{
"source_id": "server-1",
"target_id": "db-1",
"relationship_type": "CONNECTS_TO",
"properties": {"port": 5432}
}
]
)
```
[Learn more about Graph Ingestion β](/sdk/context-graph-ingestion)
### Models Service
Manage LLM models and configurations:
```python theme={null}
# List available models
models = client.models.list_models()
for model in models:
print(f"Model: {model['name']} - Provider: {model['provider']}")
# Get model details
model = client.models.get_model("claude-3-opus-20240229")
print(f"Context window: {model['context_window']}")
```
[Learn more about Models Service β](/sdk/control-plane-models)
### Runtimes Service
Configure agent runtime environments:
```python theme={null}
# List agent runtimes
runtimes = client.runtimes.list_runtimes()
for runtime in runtimes:
print(f"Runtime: {runtime['name']}")
# Get runtime configuration
runtime = client.runtimes.get_runtime("agno-runtime")
print(f"Runtime config: {runtime}")
```
[Learn more about Runtimes Service β](/sdk/control-plane-runtimes)
### Skills Service
Manage tools and capabilities:
```python theme={null}
# List available skills
skills = client.skills.list_skills()
for skill in skills:
print(f"Skill: {skill['name']} - Category: {skill['category']}")
# Get skill details
skill = client.skills.get_skill("kubernetes")
print(f"Tools: {skill['tools']}")
```
[Learn more about Skills Service β](/sdk/control-plane-skills)
### Policies Service
Manage security and access policies:
```python theme={null}
# List policies
policies = client.policies.list_policies()
for policy in policies:
print(f"Policy: {policy['name']}")
# Get policy details
policy = client.policies.get_policy("security-policy-01")
print(f"Enforcement: {policy['enforcement']}")
```
[Learn more about Policies Service β](/sdk/control-plane-policies)
### Agents Service
Manage Control Plane agents:
```python theme={null}
# List agents
agents = client.agents.list_agents()
for agent in agents:
print(f"Agent: {agent['name']} - Status: {agent['status']}")
# Get agent configuration
agent = client.agents.get_agent("devops-assistant")
print(f"Capabilities: {agent['capabilities']}")
```
[Learn more about Agents Service β](/sdk/control-plane-agents)
### Workers Service
Monitor and manage execution infrastructure:
```python theme={null}
# List workers
workers = client.workers.list_workers()
for worker in workers:
print(f"Worker: {worker['id']} - Status: {worker['status']}")
# Get worker status
worker = client.workers.get_worker("worker-123")
print(f"Health: {worker['health']}")
```
[Learn more about Workers Service β](/sdk/control-plane-workers)
### Teams Service
Create and manage multi-agent teams:
```python theme={null}
# Create a team
team = client.teams.create({
"name": "DevOps Team",
"description": "Infrastructure automation team",
"runtime": "default"
})
# Add agents to team
client.teams.add_agent(team_id=team['id'], agent_id="agent-uuid")
# Execute team task with streaming
for chunk in client.teams.execute_stream(
team_id=team['id'],
execution_data={
"prompt": "Deploy to production",
"worker_queue_id": "queue-uuid"
}
):
print(chunk.decode('utf-8'), end='', flush=True)
```
[Learn more about Teams Service β](/sdk/control-plane-teams)
### Projects Service
Organize agents and teams into projects:
```python theme={null}
# Create a project
project = client.projects.create({
"name": "E-commerce Platform",
"description": "Main e-commerce project"
})
# Add agents to project
client.projects.add_agent(
project_id=project['id'],
agent_id="agent-uuid",
role="developer"
)
# Add team to project
client.projects.add_team(
project_id=project['id'],
team_id="team-uuid",
role="infrastructure"
)
# List project agents
agents = client.projects.list_agents(project_id=project['id'])
```
[Learn more about Projects Service β](/sdk/control-plane-projects)
### Environments Service
Manage runtime environments:
```python theme={null}
# Create environment
environment = client.environments.create({
"name": "Production",
"description": "Production environment",
"status": "active"
})
# Get worker registration command
worker_info = client.environments.get_worker_command(
environment_id=environment['id']
)
print(f"Worker command: {worker_info['command']}")
# List environments
environments = client.environments.list(status_filter="active")
```
[Learn more about Environments Service β](/sdk/control-plane-environments)
### Jobs Service
Schedule and automate tasks:
```python theme={null}
# Create a cron job
job = client.jobs.create({
"name": "Daily Backup",
"trigger_type": "cron",
"cron_schedule": "0 2 * * *", # Daily at 2 AM
"planning_mode": "predefined_agent",
"entity_type": "agent",
"entity_id": "backup-agent-uuid",
"prompt_template": "Perform database backup"
})
# Create a webhook job
webhook_job = client.jobs.create({
"name": "Deploy on Push",
"trigger_type": "webhook",
"planning_mode": "predefined_team",
"entity_type": "team",
"entity_id": "deployment-team-uuid",
"prompt_template": "Deploy {{branch}} to {{environment}}"
})
print(f"Webhook URL: {webhook_job['webhook_url']}")
# Manually trigger a job
result = client.jobs.trigger(
job_id=job['id'],
parameters={"database": "prod"}
)
```
[Learn more about Jobs Service β](/sdk/control-plane-jobs)
### Context Service
Manage entity context with inheritance:
```python theme={null}
# Update agent context
client.context.update(
entity_type="agent",
entity_id="agent-uuid",
context_data={
"region": "us-west-2",
"tier": "premium",
"tags": ["production", "critical"]
}
)
# Resolve context with inheritance
resolved = client.context.resolve(
entity_type="agent",
entity_id="agent-uuid"
)
print(f"Resolved context: {resolved}")
# Get entity-specific context
context = client.context.get(
entity_type="team",
entity_id="team-uuid"
)
```
[Learn more about Context Service β](/sdk/control-plane-context)
### Task Planning Service
AI-powered task planning and decomposition:
```python theme={null}
# Generate a task plan
plan = client.task_planning.plan(
task_description="Deploy new microservice to production",
context={
"service_name": "payment-api",
"environment": "production"
}
)
print(f"Plan: {plan['title']}")
for i, step in enumerate(plan['steps'], 1):
print(f"{i}. {step['description']}")
# Stream plan generation
for update in client.task_planning.plan_stream(
task_description="Implement CI/CD pipeline"
):
print(update['data'], end='', flush=True)
```
[Learn more about Task Planning Service β](/sdk/control-plane-task-planning)
## Error Handling
The SDK provides specialized exceptions for different failure scenarios:
```python theme={null}
from kubiya.core.exceptions import (
APIError as KubiyaAPIError,
AuthenticationError as KubiyaAuthenticationError,
ConnectionError as KubiyaConnectionError,
TimeoutError as KubiyaTimeoutError,
)
from kubiya.resources.exceptions import (
GraphError,
ValidationError,
ModelError,
WorkerError,
)
# Handle graph errors
try:
result = client.graph.intelligent_search(keywords="test query")
except GraphError as e:
print(f"Graph operation failed: {e}")
# Handle authentication errors
try:
client = ControlPlaneClient(api_key="invalid-key")
client.health.check()
except KubiyaAuthenticationError as e:
print(f"Authentication failed: {e}")
# Handle API errors
try:
result = client.datasets.get_dataset("non-existent-dataset")
except KubiyaAPIError as e:
print(f"API error: {e.status_code}")
# Handle validation errors
try:
client.ingestion.ingest_node(
id="", # Invalid empty ID
labels=[],
properties={}
)
except ValidationError as e:
print(f"Validation failed: {e}")
# Handle timeout errors
try:
result = client.graph.intelligent_search(
keywords="complex query",
max_turns=10
)
except KubiyaTimeoutError as e:
print(f"Request timed out: {e}")
```
[Learn more about Error Handling β](/sdk/error-handling)
## Best Practices
### 1. Use Environment Variables for Credentials
```python theme={null}
# Good - secure and flexible
from kubiya import ControlPlaneClient
client = ControlPlaneClient() # Uses KUBIYA_API_KEY from environment
# Avoid - hardcoding credentials
client = ControlPlaneClient(api_key="hardcoded-key") # Don't do this!
```
### 2. Handle Errors Appropriately
```python theme={null}
from kubiya.resources.exceptions import GraphError
try:
result = client.graph.intelligent_search(keywords="query")
# Process result
except GraphError as e:
print(f"Search failed: {e}")
# Implement retry or fallback logic
```
### 3. Use Batch Operations for Efficiency
```python theme={null}
# Good - batch import
client.ingestion.ingest_nodes_batch(
nodes=[
{"id": "node-1", "labels": ["Type1"], "properties": {}},
{"id": "node-2", "labels": ["Type2"], "properties": {}},
# ... more nodes
]
)
# Avoid - individual imports in a loop
# for node in nodes:
# client.ingestion.ingest_node(...) # Less efficient
```
### 4. Clean Up Resources
```python theme={null}
# Always clean up sessions when done
session_id = None
try:
result = client.graph.intelligent_search(keywords="query")
session_id = result["session_id"]
# Continue conversation...
follow_up = client.graph.intelligent_search(
keywords="follow-up query",
session_id=session_id
)
finally:
if session_id:
client.graph.delete_search_session(session_id)
```
### 5. Use Type Hints
```python theme={null}
from kubiya import ControlPlaneClient
from typing import List, Dict
def search_infrastructure(
client: ControlPlaneClient,
query: str
) -> List[Dict]:
"""Search infrastructure using intelligent search."""
result = client.graph.intelligent_search(keywords=query)
return result["nodes"]
```
[Learn more about Best Practices β](/sdk/best-practices)
## Common Patterns
### Pagination
```python theme={null}
# Paginate through results
def list_all_datasets(client: ControlPlaneClient):
skip = 0
limit = 50
while True:
datasets = client.datasets.list_datasets(skip=skip, limit=limit)
if not datasets:
break
for dataset in datasets:
print(f"Dataset: {dataset['name']}")
skip += limit
```
### Retry Logic
```python theme={null}
from time import sleep
from kubiya.resources.exceptions import APIError
def retry_operation(operation, max_attempts=3):
"""Retry an operation with exponential backoff."""
for attempt in range(max_attempts):
try:
return operation()
except APIError as e:
if attempt == max_attempts - 1:
raise
print(f"Attempt {attempt + 1} failed, retrying...")
sleep(2 ** attempt) # Exponential backoff
# Usage
result = retry_operation(
lambda: client.graph.intelligent_search(keywords="query")
)
```
### Context Manager
```python theme={null}
from contextlib import contextmanager
@contextmanager
def control_plane_client(api_key: str):
"""Context manager for ControlPlaneClient."""
client = ControlPlaneClient(api_key=api_key)
try:
yield client
finally:
# Cleanup if needed
pass
# Usage
with control_plane_client("your-api-key") as client:
result = client.graph.get_stats()
print(f"Nodes: {result['node_count']}")
```
## Architecture
The Control Plane Client SDK is organized into service modules:
```
ControlPlaneClient
βββ graph # Context Graph operations
β βββ intelligent_search()
β βββ semantic_search()
β βββ store_memory()
β βββ recall_memory()
βββ datasets # Dataset management
β βββ create_dataset()
β βββ list_datasets()
β βββ get_dataset()
βββ ingestion # Graph data ingestion
β βββ ingest_node()
β βββ ingest_nodes_batch()
βββ models # LLM model management
β βββ list()
β βββ get()
β βββ get_default()
βββ runtimes # Agent runtime configs
β βββ list()
β βββ get()
βββ context # Context resolution & management
β βββ get()
β βββ update()
β βββ delete()
β βββ resolve()
βββ skills # Tool sets and capabilities
β βββ list()
β βββ get()
βββ policies # Policy management
β βββ list()
β βββ get()
βββ task_planning # AI-powered task planning
β βββ plan()
β βββ plan_stream()
βββ agents # Agent management
β βββ list()
β βββ get()
β βββ create()
β βββ update()
βββ teams # Multi-agent teams
β βββ list()
β βββ get()
β βββ create()
β βββ add_agent()
β βββ execute()
β βββ execute_stream()
βββ projects # Project organization
β βββ list()
β βββ get()
β βββ create()
β βββ add_agent()
β βββ add_team()
βββ environments # Runtime environments
β βββ list()
β βββ get()
β βββ create()
β βββ get_worker_command()
βββ jobs # Scheduled & webhook jobs
β βββ list()
β βββ get()
β βββ create()
β βββ trigger()
β βββ enable()
β βββ disable()
βββ workers # Worker management
β βββ list()
β βββ get()
βββ secrets # Secrets management
β βββ list()
β βββ get()
βββ integrations # Integration configs
βββ list()
βββ get()
```
## Next Steps
Explore the knowledge graph
AI-powered graph search
Multi-agent team collaboration
Schedule and automate tasks
Organize agents and teams
Manage runtime environments
AI-powered task planning
Context resolution with inheritance
Manage agents
Manage LLM models
Configure tools and capabilities
SDK best practices guide