> ## 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.

# Client SDK Overview

> Python client for interacting with Kubiya platform APIs

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

```python theme={null}
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:

<Card title="Agents Service" icon="robot" href="/sdk/client-agents">
  Create and manage AI agents with custom capabilities and configurations
</Card>

## Client Initialization

### Basic Initialization

The `ControlPlaneClient` supports two authentication methods:

1. **Environment variable** (recommended): Set `KUBIYA_API_KEY` in your environment
2. **Direct parameter**: Pass `api_key` directly to the constructor

```python theme={null}
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"
)
```

<Note>
  If `api_key` is not provided and `KUBIYA_API_KEY` environment variable is not set, a `ValueError` will be raised.
</Note>

### Advanced Configuration

```python theme={null}
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:

```bash theme={null}
export KUBIYA_API_KEY="your-api-key"
export KUBIYA_CONTROL_PLANE_BASE_URL="https://control-plane.kubiya.ai"  # Optional
```

Then simply initialize without parameters:

```python theme={null}
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:

```python theme={null}
# 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 →](/sdk/client-agents)

### Workflow Execution

Execute workflows with real-time streaming:

```python theme={null}
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:

```python theme={null}
# 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:

```python theme={null}
# 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:

```python theme={null}
# 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:

```python theme={null}
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:

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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 →](/sdk/control-plane-client-overview)

## Next Steps

<Card title="Agents Service" icon="robot" href="/sdk/client-agents">
  Learn how to create and manage AI agents
</Card>