API Reference

from kubiya import KubiyaClient

client = KubiyaClient(
    api_key: str = None,
    base_url: str = "https://api.kubiya.ai",
    timeout: int = 30,
    retry_attempts: int = 3,
    debug: bool = False
)

result = client.execute_workflow(
    workflow_dict=wf.to_dict(),
    stream=True,
    runner="kubiya-hosted"
)

agent = client.agents.create(
    name: str,
    description: str = None,
    llm_model: str = "claude-sonnet-4",
    ai_instructions: str = None,
    tools: List[str] = None,
    integrations: List[str] = None,
    environment: Dict[str, str] = None
)

for event in client.workflows.execute(
    workflow_definition: Union[dict, str],
    parameters: Dict[str, Any] = None,
    runner: str = None,
    stream: bool = True
):
    print(event)

result = client.integrations.activate("github_app")

client.secrets.create(
    name="api-token",
    value="secret-value",
    description="API token for external service"
)

# Simple query
response = client.knowledge.query("How do I deploy to Kubernetes?")

# Streaming query
for event in client.knowledge.query("Container security best practices", stream=True):
    print(event)

from kubiya.dsl import workflow

wf = workflow("workflow-name")

# Simple step
wf.step("hello", "echo 'Hello'")

# Advanced step
wf.step("deploy", callback=lambda s:
    s.shell("kubectl apply -f deployment.yaml")
        .description("Deploy to Kubernetes")
        .depends("build")
        .output("DEPLOYMENT_STATUS")
)

# Python step
wf.step("process", callback=lambda s:
    s.python("print('Processing data...')")
)

# Docker step
wf.step("containerized", callback=lambda s:
    s.docker(
        image="python:3.11",
        content="#!/usr/bin/env python3\nprint('In container')"
    )
)

from kubiya import StreamingKubiyaClient
import asyncio

async def main():
    async with StreamingKubiyaClient(api_key="your-key") as client:
        async for event in client.execute_workflow_stream(workflow_dict):
            print(event)

asyncio.run(main())

from kubiya.kubiya_services.exceptions import AgentError

try:
    agent = client.agents.get("invalid-uuid")
except AgentError as e:
    print(f"Agent operation failed: {e}")

from kubiya.kubiya_services.exceptions import WorkflowExecutionError

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

from typing import Dict, List, Optional, Union, Any
from kubiya import KubiyaClient

def create_agent(client: KubiyaClient, name: str) -> Dict[str, Any]:
    """Create an agent with proper type hints"""
    agent: Dict[str, Any] = client.agents.create(
        name=name,
        description="Agent description",
        llm_model="claude-sonnet-4"
    )
    return agent