Environments Service

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()

Getting Started

Control Plane Client

Context Graph & Memory

Advanced

Resources

Environments Service

Environments: detailed guide

Create an environment example

Core Operations

List Environments

Get Environment Details

Create Environment

Update Environment

Delete Environment

Worker Registration

Get Worker Command

Error Handling

Best Practices

Common Patterns

Set Up Multiple Environments

Environment Health Check

​Environments: detailed guide

​Create an environment example

​Core Operations

​List Environments

​Get Environment Details

​Create Environment

​Update Environment

​Delete Environment

​Worker Registration

​Get Worker Command

​Error Handling

​Best Practices

​Common Patterns

​Set Up Multiple Environments

​Environment Health Check

Environments: detailed guide

Create an environment example

Core Operations

List Environments

Get Environment Details

Create Environment

Update Environment

Delete Environment

Worker Registration

Get Worker Command

Error Handling

Best Practices

Common Patterns

Set Up Multiple Environments

Environment Health Check