Client SDK Overview

The Kubiya Client SDK provides a comprehensive Python interface to all Kubiya platform services, enabling you to programmatically manage agents, workflows, integrations, secrets, and knowledge. What it provides:

Type-safe Python interfaces to platform services
Automatic authentication and request handling
Streaming support for real-time updates
Comprehensive error handling with specific exceptions
Retry logic for resilient operations

Quick Start

from kubiya import ControlPlaneClient

# Initialize the client (reads KUBIYA_API_KEY from environment)
client = ControlPlaneClient()

# Or pass the API key directly
client = ControlPlaneClient(
    api_key="your-api-key",
    base_url="https://control-plane.kubiya.ai"
)

# Use platform services
agents = client.agents.list()
teams = client.teams.list()
secrets = client.secrets.list()

Available Services

The Client SDK provides access to these platform services:

Agents Service

Create and manage AI agents with custom capabilities and configurations

Client Initialization

Basic Initialization

The ControlPlaneClient supports two authentication methods:

Environment variable (recommended): Set KUBIYA_API_KEY in your environment
Direct parameter: Pass api_key directly to the constructor

from kubiya import ControlPlaneClient

# Option 1: Using environment variables (recommended)
client = ControlPlaneClient()

# Option 2: Explicit configuration
client = ControlPlaneClient(
    api_key="your-api-key",
    base_url="https://control-plane.kubiya.ai"
)

If api_key is not provided and KUBIYA_API_KEY environment variable is not set, a ValueError will be raised.

Advanced Configuration

client = ControlPlaneClient(
    api_key="your-api-key",
    base_url="https://control-plane.kubiya.ai",
    timeout=300,  # Request timeout in seconds
    max_retries=3,  # Maximum number of retry attempts
    org_name="your-org"  # Organization name for API calls
)

Using Environment Variables

Set these environment variables to avoid hardcoding credentials:

export KUBIYA_API_KEY="your-api-key"
export KUBIYA_CONTROL_PLANE_BASE_URL="https://control-plane.kubiya.ai"  # Optional

Then simply initialize without parameters:

client = ControlPlaneClient()  # Reads from environment variables

The client will automatically read:

