Agents Service

The Agents Service provides programmatic access to create, configure, and manage AI agents in the Kubiya platform. Agents are autonomous AI assistants that can perform tasks, execute workflows, and interact with various tools and integrations.

For conceptual information about agents and how they work, see Agents Core Concepts.

Use the Agents Service when you want to manage your agent catalog from code: listing and auditing existing agents, rolling out new versions with updated models or skills, enforcing consistent configurations across environments, and triggering executions from CI/CD, runbooks, or internal tools.

Overview

The Agents Service enables you to:

List agents with pagination and filtering
Get agent details including configuration, capabilities, and status
Create agents with custom configurations
Update agent settings for existing agents
Delete agents when no longer needed
Execute agents via Temporal workflows

Quick Start

from kubiya import ControlPlaneClient

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

# List all agents
agents = client.agents.list()
for agent in agents:
    print(f"Agent: {agent['name']} - Status: {agent.get('status', 'N/A')}")

# Execute an agent
execution = client.agents.execute(
    agent_id="agent-uuid",
    execution_data={"prompt": "Deploy to production"}
)
print(f"Execution ID: {execution['execution_id']}")

List Agents

List all agents in your organization with pagination and filtering.

Basic Listing

# List all agents (first 100)
agents = client.agents.list()

for agent in agents:
    print(f"""
    Name: {agent['name']}
    Description: {agent.get('description', 'N/A')}
    Status: {agent.get('status', 'active')}
    Runtime: {agent.get('runtime', 'N/A')}
    """)

Paginated Listing

# List agents with pagination
skip = 0
limit = 20

while True:
    agents = client.agents.list(skip=skip, limit=limit)

    if not agents:
        break

    for agent in agents:
        print(f"Agent: {agent['name']}")

    skip += limit

Filter by Status

# List only active agents
active_agents = client.agents.list(status_filter="active")

print(f"Active agents: {len(active_agents)}")
for agent in active_agents:
    print(f"  - {agent['name']}")

# List inactive agents
inactive_agents = client.agents.list(status_filter="inactive")
print(f"Inactive agents: {len(inactive_agents)}")

List All Agents

def list_all_agents():
    """Get all agents across all pages"""
    all_agents = []
    skip = 0
    limit = 100

    while True:
        agents = client.agents.list(skip=skip, limit=limit)

        if not agents:
            break

        all_agents.extend(agents)
        skip += limit

    return all_agents

# Usage
all_agents = list_all_agents()
print(f"Total agents: {len(all_agents)}")

Get Agent Details

Retrieve detailed information about a specific agent.

By Agent ID

# Get agent by UUID
agent = client.agents.get("agent-uuid-here")

print(f"Agent: {agent['name']}")
print(f"Description: {agent.get('description', 'N/A')}")
print(f"Runtime: {agent.get('runtime', 'N/A')}")
print(f"Model: {agent.get('model', 'N/A')}")
print(f"Skills: {agent.get('skills', [])}")
print(f"Integrations: {agent.get('integrations', [])}")

Extract Agent Information

def get_agent_info(agent_id: str):
    """Get comprehensive agent information"""
    try:
        agent = client.agents.get(agent_id)

        info = {
            "id": agent.get('uuid'),
            "name": agent['name'],
            "description": agent.get('description'),
            "status": agent.get('status', 'active'),
            "runtime": agent.get('runtime'),
            "model": agent.get('model'),
            "skills": agent.get('skills', []),
            "integrations": agent.get('integrations', []),
            "policies": agent.get('policies', []),
            "environment": agent.get('environment', {}),
            "created_at": agent.get('created_at'),
            "updated_at": agent.get('updated_at')
        }

        return info
    except Exception as e:
        print(f"Failed to get agent info: {e}")
        return None

# Usage
info = get_agent_info("agent-uuid")
if info:
    print(f"Agent: {info['name']}")
    print(f"Runtime: {info['runtime']}")
    print(f"Skills: {len(info['skills'])}")

Find Agent by Name

