Environments Service

Manage execution environments and worker registration with the Kubiya SDK

The Environments Service provides comprehensive management of execution environments through the Kubiya platform. Environments define isolated execution contexts for your agents and workflows, including worker registration and resource allocation.

Environments: detailed guide

This page focuses on the environment-specific parts of the SDK: creation, configuration, status management, and worker registration.

If you haven’t set up a ControlPlaneClient yet, see the Client Quick Start at Client Overview for initialization and authentication instructions.

Create an environment example

The following example demonstrates creating an environment using the Control Plane SDK:

from kubiya.control_plane_client import ControlPlaneClient

client = ControlPlaneClient()

environment_data = {
    "name": "PRODUCTION",
    "display_name": "Production",
    "description": "Production execution environment",
    "tags": ["prod", "critical"],
    "priority": 10,
    "policy_ids": [],
    "execution_environment": {
        "env_vars": {},
        "secrets": [],
        "integration_ids": []
    }
}

environment = client.environments.create(environment_data)
print(f"Created environment: {environment['name']} (ID: {environment['id']})")

Core Operations

List Environments

Retrieve a list of environments with optional status filtering:

# List all environments
environments = client.environments.list()

# Filter by status
active_envs = client.environments.list(status_filter="active")
ready_envs = client.environments.list(status_filter="ready")
provisioning_envs = client.environments.list(status_filter="provisioning")

for env in environments:
    print(f"Environment: {env['name']} - Status: {env.get('status')}")

Parameters:

Parameter	Type	Default	Description
status_filter	str	None	Filter by status: active, inactive, provisioning, ready, error

Environment statuses:

Status	Description
active	Environment is active and running
inactive	Environment is disabled
provisioning	Environment is being set up
ready	Environment is ready for use
error	Environment encountered an error

Get Environment Details

Fetch a single environment’s details by UUID:

environment = client.environments.get("environment-uuid-here")
print(environment)

Create Environment

Create a new environment:

environment_data = {
    "name": "STAGING",
    "display_name": "Staging",
    "description": "Staging environment for testing",
    "tags": [],
    "priority": 5,
    "policy_ids": [],
    "execution_environment": {
        "env_vars": {},
        "secrets": [],
        "integration_ids": []
    }
}

environment = client.environments.create(environment_data)
print(f"Created: {environment['id']}")

Environment configuration fields:

Field	Type	Required	Description
name	str	Yes	Environment name (typically uppercase, e.g., PRODUCTION)
display_name	str	No	Human-readable display name
description	str	No	Environment description
tags	list	No	List of tags for categorization
priority	int	No	Priority level (default: 5)
policy_ids	list	No	List of policy IDs to apply
execution_environment	dict	No	Execution environment configuration

Execution environment fields:

Field	Type	Description
env_vars	dict	Environment variables
secrets	list	List of secret names to inject
integration_ids	list	List of integration IDs to enable

Update Environment

Update an existing environment’s configuration (partial update - only provided fields are updated):

update_data = {
    "description": "Updated environment description",
    "status": "active"
}

updated = client.environments.update("environment-uuid", update_data)
print(updated)

Delete Environment

Permanently remove an environment by UUID:

client.environments.delete("environment-uuid")

---

Worker Registration

Get Worker Command

Get the worker registration command for an environment. This returns the kubiya worker start command with the worker token:

worker_info = client.environments.get_worker_command("environment-uuid")

print("Worker registration command:")
print(worker_info.get('command'))

This command can be used to register new workers with the environment, enabling them to execute tasks.

---

Error Handling

The Environments Service raises EnvironmentError for API errors:

from kubiya.resources.exceptions import EnvironmentError

try:
    environment = client.environments.get("non-existent-environment")
except EnvironmentError as e:
    print(f"Environment operation failed: {e}")

---

Best Practices

• Use descriptive environment names (e.g., production, staging, development)
• Monitor environment status, especially during provisioning
• Keep track of environment UUIDs for worker registration
• Use separate environments for different deployment stages
• Check environment status before routing tasks to workers
• Handle provisioning status appropriately in automation scripts

---

Common Patterns

Set Up Multiple Environments

# Create environments for different stages
env_configs = [
    {"name": "DEVELOPMENT", "display_name": "Development", "priority": 1},
    {"name": "STAGING", "display_name": "Staging", "priority": 5},
    {"name": "PRODUCTION", "display_name": "Production", "priority": 10}
]

created_envs = {}
for config in env_configs:
    env = client.environments.create({
        "name": config["name"],
        "display_name": config["display_name"],
        "description": f"{config['display_name']} execution environment",
        "priority": config["priority"],
        "tags": [],
        "policy_ids": [],
        "execution_environment": {
            "env_vars": {},
            "secrets": [],
            "integration_ids": []
        }
    })
    created_envs[config["name"]] = env
    print(f"Created {config['display_name']}: {env['id']}")

# Get worker commands for each
for env_name, env in created_envs.items():
    worker_info = client.environments.get_worker_command(env['id'])
    print(f"\n{env_name} worker command:")
    print(worker_info.get('command'))

Environment Health Check

def check_environment_health():
    """Check the health of all environments."""
    environments = client.environments.list()

    healthy = []
    unhealthy = []

    for env in environments:
        status = env.get('status')
        if status in ('active', 'ready'):
            healthy.append(env)
        else:
            unhealthy.append(env)

    print(f"Healthy environments: {len(healthy)}")
    for env in healthy:
        print(f"  - {env['name']}: {env['status']}")

    if unhealthy:
        print(f"\nUnhealthy environments: {len(unhealthy)}")
        for env in unhealthy:
            print(f"  - {env['name']}: {env['status']}")

    return healthy, unhealthy

# Run health check
healthy, unhealthy = check_environment_health()