Skip to main content
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 KubiyaClient

# Initialize the client
client = KubiyaClient(
    api_key="your-api-key",
    base_url="https://api.kubiya.ai"
)

# Use platform services
agents = client.agents.list()
workflows = client.workflows.execute(workflow_def, params, stream=True)
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

from kubiya import KubiyaClient

# Using environment variables
client = KubiyaClient()

# Explicit configuration
client = KubiyaClient(
    api_key="your-api-key",
    base_url="https://api.kubiya.ai"
)

​
Advanced Configuration

client = KubiyaClient(
    api_key="your-api-key",
    base_url="https://api.kubiya.ai",
    timeout=60,  # Request timeout in seconds
    retry_attempts=3,  # Number of retry attempts
    debug=False  # Enable debug logging
)

​
Using Environment Variables

Set these environment variables to avoid hardcoding credentials: 
export KUBIYA_API_KEY="your-api-key"
export KUBIYA_BASE_URL="https://api.kubiya.ai"
Then simply initialize without parameters: 
client = KubiyaClient()  # Reads from environment variables

​
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",
    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()

​
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 = KubiyaClient()  # Uses KUBIYA_API_KEY from environment

# Avoid - hardcoding credentials
client = KubiyaClient(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()

​
Next Steps

Agents Service

Learn how to create and manage AI agents