def find_agent_by_name(name: str):
    """Find an agent by name"""
    skip = 0
    limit = 100

    while True:
        agents = client.agents.list(skip=skip, limit=limit)

        if not agents:
            break

        for agent in agents:
            if agent['name'].lower() == name.lower():
                return agent

        skip += limit

    return None

# Usage
agent = find_agent_by_name("devops-assistant")
if agent:
    print(f"Found agent: {agent['uuid']}")
    agent_details = client.agents.get(agent['uuid'])
    print(f"Model: {agent_details.get('model')}")

Create Agent

Create a new AI agent with custom configuration.

Basic Agent Creation

# Create a simple agent
agent_data = {
    "name": "support-assistant",
    "description": "AI assistant for customer support",
    "runtime": "agno",
    "model": "kubiya/claude-sonnet-4",
    "skills": ["knowledge-base", "ticketing"],
    "integrations": ["slack", "zendesk"]
}

created_agent = client.agents.create(agent_data)
print(f"Created agent: {created_agent['uuid']}")
print(f"Name: {created_agent['name']}")

Agent with Environment Variables

# Create agent with environment configuration
agent_data = {
    "name": "devops-agent",
    "description": "DevOps automation agent",
    "runtime": "claude_code",
    "model": "kubiya/claude-sonnet-4",
    "skills": ["kubernetes", "terraform", "aws"],
    "integrations": ["github", "slack"],
    "environment": {
        "AWS_REGION": "us-east-1",
        "KUBECTL_VERSION": "1.28",
        "LOG_LEVEL": "info"
    }
}

created_agent = client.agents.create(agent_data)
print(f"Created DevOps agent: {created_agent['uuid']}")

Agent with Policies

# Create agent with security policies
agent_data = {
    "name": "production-agent",
    "description": "Agent for production operations",
    "runtime": "agno",
    "model": "kubiya/claude-sonnet-4",
    "skills": ["kubernetes", "monitoring"],
    "policies": [
        "production-access-control",
        "deployment-approval-required"
    ],
    "integrations": ["slack", "pagerduty"]
}

created_agent = client.agents.create(agent_data)
print(f"Created production agent: {created_agent['uuid']}")

Agent with Custom Instructions

# Create agent with custom system instructions
agent_data = {
    "name": "code-reviewer",
    "description": "Automated code review agent",
    "runtime": "claude_code",
    "model": "kubiya/claude-opus-4",
    "skills": ["github", "code-analysis"],
    "integrations": ["github"],
    "instructions": """You are a code review expert. When reviewing code:
1. Check for security vulnerabilities
2. Verify code style and best practices
3. Suggest optimizations
4. Ensure proper error handling
Always be constructive and educational in your feedback."""
}

created_agent = client.agents.create(agent_data)
print(f"Created code reviewer: {created_agent['uuid']}")

Update Agent

Update an existing agent’s configuration.

Basic Update

# Update agent settings
agent_id = "agent-uuid-here"

update_data = {
    "description": "Updated agent description",
    "model": "kubiya/claude-opus-4"
}

updated_agent = client.agents.update(agent_id, update_data)
print(f"Updated agent: {updated_agent['name']}")
print(f"New model: {updated_agent['model']}")

Add Skills to Agent

def add_skills_to_agent(agent_id: str, new_skills: list):
    """Add skills to an existing agent"""
    # Get current agent
    agent = client.agents.get(agent_id)

    # Get current skills
    current_skills = agent.get('skills', [])

    # Add new skills (avoid duplicates)
    updated_skills = list(set(current_skills + new_skills))

    # Update agent
    update_data = {"skills": updated_skills}
    updated_agent = client.agents.update(agent_id, update_data)

    return updated_agent

# Usage
new_skills = ["docker", "helm"]
updated = add_skills_to_agent("agent-uuid", new_skills)
print(f"Updated agent with {len(updated['skills'])} skills")

Update Agent Model

def upgrade_agent_model(agent_id: str, new_model: str):
    """Upgrade agent to a new model"""
    update_data = {"model": new_model}
    updated_agent = client.agents.update(agent_id, update_data)

    print(f"Upgraded agent to {new_model}")
    return updated_agent

