Kubiya Client SDK

pip install kubiya-workflow-sdk

from kubiya_workflow_sdk import KubiyaClient

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

# Create an agent
agent = client.agents.create(
    name="DevOps Helper",
    description="Automated DevOps operations",
    llm_model="claude-4-sonnet",
    tools=["kubectl", "docker", "github"]
)

# Execute a workflow
for event in client.workflows.execute(
    workflow_definition={
        "name": "deploy-service",
        "steps": [{"name": "deploy", "command": "kubectl apply -f deployment.yaml"}]
    },
    parameters={"environment": "production"},
    stream=True
):
    print(f"Workflow event: {event}")

from kubiya_workflow_sdk import KubiyaClient
from kubiya_workflow_sdk.kubiya_services.exceptions import AgentError

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

try:
    # List all agents
    agents_list = client.agents.list(limit=10)
    
    # Get specific agent
    agent = client.agents.get("agent-uuid-here")
    
    # Update agent configuration
    client.agents.edit(
        "agent-uuid-here",
        llm_model="claude-4-opus",
        add_tools=["new-tool"],
        remove_tools=["old-tool"]
    )
    
    # Delete agent
    client.agents.delete("agent-uuid-here")
    
except AgentError as e:
    print(f"Agent operation failed: {e}")

from kubiya_workflow_sdk.kubiya_services.exceptions import WorkflowExecutionError

# Define workflow
workflow_definition = {
    "name": "deploy-frontend",
    "steps": [
        {
            "name": "build",
            "command": "docker build -t frontend:latest .",
            "executor": {"type": "command", "config": {}}
        },
        {
            "name": "deploy", 
            "command": "kubectl apply -f k8s/",
            "executor": {"type": "command", "config": {}}
        }
    ]
}

try:
    # Execute workflow with streaming
    for event in client.workflows.execute(
        workflow_definition=workflow_definition,
        parameters={
            "service_name": "frontend",
            "version": "v1.2.0",
            "environment": "staging"
        },
        stream=True
    ):
        print(f"Workflow progress: {event}")
    
    # List recent executions
    executions = client.workflows.list(filter="completed", limit=10)
    
except WorkflowExecutionError as e:
    print(f"Workflow execution failed: {e}")

from kubiya_workflow_sdk.kubiya_services.exceptions import IntegrationError, IntegrationNotFoundError

try:
    # Activate GitHub App integration
    result = client.integrations.activate("github_app")
    
    if result.get("success"):
        print("✅ GitHub App integration activated successfully!")
        install_url = result.get("install_url")
        if install_url:
            print(f"Complete installation at: {install_url}")
    
except IntegrationNotFoundError:
    print("Integration already exists")
except IntegrationError as e:
    print(f"Integration activation failed: {e}")

from kubiya_workflow_sdk.kubiya_services.exceptions import SecretError, SecretValidationError

try:
    # Create secret
    create_result = client.secrets.create(
        name="github-token",
        value="ghp_xxxxxxxxxxxx",
        description="GitHub API token for CI/CD"
    )
    
    # List secrets (returns metadata only)
    secrets_list = client.secrets.list()
    
    # Get secret metadata
    secret_info = client.secrets.get("github-token")
    
    # Get actual secret value
    secret_value = client.secrets.value("github-token")
    
    # Update secret
    client.secrets.update(
        name="github-token",
        value="new-token-value",
        description="Updated GitHub token"
    )
    
    # Delete secret
    client.secrets.delete("github-token")
    
except SecretValidationError as e:
    print(f"Secret validation failed: {e}")
except SecretError as e:
    print(f"Secret operation failed: {e}")

from kubiya_workflow_sdk.kubiya_services.exceptions import KnowledgeError

try:
    # Simple knowledge query
    response = client.knowledge.query(
        prompt="How do I deploy a Kubernetes application?"
    )
    print(f"Knowledge response: {response}")
    
    # Stream search results
    for event in client.knowledge.query(
        prompt="Best practices for container security",
        stream=True
    ):
        print(f"Search event: {event}")
        
except KnowledgeError as e:
    print(f"Knowledge query failed: {e}")

from kubiya_workflow_sdk.kubiya_services.exceptions import (
    AgentError,
    WorkflowExecutionError,
    IntegrationError,
    IntegrationNotFoundError,
    SecretError,
    SecretValidationError,
    KnowledgeError
)

try:
    agent = client.agents.get("non-existent-agent")
except AgentError:
    print("Agent operation failed")

try:
    workflow_result = client.workflows.execute(invalid_workflow)
except WorkflowExecutionError as e:
    print(f"Workflow execution failed: {e}")
    if e.details.get("step"):
        print(f"Failed at step: {e.details['step']}")

try:
    integration_result = client.integrations.activate("invalid-integration")
except IntegrationError as e:
    print(f"Integration failed: {e}")
except IntegrationNotFoundError as e:
    print(f"Integration already exists: {e}")

export KUBIYA_API_KEY="your-api-key"
export KUBIYA_BASE_URL="https://api.kubiya.ai"

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

# Using environment variables
import os
client = KubiyaClient(
    api_key=os.getenv("KUBIYA_API_KEY"),
    base_url=os.getenv("KUBIYA_BASE_URL", "https://api.kubiya.ai")
)

# Always wrap service calls in try-except blocks
try:
    result = client.agents.create(name="my-agent")
except AgentError as e:
    print(f"Failed to create agent: {e}")
    # Handle the error appropriately

# Use clear, descriptive names for resources
agent = client.agents.create(
    name="production-devops-agent",
    description="Agent for production DevOps automation tasks",
    llm_model="claude-4-sonnet"
)

# Process streaming results with proper error handling
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}")
        else:
            print(f"Event: {event}")
except WorkflowExecutionError as e:
    print(f"Streaming failed: {e}")

# Validate workflow definitions before execution
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")
    if "steps" not in workflow_def:
        raise ValueError("Workflow must have steps")
    return workflow_def

# Use validation
try:
    validated_workflow = validate_workflow(my_workflow)
    result = client.workflows.execute(validated_workflow, parameters)
except ValueError as e:
    print(f"Validation failed: {e}")