KUBIYA_API_KEY for authentication (required)
KUBIYA_CONTROL_PLANE_BASE_URL for the API endpoint (optional, defaults to https://control-plane.kubiya.ai)

Core Features

Agents Management

Create, configure, and manage AI agents:

# Create an agent
agent = client.agents.create(
    name="devops-assistant",
    description="AI assistant for DevOps automation",
    llm_model="claude-sonnet-4",
    ai_instructions="You are a helpful DevOps assistant. Always be clear and concise.",
    tools=["kubectl", "terraform"],
    integrations=["slack", "github"]
)

# List agents
agents = client.agents.list(limit=10)

# Update agent
client.agents.edit(
    agent['uuid'],
    llm_model="claude-opus-4",
    add_tools=["docker"]
)

# Delete agent
client.agents.delete(agent['uuid'])

Learn more about Agents Service →

Workflow Execution

Execute workflows with real-time streaming:

workflow_def = {
    "name": "deploy-service",
    "steps": [
        {"name": "build", "command": "docker build -t myapp ."},
        {"name": "deploy", "command": "kubectl apply -f deployment.yaml"}
    ]
}

# Stream execution events
for event in client.workflows.execute(
    workflow_definition=workflow_def,
    parameters={"environment": "production"},
    stream=True
):
    print(f"Event: {event}")

Integration Management

Connect and manage external services:

# Activate GitHub App integration
result = client.integrations.activate("github_app")
if result.get("success"):
    print(f"Install GitHub App at: {result.get('install_url')}")

# List active integrations
integrations = client.integrations.list()

# Get credentials for a specific integration
credentials = client.integrations.get_integration_credentials(
    vendor="aws",
    id="aws"
)

Secrets Management

Securely store and retrieve sensitive data:

# Create secret
client.secrets.create(
    name="database-password",
    value="super-secret-password",
    description="Production database password"
)

# Get secret value
password = client.secrets.value("database-password")

# List secrets (returns metadata only)
secrets = client.secrets.list()

Knowledge Query

Search the knowledge base with AI:

# Query knowledge
response = client.knowledge.query(
    prompt="How do I deploy a Kubernetes application?"
)
print(response)

# Stream search results
for event in client.knowledge.query(
    prompt="Best practices for container security",
    stream=True
):
    print(event)

Error Handling

The SDK provides specialized exceptions for different failure scenarios:

from kubiya.kubiya_services.exceptions import (
    AgentError,
    WorkflowExecutionError,
    IntegrationError,
    SecretError,
    KnowledgeError
)

# Handle agent errors
try:
    agent = client.agents.get("non-existent-agent")
except AgentError as e:
    print(f"Agent operation failed: {e}")

# Handle workflow errors
try:
    result = client.workflows.execute(workflow_def, params)
except WorkflowExecutionError as e:
    print(f"Workflow failed: {e}")
    if e.details.get("step"):
        print(f"Failed at step: {e.details['step']}")

# Handle integration errors
try:
    client.integrations.activate("invalid-integration")
except IntegrationError as e:
    print(f"Integration failed: {e}")

Async Support

For non-blocking operations, use the async client:

from kubiya import StreamingKubiyaClient
import asyncio

async def main():
    async with StreamingKubiyaClient(api_key="your-api-key") as client:
        # Async workflow execution
        async for event in client.execute_workflow_stream(workflow_dict):
            print(f"Event: {event}")

asyncio.run(main())

Best Practices

1. Always Use Error Handling

try:
    result = client.agents.create(name="my-agent")
except AgentError as e:
    print(f"Failed to create agent: {e}")
    # Handle appropriately

2. Use Environment Variables

# Good - secure and flexible
client = ControlPlaneClient()  # Uses KUBIYA_API_KEY from environment

# Avoid - hardcoding credentials
client = ControlPlaneClient(api_key="hardcoded-key")  # Don't do this!

3. Handle Streaming Gracefully

try:
    for event in client.workflows.execute(workflow_def, params, stream=True):
        if isinstance(event, dict):
            event_type = event.get("type", "unknown")
            print(f"Event type: {event_type}")
except WorkflowExecutionError as e:
    print(f"Streaming failed: {e}")

4. Validate Inputs

def validate_workflow(workflow_def):
    if not isinstance(workflow_def, dict):
        raise ValueError("Workflow definition must be a dictionary")
    if "name" not in workflow_def:
        raise ValueError("Workflow must have a name")
    return workflow_def

# Use validation
validated_workflow = validate_workflow(my_workflow)
result = client.workflows.execute(validated_workflow, params)

Common Patterns

Pagination

# List items with pagination
limit = 10
offset = 0

while True:
    agents = client.agents.list(limit=limit, offset=offset)
    if not agents:
        break

    for agent in agents:
        print(f"Agent: {agent['name']}")

    offset += limit

Retry Logic

from time import sleep

def retry_operation(operation, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            return operation()
        except Exception as e:
            if attempt == max_attempts - 1:
                raise
            print(f"Attempt {attempt + 1} failed, retrying...")
            sleep(2 ** attempt)  # Exponential backoff

# Use retry wrapper
result = retry_operation(lambda: client.agents.get("agent-uuid"))

Context Manager

class KubiyaClientContext:
    def __init__(self, api_key):
        self.client = None
        self.api_key = api_key

    def __enter__(self):
        self.client = KubiyaClient(api_key=self.api_key)
        return self.client

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Cleanup if needed
        pass

# Usage
with KubiyaClientContext("your-api-key") as client:
    agents = client.agents.list()

Advanced: Control Plane Client

For advanced orchestration and infrastructure use cases, the SDK also includes a ControlPlaneClient for accessing internal services like the context graph, models, and skills. Learn about Control Plane Client →

Next Steps

Agents Service

Learn how to create and manage AI agents

Getting Started

Control Plane Client

Context Graph & Memory

Advanced

Resources

Client SDK Overview

Quick Start

Available Services

Agents Service

Client Initialization

Basic Initialization

Advanced Configuration

Using Environment Variables

Core Features

Agents Management

Workflow Execution

Integration Management

Secrets Management

Knowledge Query

Error Handling

Async Support

Best Practices

1. Always Use Error Handling

2. Use Environment Variables

3. Handle Streaming Gracefully

4. Validate Inputs

Common Patterns

Retry Logic

Context Manager

Advanced: Control Plane Client

Next Steps

Agents Service

​Quick Start

​Available Services

Agents Service

​Client Initialization

​Basic Initialization

​Advanced Configuration

​Using Environment Variables

​Core Features

​Agents Management

​Workflow Execution

​Integration Management

​Secrets Management

​Knowledge Query

​Error Handling

​Async Support

​Best Practices

​1. Always Use Error Handling

​2. Use Environment Variables

​3. Handle Streaming Gracefully

​4. Validate Inputs

​Common Patterns

​Pagination

​Retry Logic

​Context Manager

​Advanced: Control Plane Client

​Next Steps

Agents Service

Quick Start

Available Services

Client Initialization

Basic Initialization

Advanced Configuration

Using Environment Variables

Core Features

Agents Management

Workflow Execution

Integration Management

Secrets Management

Knowledge Query

Error Handling

Async Support

Best Practices

1. Always Use Error Handling

2. Use Environment Variables

3. Handle Streaming Gracefully

4. Validate Inputs

Common Patterns

Pagination

Retry Logic

Context Manager

Advanced: Control Plane Client

Next Steps