# Usage
upgrade_agent_model("agent-uuid", "kubiya/claude-opus-4")

Enable/Disable Agent

def toggle_agent_status(agent_id: str, status: str):
    """Enable or disable an agent"""
    if status not in ["active", "inactive"]:
        raise ValueError("Status must be 'active' or 'inactive'")

    update_data = {"status": status}
    updated_agent = client.agents.update(agent_id, update_data)

    print(f"Agent status set to: {status}")
    return updated_agent

# Usage
toggle_agent_status("agent-uuid", "inactive")  # Disable
toggle_agent_status("agent-uuid", "active")    # Enable

Delete Agent

Delete an agent.

Deleting an agent is permanent and cannot be undone. Ensure the agent is no longer in use before deletion.

# Delete agent
agent_id = "agent-uuid-to-delete"

result = client.agents.delete(agent_id)
print(f"Deletion result: {result}")

Delete with Safety Check

from kubiya.resources.exceptions import AgentError

def delete_agent_safe(agent_id: str):
    """Delete agent with confirmation"""
    try:
        # Get agent details first
        agent = client.agents.get(agent_id)

        print(f"About to delete agent: {agent['name']}")
        print(f"Description: {agent.get('description')}")
        print(f"Runtime: {agent.get('runtime')}")

        # Perform deletion
        result = client.agents.delete(agent_id)

        return {
            "success": True,
            "message": f"Deleted agent: {agent['name']}",
            "result": result
        }
    except AgentError as e:
        return {
            "success": False,
            "message": str(e)
        }

# Usage
result = delete_agent_safe("agent-uuid")
if result['success']:
    print(f"✅ {result['message']}")
else:
    print(f"❌ Deletion failed: {result['message']}")

Execute Agent

Execute an agent via Temporal workflow.

Basic Execution

# Execute an agent
execution_data = {
    "prompt": "Check the health of all production services"
}

execution = client.agents.execute(
    agent_id="agent-uuid",
    execution_data=execution_data
)

print(f"Execution started: {execution['execution_id']}")
print(f"Status: {execution.get('status')}")

Execute with Parameters

# Execute agent with custom parameters
execution_data = {
    "prompt": "Deploy application to production",
    "parameters": {
        "environment": "production",
        "version": "2.1.0",
        "rollback_enabled": True
    }
}

execution = client.agents.execute(
    agent_id="agent-uuid",
    execution_data=execution_data
)

print(f"Execution ID: {execution['execution_id']}")
print(f"Workflow ID: {execution.get('workflow_id')}")

Execute with Callback

# Execute agent with callback URL
execution_data = {
    "prompt": "Perform security scan",
    "callback_url": "https://api.example.com/webhooks/agent-callback",
    "metadata": {
        "scan_type": "full",
        "priority": "high"
    }
}

execution = client.agents.execute(
    agent_id="agent-uuid",
    execution_data=execution_data
)

print(f"Execution started: {execution['execution_id']}")
print(f"Callback will be sent to: {execution_data['callback_url']}")

Practical Examples

The following examples show how to use the Agents Service for common real-world scenarios, such as creating, updating, and executing agents. Each example includes a short explanation of when and why you might use it.

1. Agent Inventory Report

Use this pattern to generate a high-level inventory of all agents in your organization so you can understand coverage by runtime, model, skills, and integrations.

def generate_agent_inventory():
    """Generate comprehensive agent inventory"""
    all_agents = []
    skip = 0
    limit = 100

    while True:
        agents = client.agents.list(skip=skip, limit=limit)
        if not agents:
            break
        all_agents.extend(agents)
        skip += limit

    inventory = {
        "total_agents": len(all_agents),
        "by_runtime": {},
        "by_status": {},
        "by_model": {},
        "skills_usage": {},
        "integration_usage": {}
    }

    for agent in all_agents:
        # Count by runtime
        runtime = agent.get('runtime', 'unknown')
        inventory['by_runtime'][runtime] = inventory['by_runtime'].get(runtime, 0) + 1

        # Count by status
        status = agent.get('status', 'active')
        inventory['by_status'][status] = inventory['by_status'].get(status, 0) + 1

        # Count by model
        model = agent.get('model', 'unknown')
        inventory['by_model'][model] = inventory['by_model'].get(model, 0) + 1

        # Track skill usage
        for skill in agent.get('skills', []):
            inventory['skills_usage'][skill] = inventory['skills_usage'].get(skill, 0) + 1

        # Track integration usage
        for integration in agent.get('integrations', []):
            inventory['integration_usage'][integration] = inventory['integration_usage'].get(integration, 0) + 1

    return inventory

