Use this file to discover all available pages before exploring further.
The Runtimes Service gives you programmatic control over the environments in which your agents operate on the Kubiya platform. It acts as the main interface for discovering, configuring, and validating agent runtimes, whether you are using built-in types like agno and claude_code, or custom runtimes tailored to your needs.With this service, you can:
List and filter available runtimes to see what execution environments are available for your agents, including their features and status.
Retrieve detailed runtime requirements to understand what dependencies, features, or models are needed for each runtime type.
Validate runtime configurations before deploying agents, ensuring your setup meets all requirements and will work as expected.
This service is especially valuable for platform administrators and advanced users who want to ensure agents are deployed in the right environment, with all necessary features and compliance. By using the Runtimes Service, you can confidently manage agent execution, avoid misconfigurations, and streamline the deployment process.
For conceptual information about runtimes and how they’re used in agents, see Runtimes Core Concepts.
The Runtimes Service provides a set of high-level methods designed to be intuitive and flexible, supporting a wide range of operational and administrative tasks:
list(): Retrieve all available runtimes, with details about their type, features, and status. This is the primary entry point for discovering which execution environments are available in your workspace.
get_requirements(runtime_id): Fetch detailed requirements for a specific runtime, including required features, model compatibility, and environment settings. Use this to ensure your agent configuration matches the runtime’s needs.
validate(runtime_config): Validate a runtime configuration before deploying an agent. This helps catch errors early and ensures your deployment will succeed without unexpected issues.
By using these methods, you can build robust, dynamic workflows that adapt to changes in runtime availability, 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 Runtimes Service effectively in your own projects.
from kubiya import ControlPlaneClient# Initialize clientclient = ControlPlaneClient(api_key="your-api-key")# List all available runtimesruntimes = client.runtimes.list()for runtime in runtimes: print(f"Runtime: {runtime['name']} - Type: {runtime['type']}")# Get requirements for a specific runtimerequirements = client.runtimes.get_requirements("claude_code")print(f"Required features: {requirements}")
Use this helper when you want to automatically select the most suitable runtime for your agent based on required features or model provider. This is useful for dynamic agent deployment or when your requirements may change over time.
def select_best_runtime(requirements: dict): """Select the best runtime based on requirements""" runtimes = client.runtimes.list() required_features = requirements.get('features', []) model_preference = requirements.get('model_provider', '') suitable_runtimes = [] for runtime in runtimes: runtime_id = runtime.get('name') try: runtime_reqs = client.runtimes.get_requirements(runtime_id) runtime_features = runtime_reqs.get('supported_features', []) # Check if runtime supports all required features if all(feature in runtime_features for feature in required_features): suitable_runtimes.append({ "runtime_id": runtime_id, "runtime": runtime, "requirements": runtime_reqs, "match_score": len(set(required_features) & set(runtime_features)) }) except RuntimeError: continue # Sort by match score suitable_runtimes.sort(key=lambda x: x['match_score'], reverse=True) return suitable_runtimes[0] if suitable_runtimes else None# Usagerequirements = { "features": ["code_execution", "file_system_access"], "model_provider": "Anthropic"}best_runtime = select_best_runtime(requirements)if best_runtime: print(f"Recommended runtime: {best_runtime['runtime_id']}") print(f"Match score: {best_runtime['match_score']}")
This example helps you generate a matrix comparing all available runtimes, their features, and compatibility. It’s valuable for platform administrators or teams evaluating which runtimes best fit their needs.
def generate_compatibility_matrix(): """Generate a compatibility matrix for all runtimes""" runtimes = client.runtimes.list() matrix = {} for runtime in runtimes: runtime_id = runtime.get('name') try: requirements = client.runtimes.get_requirements(runtime_id) matrix[runtime_id] = { "name": runtime.get('name'), "type": runtime.get('type'), "features": requirements.get('supported_features', []), "model_compatibility": requirements.get('model_requirements', {}), "status": runtime.get('status', 'active') } except RuntimeError: matrix[runtime_id] = {"error": "Failed to get requirements"} return matrix# Usagematrix = generate_compatibility_matrix()print("Runtime Compatibility Matrix:")print(f"{'Runtime':<20} {'Type':<15} {'Features':<30} {'Status':<10}")print("-" * 75)for runtime_id, info in matrix.items(): if "error" in info: print(f"{runtime_id:<20} {'ERROR':<15} {info['error']:<30} {'N/A':<10}") else: features = ", ".join(info['features'][:2]) + "..." if len(info['features']) > 2 else ", ".join(info['features']) print(f"{info['name']:<20} {info['type']:<15} {features:<30} {info['status']:<10}")
Use this builder class to construct and validate runtime configurations step by step. This approach is helpful for complex deployments where you want to ensure all configuration details are correct before deploying an agent.
class RuntimeConfigBuilder: """Helper class to build validated runtime configurations""" def __init__(self, client): self.client = client self.config = {} def set_runtime(self, runtime_type: str): """Set the runtime type""" self.config['runtime_type'] = runtime_type return self def set_model(self, model: str): """Set the model""" self.config['model'] = model return self def add_features(self, features: list): """Add features to the configuration""" self.config['features'] = features return self def set_environment(self, env_config: dict): """Set environment configuration""" self.config['environment'] = env_config return self def validate(self): """Validate the configuration""" return self.client.runtimes.validate(self.config) def build(self): """Build and return the configuration""" # Validate before building validation = self.validate() if not validation.get('valid'): raise ValueError(f"Invalid configuration: {validation.get('errors')}") return self.config# Usagebuilder = RuntimeConfigBuilder(client)try: config = (builder .set_runtime("agno") .set_model("kubiya/claude-sonnet-4") .add_features(["code_execution", "web_browsing"]) .set_environment({"timeout": 300, "max_retries": 3}) .build()) print(f"✅ Built valid configuration: {config}")except ValueError as e: print(f"❌ Configuration error: {e}")
Monitor the health status of all runtimes in your environment. This is useful for ongoing operations and maintenance, allowing you to quickly identify and address issues with inactive or unhealthy runtimes.
def monitor_runtime_health(): """Monitor health of all runtimes""" runtimes = client.runtimes.list() health_status = { "healthy": [], "unhealthy": [], "unknown": [] } for runtime in runtimes: runtime_id = runtime.get('name') status = runtime.get('status', 'unknown') if status == 'active': health_status['healthy'].append(runtime_id) elif status == 'inactive' or status == 'error': health_status['unhealthy'].append(runtime_id) else: health_status['unknown'].append(runtime_id) return health_status# Usagehealth = monitor_runtime_health()print(f"Runtime Health Status:")print(f" ✅ Healthy: {len(health['healthy'])} - {health['healthy']}")print(f" ❌ Unhealthy: {len(health['unhealthy'])} - {health['unhealthy']}")print(f" ⚠️ Unknown: {len(health['unknown'])} - {health['unknown']}")
When working with runtimes, you may encounter errors such as missing runtimes, invalid configurations, or unavailable features. These examples show how to handle such errors gracefully and provide fallback options to keep your workflows running smoothly.
from kubiya.resources.exceptions import RuntimeErrortry: # Try to get runtime requirements requirements = client.runtimes.get_requirements("non-existent-runtime")except RuntimeError as e: print(f"Runtime error: {e}") # Handle error - maybe fall back to default runtime print("Falling back to default runtime") runtimes = client.runtimes.list() default_runtime = runtimes[0] if runtimes else None if default_runtime: print(f"Using runtime: {default_runtime['name']}")# Validate with error handlingtry: config = {"runtime_type": "invalid"} validation = client.runtimes.validate(config) if not validation.get('valid'): print(f"Configuration invalid: {validation.get('errors')}")except RuntimeError as e: print(f"Validation error: {e}")
Follow these best practices to make your use of the Runtimes Service more robust, efficient, and maintainable. These patterns help you avoid common pitfalls and ensure your code is resilient to changes in runtime availability or configuration.
To reduce API calls and improve performance, cache runtime and requirements information locally after the first retrieval. This is especially useful in applications that repeatedly access the same runtime details.
# Cache runtimes to reduce API callsclass RuntimeCache: def __init__(self, client): self.client = client self._runtimes_cache = None self._requirements_cache = {} def get_runtimes(self): """Get cached runtimes""" if self._runtimes_cache is None: self._runtimes_cache = self.client.runtimes.list() return self._runtimes_cache def get_requirements(self, runtime_id: str): """Get cached requirements for a runtime""" if runtime_id not in self._requirements_cache: self._requirements_cache[runtime_id] = self.client.runtimes.get_requirements(runtime_id) return self._requirements_cache[runtime_id] def invalidate_cache(self): """Clear the cache""" self._runtimes_cache = None self._requirements_cache = {}# Usagecache = RuntimeCache(client)runtimes = cache.get_runtimes() # API callruntimes_again = cache.get_runtimes() # From cache
Always validate your runtime configuration before deploying an agent. This helps catch errors early and ensures your deployment will succeed without unexpected issues.
# Always validate runtime configuration before usingdef deploy_agent_safe(agent_config: dict): """Deploy agent with pre-validation""" # Validate runtime configuration first validation = client.runtimes.validate(agent_config) if not validation.get('valid'): raise ValueError(f"Invalid runtime configuration: {validation.get('errors')}") # Proceed with deployment # ... deployment logic ... return {"success": True, "config": agent_config}# Usageconfig = { "runtime_type": "agno", "model": "kubiya/claude-sonnet-4"}try: result = deploy_agent_safe(config) print(f"✅ Agent deployed successfully")except ValueError as e: print(f"❌ Deployment failed: {e}")
Before building a runtime configuration, check the requirements for the selected runtime. This ensures your configuration will meet all necessary criteria and reduces the risk of deployment failures.
def build_config_from_requirements(runtime_id: str): """Build configuration based on runtime requirements""" # Get requirements first requirements = client.runtimes.get_requirements(runtime_id) # Build config based on requirements config = { "runtime_type": runtime_id, "features": requirements.get('required_features', []), "environment": requirements.get('default_environment', {}) } # Add model if specified in requirements if 'model_requirements' in requirements: model_reqs = requirements['model_requirements'] if 'default_model' in model_reqs: config['model'] = model_reqs['default_model'] return config# Usageconfig = build_config_from_requirements("agno")print(f"Generated configuration: {config}")
Always provide a fallback mechanism in case your preferred runtime is unavailable. This keeps your application resilient and ensures continuity of service even if a runtime is removed or temporarily inaccessible.
def get_runtime_with_fallback(preferred_runtime: str, fallback_runtime: str = "agno"): """Get runtime with automatic fallback""" try: # Try to get preferred runtime requirements requirements = client.runtimes.get_requirements(preferred_runtime) return { "runtime_id": preferred_runtime, "requirements": requirements, "is_fallback": False } except RuntimeError as e: print(f"Preferred runtime unavailable: {e}") print(f"Falling back to: {fallback_runtime}") try: requirements = client.runtimes.get_requirements(fallback_runtime) return { "runtime_id": fallback_runtime, "requirements": requirements, "is_fallback": True } except RuntimeError: raise ValueError("Both preferred and fallback runtimes unavailable")# Usageruntime = get_runtime_with_fallback("custom-runtime", "agno")print(f"Using runtime: {runtime['runtime_id']}")if runtime['is_fallback']: print("⚠️ Using fallback runtime")
{ "valid": bool, # Whether configuration is valid "errors": ["string"], # List of validation errors "warnings": ["string"], # List of validation warnings "requirements_met": bool, # Whether all requirements are met "validated_config": dict # Validated configuration}