Error Handling

The Kubiya SDK provides a comprehensive exception hierarchy to help you handle errors gracefully and build resilient applications.

Overview

The SDK uses specialized exception classes to indicate different failure scenarios:

Core Exceptions: Authentication, connection, timeout errors
Resource Exceptions: Service-specific errors (graph, models, agents, etc.)
Validation Errors: Invalid parameters or data structures
API Errors: HTTP-level failures with status codes

Always catch specific exceptions before generic ones to handle different error types appropriately.

Exception Hierarchy

KubiyaSDKError (base)
├── Core Exceptions (kubiya.core.exceptions)
│   ├── APIError
│   ├── AuthenticationError
│   ├── ConnectionError
│   ├── TimeoutError
│   └── RateLimitError
└── Resource Exceptions (kubiya.resources.exceptions)
    ├── ControlPlaneError (base for Control Plane services)
    │   ├── GraphError
    │   ├── ModelError
    │   ├── RuntimeError
    │   ├── SkillError
    │   ├── PolicyError
    │   ├── AgentError
    │   ├── WorkerError
    │   └── IntegrationError
    └── ValidationError

Quick Start

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import (
    APIError as KubiyaAPIError,
    AuthenticationError as KubiyaAuthenticationError,
    TimeoutError as KubiyaTimeoutError
)
from kubiya.resources.exceptions import GraphError

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

try:
    result = client.graph.intelligent_search(keywords="test query")
except GraphError as e:
    print(f"Graph operation failed: {e}")
except KubiyaAuthenticationError as e:
    print(f"Authentication failed: {e}")
except KubiyaTimeoutError as e:
    print(f"Request timed out: {e}")
except KubiyaAPIError as e:
    print(f"API error {e.status_code}: {e.message}")

Core Exceptions

APIError

Generic API-level errors with HTTP status codes:

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import APIError as KubiyaAPIError

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

try:
    result = client.graph.intelligent_search(keywords="test")
except KubiyaAPIError as e:
    print(f"Status Code: {e.status_code}")
    print(f"Message: {e.message}")

    if e.status_code == 400:
        print("Bad request - check parameters")
    elif e.status_code == 404:
        print("Resource not found")
    elif e.status_code == 500:
        print("Server error - retry later")

AuthenticationError

Authentication and authorization failures:

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import AuthenticationError as KubiyaAuthenticationError

try:
    client = ControlPlaneClient(api_key="invalid-key")
    client.datasets.list_datasets()
except KubiyaAuthenticationError as e:
    print(f"Authentication failed: {e}")
    print("Please check your API key")

ConnectionError

Network connectivity issues:

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import ConnectionError as KubiyaConnectionError

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

try:
    result = client.graph.intelligent_search(keywords="test")
except KubiyaConnectionError as e:
    print(f"Connection failed: {e}")
    print("Check network connectivity and firewall settings")

TimeoutError

Request timeout failures:

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import TimeoutError as KubiyaTimeoutError

client = ControlPlaneClient(api_key="your-api-key", timeout=5)  # 5 second timeout

try:
    result = client.graph.intelligent_search(
        keywords="complex query",
        max_turns=20  # Might take a while
    )
except KubiyaTimeoutError as e:
    print(f"Request timed out: {e}")
    print("Consider increasing timeout or reducing complexity")

RateLimitError

API rate limit exceeded:

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import RateLimitError as KubiyaRateLimitError
import time

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

try:
    for i in range(1000):
        client.datasets.list_datasets()
except KubiyaRateLimitError as e:
    print(f"Rate limit exceeded: {e}")
    retry_after = e.retry_after or 60
    print(f"Retry after {retry_after} seconds")
    time.sleep(retry_after)

Resource-Specific Exceptions

GraphError

Context Graph operation failures:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import GraphError

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

try:
    # Intelligent search
    result = client.graph.intelligent_search(keywords="test")
except GraphError as e:
    if "session" in str(e).lower():
        print("Session error - may have expired")
    elif "not found" in str(e).lower():
        print("Resource not found in graph")
    else:
        print(f"Graph operation failed: {e}")

try:
    # Memory operations
    memories = client.graph.store_memory(
        dataset_id="non-existent-dataset",
        context="test"
    )
except GraphError as e:
    print(f"Memory operation failed: {e}")

ModelError

LLM model management errors:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import ModelError

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

try:
    model = client.models.get(model_id="non-existent-model")
except ModelError as e:
    if "not found" in str(e).lower():
        print("Model not found - check model ID")
    else:
        print(f"Model operation failed: {e}")

AgentError

Agent management errors:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import AgentError

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

try:
    agent = client.agents.execute(
        agent_id="my-agent",
        execution_data={"input": "test"}
    )
except AgentError as e:
    if "not found" in str(e).lower():
        print("Agent not found")
    elif "execution" in str(e).lower():
        print("Agent execution failed - check agent configuration")
    else:
        print(f"Agent error: {e}")

PolicyError

Policy management errors:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import PolicyError

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

try:
    result = client.policies.check_authorization(
        policy_id="security-policy",
        context={"action": "delete", "resource": "database"}
    )
