The Skills Service gives you programmatic control over the capabilities and tool sets that agents can use on the Kubiya platform. Skills are collections of tools and permissions that define what an agent can do, such as managing cloud resources, running scripts, or integrating with external systems. By managing skills, you can tailor agent abilities to your organization’s needs, enforce best practices, and ensure agents only have access to the tools they require.
With this service, you can:
- List and filter available skills to see what capabilities are available for your agents, including their categories and tool sets.
- Retrieve detailed skill information to understand what each skill enables, which tools it includes, and any required integrations.
- Create and update custom skills to define new sets of capabilities or modify existing ones as your requirements evolve.
- Validate skill configurations before deployment, ensuring your skills are correctly set up and ready for use.
- Associate skills with agents, teams, or other entities to control who can use which capabilities.
This service is especially valuable for platform administrators and advanced users who want to manage agent permissions, standardize tool usage, and maintain security and compliance. By using the Skills Service, you can confidently control what agents can do, avoid misconfigurations, and streamline the process of updating agent capabilities.
Overview
The Skills Service provides a set of high-level methods designed to be intuitive and flexible, supporting a wide range of operational and administrative tasks:
- list(skip, limit): Retrieve all available skills, with support for pagination. This is the primary entry point for discovering which skills are available in your workspace.
- get(skill_id): Fetch detailed information for a specific skill, including its tools, capabilities, and configuration. Use this to inspect or audit skills before assigning them to agents.
- create(skill_data): Register a new custom skill. This is typically used to define new sets of capabilities or tool sets for your organization.
- update(skill_id, skill_data): Modify the configuration of an existing skill, such as updating its description, tools, or required integrations.
- delete(skill_id): Remove a skill from the registry. This operation is restricted to ensure that no active agents or teams are using the skill at the time of deletion.
- associate(entity_id, skill_ids): Assign skills to agents, teams, or other entities to control access to capabilities.
- validate(skill_config): Validate a skill configuration before deployment. This helps catch errors early and ensures your skills are ready for use.
By using these methods, you can build robust, dynamic workflows that adapt to changes in skill requirements, enforce organizational policies, and provide a seamless experience for both developers and end-users.
The following sections provide practical guidance, detailed examples, and best practices for leveraging the Skills Service effectively in your own projects.
Quick Start
from kubiya import ControlPlaneClient
# Initialize client
client = ControlPlaneClient(api_key="your-api-key")
# List all available skills
skills = client.skills.list()
for skill in skills:
print(f"Skill: {skill['name']} - Category: {skill.get('category', 'N/A')}")
# Get specific skill details
skill = client.skills.get("skill-uuid")
print(f"Tools: {skill.get('tools', [])}")
List Skills
List all available skills with pagination support.
Basic Listing
# List all skills (first 100)
skills = client.skills.list()
for skill in skills:
print(f"""
Name: {skill['name']}
Category: {skill.get('category', 'N/A')}
Description: {skill.get('description', 'N/A')}
Tools: {len(skill.get('tools', []))} tools
""")
Paginated Listing
# List skills with pagination
skip = 0
limit = 20
while True:
skills = client.skills.list(skip=skip, limit=limit)
if not skills:
break
for skill in skills:
print(f"Skill: {skill['name']}")
skip += limit
List All Skills
def list_all_skills():
"""Get all skills across all pages"""
all_skills = []
skip = 0
limit = 100
while True:
skills = client.skills.list(skip=skip, limit=limit)
if not skills:
break
all_skills.extend(skills)
skip += limit
return all_skills
# Usage
all_skills = list_all_skills()
print(f"Total skills: {len(all_skills)}")
Filter Skills by Category
def filter_skills_by_category(category: str):
"""Filter skills by category"""
all_skills = client.skills.list(limit=1000) # Get all skills
return [s for s in all_skills if s.get('category') == category]
# Usage
cloud_skills = filter_skills_by_category("Cloud")
print(f"Cloud skills: {len(cloud_skills)}")
devops_skills = filter_skills_by_category("DevOps")
print(f"DevOps skills: {len(devops_skills)}")
Get Skill Details
Retrieve detailed information about a specific skill.
By Skill ID
# Get skill by UUID
skill = client.skills.get("skill-uuid-here")
print(f"Skill: {skill['name']}")
print(f"Description: {skill.get('description', 'N/A')}")
print(f"Category: {skill.get('category', 'N/A')}")
print(f"Tools: {skill.get('tools', [])}")
print(f"Capabilities: {skill.get('capabilities', [])}")
def get_skill_info(skill_id: str):
"""Get comprehensive skill information"""
try:
skill = client.skills.get(skill_id)
info = {
"id": skill.get('uuid'),
"name": skill['name'],
"description": skill.get('description'),
"category": skill.get('category'),
"tools": skill.get('tools', []),
"capabilities": skill.get('capabilities', []),
"required_integrations": skill.get('required_integrations', []),
"configuration": skill.get('configuration', {}),
"enabled": skill.get('enabled', True)
}
return info
except Exception as e:
print(f"Failed to get skill info: {e}")
return None
# Usage
info = get_skill_info("skill-uuid")
if info:
print(f"Skill: {info['name']}")
print(f"Tools: {len(info['tools'])}")
print(f"Capabilities: {info['capabilities']}")
Find Skill by Name
def find_skill_by_name(name: str):
"""Find a skill by name"""
all_skills = client.skills.list(limit=1000)
for skill in all_skills:
if skill['name'].lower() == name.lower():
return skill
return None
# Usage
kubernetes_skill = find_skill_by_name("kubernetes")
if kubernetes_skill:
print(f"Found skill: {kubernetes_skill['uuid']}")
skill_details = client.skills.get(kubernetes_skill['uuid'])
print(f"Tools: {skill_details.get('tools', [])}")
Create Custom Skill
Create a new custom skill with specific tools and capabilities.
Basic Creation
# Create a custom skill
skill_data = {
"name": "custom-database-admin",
"description": "Database administration tools",
"category": "Database",
"tools": [
{
"name": "query_database",
"description": "Execute database queries",
"type": "shell"
},
{
"name": "backup_database",
"description": "Create database backups",
"type": "shell"
}
],
"capabilities": ["database_query", "database_backup"],
"enabled": True
}
created_skill = client.skills.create(skill_data)
print(f"Created skill: {created_skill['uuid']}")
print(f"Name: {created_skill['name']}")
Create with Integrations
# Create skill with required integrations
skill_data = {
"name": "aws-infrastructure",
"description": "AWS infrastructure management",
"category": "Cloud",
"tools": [
{
"name": "describe_instances",
"description": "List EC2 instances",
"type": "shell",
"command": "aws ec2 describe-instances"
},
{
"name": "list_s3_buckets",
"description": "List S3 buckets",
"type": "shell",
"command": "aws s3 ls"
}
],
"required_integrations": ["AWS"],
"capabilities": ["cloud_management", "infrastructure_automation"],
"configuration": {
"region": "us-east-1",
"output_format": "json"
}
}
created_skill = client.skills.create(skill_data)
print(f"Created AWS skill: {created_skill['uuid']}")
Create with Validation
from kubiya.resources.exceptions import SkillError
def create_skill_safe(skill_data: dict):
"""Create skill with pre-validation"""
try:
# Validate configuration first
validation = client.skills.validate(skill_data)
if not validation.get('valid'):
return {
"success": False,
"errors": validation.get('errors', [])
}
# Create skill
skill = client.skills.create(skill_data)
return {
"success": True,
"skill_id": skill['uuid'],
"skill": skill
}
except SkillError as e:
return {
"success": False,
"errors": [str(e)]
}
# Usage
skill_data = {
"name": "custom-monitoring",
"description": "System monitoring tools",
"category": "Monitoring",
"tools": [{"name": "check_health", "type": "shell"}]
}
result = create_skill_safe(skill_data)
if result['success']:
print(f"✅ Created skill: {result['skill_id']}")
else:
print(f"❌ Creation failed: {result['errors']}")
Update Skill
Update an existing skill configuration.
Basic Update
# Update skill settings
skill_id = "skill-uuid-here"
update_data = {
"description": "Updated description",
"enabled": True,
"tools": [
{
"name": "new_tool",
"description": "A new tool",
"type": "shell"
}
]
}
updated_skill = client.skills.update(skill_id, update_data)
print(f"Updated skill: {updated_skill['name']}")
print(f"Description: {updated_skill['description']}")
def add_tools_to_skill(skill_id: str, new_tools: list):
"""Add tools to an existing skill"""
# Get current skill
skill = client.skills.get(skill_id)
# Get current tools
current_tools = skill.get('tools', [])
# Add new tools
updated_tools = current_tools + new_tools
# Update skill
update_data = {"tools": updated_tools}
updated_skill = client.skills.update(skill_id, update_data)
return updated_skill
# Usage
new_tools = [
{
"name": "analyze_logs",
"description": "Analyze system logs",
"type": "python"
}
]
updated = add_tools_to_skill("skill-uuid", new_tools)
print(f"Updated skill with {len(updated['tools'])} tools")
Enable/Disable Skill
def toggle_skill(skill_id: str, enabled: bool):
"""Enable or disable a skill"""
update_data = {"enabled": enabled}
updated_skill = client.skills.update(skill_id, update_data)
status = "enabled" if enabled else "disabled"
print(f"Skill {skill_id} is now {status}")
return updated_skill
# Usage
toggle_skill("skill-uuid", False) # Disable
toggle_skill("skill-uuid", True) # Enable
Delete Skill
Delete a custom skill.
Deleting a skill affects any agents or entities using it. Ensure no active agents depend on the skill before deletion.
# Delete skill
skill_id = "skill-uuid-to-delete"
result = client.skills.delete(skill_id)
print(f"Deletion result: {result}")
Delete with Safety Check
def delete_skill_safe(skill_id: str):
"""Delete skill with confirmation"""
try:
# Get skill details first
skill = client.skills.get(skill_id)
print(f"About to delete skill: {skill['name']}")
print(f"Category: {skill.get('category')}")
# Perform deletion
result = client.skills.delete(skill_id)
return {
"success": True,
"message": f"Deleted skill: {skill['name']}",
"result": result
}
except SkillError as e:
return {
"success": False,
"message": str(e)
}
# Usage
result = delete_skill_safe("skill-uuid")
if result['success']:
print(f"✅ {result['message']}")
else:
print(f"❌ Deletion failed: {result['message']}")
Associate Skills with Entities
Associate skills with agents, teams, or other entities.
Associate Skills with Agent
# Associate skills with an agent
agent_id = "agent-uuid"
skill_ids = ["skill-uuid-1", "skill-uuid-2", "skill-uuid-3"]
result = client.skills.associate(agent_id, skill_ids)
print(f"Association result: {result}")
Bulk Association
def associate_skills_bulk(entity_id: str, skill_names: list):
"""Associate skills by name"""
# Find skill IDs by name
all_skills = client.skills.list(limit=1000)
skill_ids = []
for name in skill_names:
for skill in all_skills:
if skill['name'].lower() == name.lower():
skill_ids.append(skill['uuid'])
break
if len(skill_ids) != len(skill_names):
print(f"Warning: Only found {len(skill_ids)} of {len(skill_names)} skills")
# Associate skills
result = client.skills.associate(entity_id, skill_ids)
return result
# Usage
skill_names = ["kubernetes", "terraform", "aws"]
result = associate_skills_bulk("agent-uuid", skill_names)
print(f"Associated {len(skill_names)} skills")
Replace Skills for Entity
def replace_entity_skills(entity_id: str, new_skill_ids: list):
"""Replace all skills for an entity"""
# Associate new skills (this replaces existing associations)
result = client.skills.associate(entity_id, new_skill_ids)
print(f"Replaced skills for entity {entity_id}")
print(f"New skill count: {len(new_skill_ids)}")
return result
# Usage
new_skills = ["skill-uuid-1", "skill-uuid-2"]
replace_entity_skills("agent-uuid", new_skills)
Validate Skill Configuration
Validate a skill configuration before creation or update.
Basic Validation
# Validate skill configuration
skill_config = {
"name": "test-skill",
"description": "Test skill for validation",
"category": "Testing",
"tools": [
{
"name": "test_tool",
"description": "A test tool",
"type": "shell"
}
],
"capabilities": ["testing"]
}
validation_result = client.skills.validate(skill_config)
if validation_result.get('valid'):
print("✅ Configuration is valid!")
else:
print("❌ Configuration is invalid:")
for error in validation_result.get('errors', []):
print(f" - {error}")
Validate Before Creation
def create_skill_with_validation(skill_data: dict):
"""Create skill with automatic validation"""
# Validate first
validation = client.skills.validate(skill_data)
if not validation.get('valid'):
return {
"success": False,
"message": "Validation failed",
"errors": validation.get('errors', [])
}
# Create skill
try:
skill = client.skills.create(skill_data)
return {
"success": True,
"message": "Skill created successfully",
"skill_id": skill['uuid'],
"skill": skill
}
except SkillError as e:
return {
"success": False,
"message": "Creation failed",
"errors": [str(e)]
}
# Usage
skill_data = {
"name": "monitoring-alerts",
"category": "Monitoring",
"tools": [{"name": "send_alert", "type": "shell"}]
}
result = create_skill_with_validation(skill_data)
if result['success']:
print(f"✅ {result['message']}: {result['skill_id']}")
else:
print(f"❌ {result['message']}")
for error in result['errors']:
print(f" - {error}")
Practical Examples
The following examples show how to use the Skills Service for common real-world scenarios, such as discovering skills that match specific requirements, auditing your skill inventory, and managing skill configurations. Each example includes a short explanation of when and why you might use it.
1. Skill Discovery System
Use this approach to find skills that match a set of required capabilities or categories. This is helpful when you want to assign the most relevant skills to an agent or team based on their responsibilities.
def discover_skills(requirements: dict):
"""Discover skills matching requirements"""
all_skills = client.skills.list(limit=1000)
required_capabilities = set(requirements.get('capabilities', []))
required_category = requirements.get('category')
matching_skills = []
for skill in all_skills:
skill_capabilities = set(skill.get('capabilities', []))
skill_category = skill.get('category')
# Check category match
if required_category and skill_category != required_category:
continue
# Check capabilities match
if required_capabilities and not required_capabilities.issubset(skill_capabilities):
continue
# Calculate match score
match_score = len(required_capabilities & skill_capabilities)
matching_skills.append({
"skill": skill,
"match_score": match_score
})
# Sort by match score
matching_skills.sort(key=lambda x: x['match_score'], reverse=True)
return matching_skills
# Usage
requirements = {
"capabilities": ["cloud_management", "automation"],
"category": "Cloud"
}
matches = discover_skills(requirements)
print(f"Found {len(matches)} matching skills:")
for match in matches[:5]: # Top 5
skill = match['skill']
print(f" - {skill['name']} (score: {match['match_score']})")
2. Skill Audit Report
Generate a summary report of all skills in your environment, including how many are enabled, disabled, or missing descriptions. This is useful for platform administrators who need to audit skill inventory or prepare compliance reports.
def generate_skill_audit():
"""Generate audit report of all skills"""
all_skills = client.skills.list(limit=1000)
audit = {
"total_skills": len(all_skills),
"enabled": 0,
"disabled": 0,
"by_category": {},
"tool_count": 0,
"skills_without_description": []
}
for skill in all_skills:
# Count enabled/disabled
if skill.get('enabled', True):
audit['enabled'] += 1
else:
audit['disabled'] += 1
# Count by category
category = skill.get('category', 'Uncategorized')
audit['by_category'][category] = audit['by_category'].get(category, 0) + 1
# Count tools
audit['tool_count'] += len(skill.get('tools', []))
# Track skills without description
if not skill.get('description'):
audit['skills_without_description'].append(skill['name'])
return audit
# Usage
audit = generate_skill_audit()
print(f"Skill Audit Report:")
print(f" Total Skills: {audit['total_skills']}")
print(f" Enabled: {audit['enabled']}")
print(f" Disabled: {audit['disabled']}")
print(f" Total Tools: {audit['tool_count']}")
print(f"\nBy Category:")
for category, count in audit['by_category'].items():
print(f" {category}: {count}")
3. Skill Configuration Manager
Use this helper class to create, validate, and clone skill configurations efficiently. This is valuable for organizations that want to standardize skill creation or quickly replicate existing skills for new use cases.
class SkillConfigManager:
"""Helper class for managing skill configurations"""
def __init__(self, client):
self.client = client
def create_from_template(self, template_name: str, custom_config: dict):
"""Create skill from template"""
templates = {
"monitoring": {
"category": "Monitoring",
"capabilities": ["monitoring", "alerting"],
"tools": [
{"name": "check_status", "type": "shell"},
{"name": "send_alert", "type": "shell"}
]
},
"deployment": {
"category": "DevOps",
"capabilities": ["deployment", "automation"],
"tools": [
{"name": "deploy_app", "type": "shell"},
{"name": "rollback", "type": "shell"}
]
}
}
if template_name not in templates:
raise ValueError(f"Unknown template: {template_name}")
# Merge template with custom config
skill_config = {**templates[template_name], **custom_config}
# Validate
validation = self.client.skills.validate(skill_config)
if not validation.get('valid'):
raise ValueError(f"Invalid configuration: {validation.get('errors')}")
# Create skill
skill = self.client.skills.create(skill_config)
return skill
def clone_skill(self, source_skill_id: str, new_name: str):
"""Clone an existing skill"""
# Get source skill
source = self.client.skills.get(source_skill_id)
# Create new skill config
new_config = {
"name": new_name,
"description": f"Cloned from {source['name']}",
"category": source.get('category'),
"tools": source.get('tools', []),
"capabilities": source.get('capabilities', []),
"configuration": source.get('configuration', {})
}
# Create new skill
skill = self.client.skills.create(new_config)
return skill
# Usage
manager = SkillConfigManager(client)
# Create from template
new_skill = manager.create_from_template("monitoring", {
"name": "production-monitoring",
"description": "Production environment monitoring"
})
print(f"Created from template: {new_skill['uuid']}")
# Clone existing skill
cloned = manager.clone_skill("existing-skill-uuid", "cloned-skill-name")
print(f"Cloned skill: {cloned['uuid']}")
4. Skill Dependencies Validator
Check that all required integrations for a skill are available before deployment. This helps prevent runtime errors and ensures your skills are ready to use in your environment.
def validate_skill_dependencies(skill_config: dict):
"""Validate that all required integrations are available"""
required_integrations = skill_config.get('required_integrations', [])
if not required_integrations:
return {"valid": True, "message": "No dependencies required"}
# Get available integrations
available_integrations = client.integrations.list()
available_names = [i['name'] for i in available_integrations]
missing = []
for required in required_integrations:
if required not in available_names:
missing.append(required)
if missing:
return {
"valid": False,
"message": "Missing required integrations",
"missing": missing
}
return {
"valid": True,
"message": "All dependencies available",
"required": required_integrations
}
# Usage
skill_config = {
"name": "aws-ops",
"required_integrations": ["AWS", "Slack"]
}
deps_check = validate_skill_dependencies(skill_config)
if deps_check['valid']:
print(f"✅ {deps_check['message']}")
else:
print(f"❌ {deps_check['message']}")
print(f"Missing: {deps_check['missing']}")
Error Handling
When working with skills, you may encounter errors such as requesting a non-existent skill, creating a skill with missing fields, or failing validation. These examples show how to handle such errors gracefully and provide fallback options to keep your workflows running smoothly.
from kubiya.resources.exceptions import SkillError
try:
# Try to get a skill
skill = client.skills.get("non-existent-skill")
except SkillError as e:
print(f"Skill error: {e}")
# Handle error - maybe list all skills instead
print("Available skills:")
skills = client.skills.list(limit=10)
for skill in skills:
print(f" - {skill['name']}")
# Create with error handling
try:
skill_data = {
"name": "test-skill",
"category": "Testing"
}
skill = client.skills.create(skill_data)
print(f"Created skill: {skill['uuid']}")
except SkillError as e:
print(f"Creation failed: {e}")
# Validate with error handling
try:
validation = client.skills.validate(skill_data)
if not validation.get('valid'):
print(f"Validation errors: {validation.get('errors')}")
except SkillError as e:
print(f"Validation error: {e}")
Best Practices
Follow these best practices to make your use of the Skills Service more robust, efficient, and maintainable. These patterns help you avoid common pitfalls and ensure your code is resilient to changes in skill availability or configuration.
1. Always Validate Before Creating
Always validate your skill configuration before creating or updating a skill. This helps catch errors early and ensures your skills are correctly set up for use by agents and teams.
# Always validate skill configuration before creating
def create_skill_safely(skill_data: dict):
"""Create skill with validation"""
# Validate first
validation = client.skills.validate(skill_data)
if not validation.get('valid'):
raise ValueError(f"Invalid skill configuration: {validation.get('errors')}")
# Create skill
skill = client.skills.create(skill_data)
return skill
# Usage
skill_data = {
"name": "my-skill",
"category": "Custom",
"tools": [{"name": "tool1", "type": "shell"}]
}
try:
skill = create_skill_safely(skill_data)
print(f"✅ Created skill: {skill['uuid']}")
except ValueError as e:
print(f"❌ {e}")
# Cache skills to reduce API calls
class SkillCache:
def __init__(self, client):
self.client = client
self._skills_cache = None
self._skill_details_cache = {}
def get_all_skills(self, force_refresh=False):
"""Get cached skills"""
if self._skills_cache is None or force_refresh:
self._skills_cache = self.client.skills.list(limit=1000)
return self._skills_cache
def get_skill(self, skill_id: str, force_refresh=False):
"""Get cached skill details"""
if skill_id not in self._skill_details_cache or force_refresh:
self._skill_details_cache[skill_id] = self.client.skills.get(skill_id)
return self._skill_details_cache[skill_id]
def invalidate_cache(self):
"""Clear the cache"""
self._skills_cache = None
self._skill_details_cache = {}
# Usage
cache = SkillCache(client)
skills = cache.get_all_skills() # API call
skills_again = cache.get_all_skills() # From cache
3. Use Descriptive Names and Categories
Choose clear, descriptive names and categories for your skills. This makes it easier for others to understand what each skill does and helps with auditing and compliance.
# Good - descriptive names
skill_data = {
"name": "aws-ec2-management",
"description": "Manage AWS EC2 instances including start, stop, and monitoring",
"category": "Cloud/AWS",
"tools": [...]
}
# Bad - vague names
skill_data = {
"name": "skill1",
"description": "Does stuff",
"category": "Other",
"tools": [...]
}
# Include clear descriptions for all tools
skill_data = {
"name": "database-admin",
"category": "Database",
"tools": [
{
"name": "query_database",
"description": "Execute SQL queries against the database",
"type": "shell",
"command": "psql -c \"{{query}}\"",
"parameters": [
{
"name": "query",
"description": "SQL query to execute",
"required": True
}
]
}
]
}
API Reference
Methods
| Method | Description | Parameters |
|---|
list(skip, limit) | List all skills | skip: Records to skip, limit: Max records |
get(skill_id) | Get specific skill | skill_id: Skill UUID |
create(skill_data) | Create custom skill | skill_data: Skill configuration dictionary |
update(skill_id, skill_data) | Update skill | skill_id: UUID, skill_data: Updates |
delete(skill_id) | Delete skill | skill_id: Skill UUID |
associate(entity_id, skill_ids) | Associate skills with entity | entity_id: Entity UUID, skill_ids: List of skill UUIDs |
validate(skill_config) | Validate skill configuration | skill_config: Configuration dictionary |
Skill Object Structure
{
"uuid": "string", # Unique identifier
"name": "string", # Skill name
"description": "string", # Skill description
"category": "string", # Category (e.g., "Cloud", "DevOps")
"tools": [ # List of tools
{
"name": "string",
"description": "string",
"type": "string", # Tool type (shell, python, etc.)
"command": "string", # Command to execute
"parameters": [dict] # Tool parameters
}
],
"capabilities": ["string"], # List of capabilities
"required_integrations": ["string"], # Required integrations
"configuration": dict, # Skill configuration
"enabled": bool, # Whether skill is enabled
"created_at": "string", # Creation timestamp
"updated_at": "string" # Last update timestamp
}
Validation Result Object Structure
{
"valid": bool, # Whether configuration is valid
"errors": ["string"], # List of validation errors
"warnings": ["string"], # List of validation warnings
"validated_config": dict # Validated configuration
}
Next Steps