# Usage
inventory = generate_agent_inventory()
print(f"Agent Inventory Report:")
print(f"  Total Agents: {inventory['total_agents']}")
print(f"\nBy Runtime:")
for runtime, count in inventory['by_runtime'].items():
    print(f"  {runtime}: {count}")
print(f"\nTop 5 Skills:")
top_skills = sorted(inventory['skills_usage'].items(), key=lambda x: x[1], reverse=True)[:5]
for skill, count in top_skills:
    print(f"  {skill}: {count} agents")

2. Agent Configuration Validator

Use this validator before creating or updating agents to catch missing fields or questionable configurations early, instead of discovering issues at execution time.

def validate_agent_config(agent_config: dict):
    """Validate agent configuration before creation"""
    errors = []
    warnings = []

    # Required fields
    if not agent_config.get('name'):
        errors.append("Agent name is required")

    if not agent_config.get('runtime'):
        errors.append("Runtime is required")

    if not agent_config.get('model'):
        errors.append("Model is required")

    # Validate runtime
    valid_runtimes = ["agno", "claude_code"]
    if agent_config.get('runtime') not in valid_runtimes:
        errors.append(f"Runtime must be one of: {valid_runtimes}")

    # Validate skills
    if not agent_config.get('skills'):
        warnings.append("No skills specified - agent may have limited capabilities")

    # Validate integrations
    if not agent_config.get('integrations'):
        warnings.append("No integrations specified")

    return {
        "valid": len(errors) == 0,
        "errors": errors,
        "warnings": warnings
    }

# Usage
agent_config = {
    "name": "my-agent",
    "runtime": "agno",
    "model": "kubiya/claude-sonnet-4"
}

validation = validate_agent_config(agent_config)
if validation['valid']:
    print("✅ Configuration is valid")
    if validation['warnings']:
        print("⚠️  Warnings:")
        for warning in validation['warnings']:
            print(f"  - {warning}")
    # Create agent
    agent = client.agents.create(agent_config)
else:
    print("❌ Configuration is invalid:")
    for error in validation['errors']:
        print(f"  - {error}")

3. Agent Cloning

Use cloning when you want to spin up a new agent that is similar to an existing one (for a new team, environment, or experiment) while safely applying only a small set of changes.

def clone_agent(source_agent_id: str, new_name: str, modifications: dict = None):
    """Clone an existing agent with optional modifications"""
    # Get source agent
    source_agent = client.agents.get(source_agent_id)

    # Create new agent config
    new_config = {
        "name": new_name,
        "description": source_agent.get('description', '') + " (Cloned)",
        "runtime": source_agent.get('runtime'),
        "model": source_agent.get('model'),
        "skills": source_agent.get('skills', []),
        "integrations": source_agent.get('integrations', []),
        "policies": source_agent.get('policies', []),
        "environment": source_agent.get('environment', {}),
        "instructions": source_agent.get('instructions')
    }

    # Apply modifications
    if modifications:
        new_config.update(modifications)

    # Create cloned agent
    cloned_agent = client.agents.create(new_config)

    return cloned_agent

# Usage
cloned = clone_agent(
    "existing-agent-uuid",
    "cloned-devops-agent",
    modifications={"model": "kubiya/claude-opus-4"}
)
print(f"Cloned agent created: {cloned['uuid']}")

4. Bulk Agent Operations

Use bulk updates when you need to roll out the same change (such as a new default model or status) across many agents in a controlled way.