except PolicyError as e:
    print(f"Policy check failed: {e}")

WorkerError

Worker management errors:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import WorkerError

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

try:
    workers = client.workers.list()
except WorkerError as e:
    print(f"Worker operation failed: {e}")

IntegrationError

Integration management errors:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError

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

try:
    credentials = client.integrations.get_integration_credentials(
        vendor="github",
        id="github-prod"
    )
except IntegrationError as e:
    if "not found" in str(e).lower():
        print("Integration not configured")
    elif "permission" in str(e).lower():
        print("Insufficient permissions to access integration")
    else:
        print(f"Integration error: {e}")

Error Handling Patterns

Try-Except-Finally

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import GraphError

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

try:
    result = client.graph.intelligent_search(keywords="test query")
    session_id = result["session_id"]

    # Use session for follow-up queries
    follow_up = client.graph.intelligent_search(
        keywords="follow-up",
        session_id=session_id
    )

except GraphError as e:
    print(f"Search failed: {e}")

finally:
    # Always clean up session
    if session_id:
        try:
            client.graph.delete_search_session(session_id)
        except Exception:
            pass  # Ignore cleanup errors

Multiple Exception Types

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import (
    AuthenticationError as KubiyaAuthenticationError,
    ConnectionError as KubiyaConnectionError,
    TimeoutError as KubiyaTimeoutError,
    APIError as KubiyaAPIError
)
from kubiya.resources.exceptions import GraphError

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

try:
    result = client.graph.intelligent_search(keywords="query")

except GraphError as e:
    # Handle graph-specific errors
    print(f"Graph error: {e}")
    # Implement retry or alternative search method

except KubiyaAuthenticationError as e:
    # Handle authentication errors
    print(f"Authentication failed: {e}")
    # Prompt for new credentials or exit

except KubiyaConnectionError as e:
    # Handle connection errors
    print(f"Connection failed: {e}")
    # Wait and retry with backoff

except KubiyaTimeoutError as e:
    # Handle timeout errors
    print(f"Request timed out: {e}")
    # Retry with reduced complexity or increased timeout

except KubiyaAPIError as e:
    # Handle general API errors
    print(f"API error {e.status_code}: {e.message}")
    # Log error and notify monitoring

Retry with Exponential Backoff

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import (
    ConnectionError as KubiyaConnectionError,
    TimeoutError as KubiyaTimeoutError,
    APIError as KubiyaAPIError
)
import time