def bulk_update_agents(agent_ids: list, update_data: dict):
    """Update multiple agents with the same configuration"""
    results = {
        "success": [],
        "failed": []
    }

    for agent_id in agent_ids:
        try:
            updated = client.agents.update(agent_id, update_data)
            results['success'].append({
                "agent_id": agent_id,
                "name": updated['name']
            })
        except Exception as e:
            results['failed'].append({
                "agent_id": agent_id,
                "error": str(e)
            })

    return results

# Usage - upgrade all agents to new model
agent_ids = ["agent-1", "agent-2", "agent-3"]
update_data = {"model": "kubiya/claude-opus-4"}

results = bulk_update_agents(agent_ids, update_data)
print(f"✅ Updated: {len(results['success'])}")
print(f"❌ Failed: {len(results['failed'])}")

Error Handling

Agent operations can fail if an agent does not exist, has an invalid configuration, or if there are connectivity or permission issues when executing workflows. The following examples show how to catch AgentError, surface useful information, and fall back to safer behavior like listing available agents.

from kubiya.resources.exceptions import AgentError

try:
    # Try to get an agent
    agent = client.agents.get("non-existent-agent")
except AgentError as e:
    print(f"Agent error: {e}")
    # Handle error - maybe list all agents
    print("Available agents:")
    agents = client.agents.list(limit=10)
    for agent in agents:
        print(f"  - {agent['name']}")

# Execute with error handling
try:
    execution = client.agents.execute(
        agent_id="agent-uuid",
        execution_data={"prompt": "Deploy app"}
    )
    print(f"Execution started: {execution['execution_id']}")
except AgentError as e:
    print(f"Execution failed: {e}")

Best Practices

1. Use Descriptive Names

Give agents clear, descriptive names and descriptions so platform teams and end users can quickly understand what each agent is responsible for.

# Good - descriptive names
agent_data = {
    "name": "production-kubernetes-operator",
    "description": "Manages Kubernetes clusters in production environment",
    "runtime": "agno"
}

# Bad - vague names
agent_data = {
    "name": "agent1",
    "description": "Some agent",
    "runtime": "agno"
}

2. Validate Configuration Before Creation

Always validate agent configuration before creating it in the Control Plane so you avoid half-configured agents and confusing runtime errors.

def create_agent_safely(agent_config: dict):
    """Create agent with pre-validation"""
    # Validate first
    validation = validate_agent_config(agent_config)

    if not validation['valid']:
        raise ValueError(f"Invalid configuration: {validation['errors']}")

    # Create agent
    agent = client.agents.create(agent_config)
    return agent

# Usage
try:
    agent = create_agent_safely(agent_config)
    print(f"✅ Created agent: {agent['uuid']}")
except ValueError as e:
    print(f"❌ {e}")

3. Use Appropriate Models for Tasks

Choose runtimes and models that match the kind of work the agent will perform, balancing capability and cost for code-heavy versus general-purpose scenarios.

# For code-heavy tasks
code_agent = {
    "name": "code-assistant",
    "runtime": "claude_code",
    "model": "kubiya/claude-opus-4",  # Best for code
    "skills": ["github", "code-analysis"]
}

# For general tasks
general_agent = {
    "name": "general-assistant",
    "runtime": "agno",
    "model": "kubiya/claude-sonnet-4",  # Cost-effective
    "skills": ["knowledge-base", "ticketing"]
}

4. Handle Agent Lifecycle

Treat agents as long-lived resources with a lifecycle: pause when not in use, resume when needed, and retire carefully after archiving any important state or configuration.

def manage_agent_lifecycle(agent_id: str, action: str):
    """Manage agent through its lifecycle"""
    if action == "pause":
        client.agents.update(agent_id, {"status": "inactive"})
        print(f"Agent paused")
    elif action == "resume":
        client.agents.update(agent_id, {"status": "active"})
        print(f"Agent resumed")
    elif action == "retire":
        # Archive agent data before deletion
        agent = client.agents.get(agent_id)
        print(f"Archiving agent: {agent['name']}")
        # ... archive logic ...
        client.agents.delete(agent_id)
        print(f"Agent retired")

# Usage
manage_agent_lifecycle("agent-uuid", "pause")

API Reference

Methods

Method	Description	Parameters
`list(skip, limit, status_filter)`	List all agents	`skip`: Records to skip, `limit`: Max records, `status_filter`: Optional status
`get(agent_id)`	Get specific agent	`agent_id`: Agent UUID
`create(agent_data)`	Create agent	`agent_data`: Agent configuration dictionary
`update(agent_id, agent_data)`	Update agent	`agent_id`: UUID, `agent_data`: Updates
`delete(agent_id)`	Delete agent	`agent_id`: Agent UUID
`execute(agent_id, execution_data)`	Execute agent	`agent_id`: UUID, `execution_data`: Execution configuration

Agent Object Structure

{
    "uuid": "string",                    # Unique identifier
    "name": "string",                    # Agent name
    "description": "string",             # Agent description
    "status": "string",                  # Agent status (active, inactive)
    "runtime": "string",                 # Runtime type (agno, claude_code)
    "model": "string",                   # LLM model (e.g., kubiya/claude-sonnet-4)
    "skills": ["string"],                # List of skill names
    "integrations": ["string"],          # List of integration names
    "policies": ["string"],              # List of policy names
    "environment": dict,                 # Environment variables
    "instructions": "string",            # Custom system instructions
    "created_at": "string",              # Creation timestamp
    "updated_at": "string"               # Last update timestamp
}

Execution Object Structure

{
    "execution_id": "string",            # Unique execution identifier
    "workflow_id": "string",             # Temporal workflow ID
    "status": "string",                  # Execution status
    "agent_id": "string",                # Agent UUID
    "prompt": "string",                  # Execution prompt
    "parameters": dict,                  # Execution parameters
    "callback_url": "string",            # Optional callback URL
    "metadata": dict,                    # Additional metadata
    "created_at": "string"               # Execution start time
}

Next Steps

Skills Service

Manage agent skills and capabilities

Policies Service

Configure agent policies

Runtimes Service

Configure agent runtimes

​Overview

​Quick Start

​List Agents

​Basic Listing

​Paginated Listing

​Filter by Status

​List All Agents

​Get Agent Details

​By Agent ID

​Extract Agent Information

​Find Agent by Name

​Create Agent

​Basic Agent Creation

​Agent with Environment Variables

​Agent with Policies

​Agent with Custom Instructions

​Update Agent

​Basic Update

​Add Skills to Agent

​Update Agent Model

​Enable/Disable Agent

​Delete Agent

​Delete with Safety Check

​Execute Agent

​Basic Execution

​Execute with Parameters

​Execute with Callback

​Practical Examples

​1. Agent Inventory Report

​2. Agent Configuration Validator

​3. Agent Cloning

​4. Bulk Agent Operations

​Error Handling

​Best Practices

​1. Use Descriptive Names

​2. Validate Configuration Before Creation

​3. Use Appropriate Models for Tasks

​4. Handle Agent Lifecycle

​API Reference

​Methods

​Agent Object Structure

​Execution Object Structure

​Next Steps

Skills Service

Policies Service

Runtimes Service

Overview

Quick Start

List Agents

Basic Listing

Paginated Listing

Filter by Status

List All Agents

Get Agent Details

By Agent ID

Extract Agent Information

Find Agent by Name

Create Agent

Basic Agent Creation

Agent with Environment Variables

Agent with Policies

Agent with Custom Instructions

Update Agent

Basic Update

Add Skills to Agent

Update Agent Model

Enable/Disable Agent

Delete Agent

Delete with Safety Check

Execute Agent

Basic Execution

Execute with Parameters

Execute with Callback

Practical Examples

1. Agent Inventory Report

2. Agent Configuration Validator

3. Agent Cloning

4. Bulk Agent Operations

Error Handling

Best Practices

1. Use Descriptive Names

2. Validate Configuration Before Creation

3. Use Appropriate Models for Tasks

4. Handle Agent Lifecycle

API Reference

Methods

Agent Object Structure

Execution Object Structure

Next Steps