def retry_with_backoff(operation, max_retries=3, base_delay=1):
    """Retry operation with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return operation()

        except (KubiyaConnectionError, KubiyaTimeoutError) as e:
            if attempt == max_retries - 1:
                raise  # Final attempt failed

            delay = base_delay * (2 ** attempt)
            print(f"Attempt {attempt + 1} failed: {e}")
            print(f"Retrying in {delay} seconds...")
            time.sleep(delay)

        except KubiyaAPIError as e:
            # Don't retry on certain status codes
            if e.status_code in [400, 401, 403, 404]:
                raise  # Client error - don't retry

            if attempt == max_retries - 1:
                raise

            delay = base_delay * (2 ** attempt)
            print(f"API error, retrying in {delay} seconds...")
            time.sleep(delay)

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

result = retry_with_backoff(
    lambda: client.graph.intelligent_search(keywords="test")
)

Context Manager for Cleanup

from kubiya import ControlPlaneClient
from contextlib import contextmanager

@contextmanager
def search_session(client: ControlPlaneClient, keywords: str):
    """Context manager for intelligent search with automatic cleanup."""
    session_id = None

    try:
        result = client.graph.intelligent_search(keywords=keywords)
        session_id = result["session_id"]
        yield result

    finally:
        if session_id:
            try:
                client.graph.delete_search_session(session_id)
            except Exception:
                pass  # Ignore cleanup errors

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

with search_session(client, "test query") as result:
    print(result["answer"])

    # Session automatically cleaned up on exit

Custom Error Handler

from kubiya import ControlPlaneClient
from kubiya.core.exceptions import APIError as KubiyaAPIError
from kubiya.resources.exceptions import GraphError
import logging

logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

def handle_sdk_error(error: Exception, operation: str):
    """Centralized error handler for SDK operations."""
    if isinstance(error, GraphError):
        logger.error(f"Graph error during {operation}: {error}")
        return {"success": False, "error_type": "graph", "message": str(error)}

    elif isinstance(error, KubiyaAPIError):
        logger.error(f"API error during {operation}: {error.status_code} - {error.message}")
        return {
            "success": False,
            "error_type": "api",
            "status_code": error.status_code,
            "message": error.message
        }

    else:
        logger.error(f"Unexpected error during {operation}: {error}")
        return {"success": False, "error_type": "unknown", "message": str(error)}

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

try:
    result = client.graph.intelligent_search(keywords="test")
    print(result)

except Exception as e:
    error_info = handle_sdk_error(e, "intelligent_search")
    print(f"Operation failed: {error_info}")

Validation Errors

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import ValidationError

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

try:
    # Attempt invalid operation
    result = client.ingestion.ingest_node(
        id="",  # Empty ID - invalid
        labels=[],  # Empty labels - invalid
        properties={}
    )

except ValidationError as e:
    print(f"Validation error: {e}")
    print("Please provide valid node ID and at least one label")

Best Practices

1. Catch Specific Exceptions First

# ✅ GOOD - Specific exceptions first
try:
    result = client.graph.intelligent_search(keywords="test")
except GraphError as e:
    print(f"Graph error: {e}")
except KubiyaAPIError as e:
    print(f"API error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

# ❌ BAD - Generic exception catches specific ones
try:
    result = client.graph.intelligent_search(keywords="test")
except Exception as e:  # Too broad
    print(f"Error: {e}")

2. Log Errors Appropriately

import logging
from kubiya import ControlPlaneClient
from kubiya.core.exceptions import APIError as KubiyaAPIError

logger = logging.getLogger(__name__)
client = ControlPlaneClient(api_key="your-api-key")

try:
    result = client.graph.intelligent_search(keywords="test")
except KubiyaAPIError as e:
    logger.error(
        "API error during search",
        extra={
            "status_code": e.status_code,
            "message": e.message,
            "operation": "intelligent_search"
        }
    )
    raise  # Re-raise after logging

3. Provide Context in Error Messages

def search_with_context(client, query):
    """Search with contextual error messages."""
    try:
        return client.graph.intelligent_search(keywords=query)

    except GraphError as e:
        raise RuntimeError(f"Failed to search for '{query}': {e}") from e

    except KubiyaAuthenticationError as e:
        raise RuntimeError("Authentication failed - check API key") from e

4. Don’t Swallow Exceptions Silently

# ❌ BAD - Silent failure
try:
    result = client.graph.intelligent_search(keywords="test")
except Exception:
    pass  # Error ignored!

# ✅ GOOD - Log and handle
try:
    result = client.graph.intelligent_search(keywords="test")
except Exception as e:
    logger.error(f"Search failed: {e}")
    # Handle appropriately or re-raise
    raise

5. Clean Up Resources in Finally Blocks

session_id = None
dataset_id = None

try:
    # Create temporary resources
    dataset = client.datasets.create_dataset(name="temp", scope="user")
    dataset_id = dataset['id']

    result = client.graph.intelligent_search(keywords="test")
    session_id = result["session_id"]

    # Use resources...

except Exception as e:
    print(f"Operation failed: {e}")

finally:
    # Always clean up
    if session_id:
        try:
            client.graph.delete_search_session(session_id)
        except Exception:
            pass

    if dataset_id:
        try:
            client.datasets.delete_dataset(dataset_id=dataset_id)
        except Exception:
            pass

Testing Error Handling

import pytest
from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import GraphError

def test_error_handling():
    """Test that errors are handled correctly."""
    client = ControlPlaneClient(api_key="invalid-key")

    with pytest.raises(GraphError):
        client.graph.intelligent_search(keywords="test")

def test_error_message():
    """Test error message content."""
    client = ControlPlaneClient(api_key="invalid-key")

    with pytest.raises(GraphError) as exc_info:
        client.graph.intelligent_search(keywords="test")

    assert "authentication" in str(exc_info.value).lower() or "unauthorized" in str(exc_info.value).lower()

Next Steps

Best Practices

SDK best practices guide

Control Plane Client

Learn the Control Plane client services

Troubleshooting

Common issues and solutions

Examples

Code examples and recipes

​Overview

​Exception Hierarchy

​Quick Start

​Core Exceptions

​APIError

​AuthenticationError

​ConnectionError

​TimeoutError

​RateLimitError

​Resource-Specific Exceptions

​GraphError

​ModelError

​AgentError

​PolicyError

​WorkerError

​IntegrationError

​Error Handling Patterns

​Try-Except-Finally

​Multiple Exception Types

​Retry with Exponential Backoff

​Context Manager for Cleanup

​Custom Error Handler

​Validation Errors

​Best Practices

​1. Catch Specific Exceptions First

​2. Log Errors Appropriately

​3. Provide Context in Error Messages

​4. Don’t Swallow Exceptions Silently

​5. Clean Up Resources in Finally Blocks

​Testing Error Handling

​Next Steps

Best Practices

Control Plane Client

Troubleshooting

Examples

Overview

Exception Hierarchy

Quick Start

Core Exceptions

APIError

AuthenticationError

ConnectionError

TimeoutError

RateLimitError

Resource-Specific Exceptions

GraphError

ModelError

AgentError

PolicyError

WorkerError

IntegrationError

Error Handling Patterns

Try-Except-Finally

Multiple Exception Types

Retry with Exponential Backoff

Context Manager for Cleanup

Custom Error Handler

Validation Errors

Best Practices

1. Catch Specific Exceptions First

2. Log Errors Appropriately

3. Provide Context in Error Messages

4. Don’t Swallow Exceptions Silently

5. Clean Up Resources in Finally Blocks

Testing Error Handling

Next Steps