Environments Service

The Environments Service enables you to manage runtime environments for agent and team execution. Environments provide isolated execution contexts with their own worker pools, configurations, and resources.

Quick Start

from kubiya import ControlPlaneClient

client = ControlPlaneClient()

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

# Get a specific environment
environment = client.environments.get(environment_id="env-uuid")

# Create a new environment
env_data = {
    "name": "Production",
    "description": "Production environment",
    "status": "active"
}
environment = client.environments.create(env_data)

# Get worker registration command
worker_cmd = client.environments.get_worker_command(environment_id=environment['id'])
print(f"Worker command: {worker_cmd['command']}")

Features

Environment Management

Create and configure isolated environments

Worker Registration

Generate commands for worker registration

Auto-Provisioning

Automatic Temporal namespace setup

Status Tracking

Monitor environment provisioning status

List Environments

Retrieve all environments in your organization:

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

# List with status filter
active_envs = client.environments.list(status_filter="active")
provisioning_envs = client.environments.list(status_filter="provisioning")
ready_envs = client.environments.list(status_filter="ready")
error_envs = client.environments.list(status_filter="error")

Parameters:

status_filter (str, optional): Filter by status: ‘active’, ‘inactive’, ‘provisioning’, ‘ready’, or ‘error’

Returns: List of environment dictionaries

Get Environment

Retrieve a specific environment by ID:

environment = client.environments.get(environment_id="550e8400-e29b-41d4-a716-446655440000")

print(f"Environment: {environment['name']}")
print(f"Status: {environment['status']}")
print(f"Created: {environment['created_at']}")
print(f"Namespace: {environment.get('temporal_namespace', 'N/A')}")

Parameters:

environment_id (str, required): Environment UUID

Returns: Dictionary containing environment details

Create Environment

Create a new environment:

env_data = {
    "name": "Staging",
    "description": "Staging environment for testing",
    "status": "active",
    "configuration": {
        "max_workers": 10,
        "timeout_seconds": 3600
    },
    "metadata": {
        "team": "platform",
        "cost_center": "engineering"
    }
}

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

Parameters:

environment_data (dict, required): Environment configuration
- name (str): Environment name
- description (str): Environment description
- status (str, optional): Environment status (default: ‘active’)
- configuration (dict, optional): Environment-specific configuration
- metadata (dict, optional): Additional metadata

Returns: Dictionary containing created environment details

If this is the first environment for your organization, it will trigger the Temporal Cloud namespace provisioning workflow automatically.

Update Environment

Update an existing environment (partial update):

# Update environment name
updated_env = client.environments.update(
    environment_id="env-uuid",
    environment_data={
        "name": "Production-US-East",
        "description": "US East production environment"
    }
)

# Update configuration
updated_env = client.environments.update(
    environment_id="env-uuid",
    environment_data={
        "configuration": {
            "max_workers": 20,
            "timeout_seconds": 7200
        }
    }
)

# Update status
updated_env = client.environments.update(
    environment_id="env-uuid",
    environment_data={
        "status": "inactive"
    }
)

Parameters:

environment_id (str, required): Environment UUID
environment_data (dict, required): Fields to update (partial update)
- name (str, optional): Environment name
- description (str, optional): Description
- status (str, optional): Environment status
- configuration (dict, optional): Configuration settings
- metadata (dict, optional): Additional metadata

Returns: Dictionary containing updated environment details

Delete Environment

Delete an environment:

client.environments.delete(environment_id="env-uuid")
print("Environment deleted successfully")

Parameters:

environment_id (str, required): Environment UUID

Returns: None (204 No Content on success)

You cannot delete the default environment. Deleting an environment will stop all workers registered to it.

Worker Registration

Get Worker Command

Get the registration command for workers in an environment:

worker_info = client.environments.get_worker_command(environment_id="env-uuid")

print(f"Worker Token: {worker_info['worker_token']}")
print(f"Command: {worker_info['command']}")
print(f"Namespace: {worker_info['temporal_namespace']}")

# The command will be similar to:
# kubiya worker start --token <worker-token> --environment <env-name>

Parameters:

environment_id (str, required): Environment UUID

Returns: Dictionary containing:

command (str): Full worker start command
worker_token (str): Worker authentication token
temporal_namespace (str): Temporal namespace
environment_name (str): Environment name

Copy the worker command and run it on your worker machines to register them with the environment.

Worker Registration Example

# On your worker machine, run the command provided:
kubiya worker start --token eyJhbGc... --environment production

# Or with Docker:
docker run -d \
  --name kubiya-worker \
  kubiya/worker:latest \
  start --token eyJhbGc... --environment production

Complete Example

Here’s a complete example of setting up environments for different stages:

from kubiya import ControlPlaneClient

client = ControlPlaneClient()

# 1. Create environments for each stage
environments = [
    {
        "name": "Development",
        "description": "Development environment",
        "configuration": {
            "max_workers": 5,
            "timeout_seconds": 1800
        },
        "metadata": {"stage": "dev"}
    },
    {
        "name": "Staging",
        "description": "Staging environment",
        "configuration": {
            "max_workers": 10,
            "timeout_seconds": 3600
        },
        "metadata": {"stage": "staging"}
    },
    {
        "name": "Production",
        "description": "Production environment",
        "configuration": {
            "max_workers": 20,
            "timeout_seconds": 7200
        },
        "metadata": {"stage": "production"}
    }
]

created_envs = []
for env_data in environments:
    env = client.environments.create(env_data)
    created_envs.append(env)
    print(f"Created {env['name']}: {env['id']}")

# 2. Get worker commands for each environment
print("\nWorker Registration Commands:")
for env in created_envs:
    worker_info = client.environments.get_worker_command(environment_id=env['id'])
    print(f"\n{env['name']}:")
    print(f"  {worker_info['command']}")

# 3. List all environments
all_envs = client.environments.list()
print(f"\nTotal environments: {len(all_envs)}")

# 4. Monitor environment status
for env in created_envs:
    current = client.environments.get(environment_id=env['id'])
    print(f"{current['name']}: {current['status']}")

Error Handling

Handle environment-related errors:

from kubiya.resources.exceptions import EnvironmentError

try:
    environment = client.environments.get(environment_id="invalid-id")
except EnvironmentError as e:
    print(f"Environment error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Environment Provisioning

Understanding Environment States

Environments go through several states during provisioning:

provisioning: Initial state when environment is created
ready: Temporal namespace is provisioned and ready
active: Environment is active and accepting work
error: Provisioning failed or environment error occurred
inactive: Environment is disabled

Monitoring Provisioning

import time

# Create environment
env = client.environments.create({
    "name": "New Environment",
    "description": "Newly provisioned environment"
})

# Poll until ready
print("Waiting for environment provisioning...")
while True:
    current = client.environments.get(environment_id=env['id'])
    status = current['status']
    print(f"Status: {status}")

    if status == 'ready' or status == 'active':
        print("Environment ready!")
        break
    elif status == 'error':
        print("Provisioning failed!")
        break

    time.sleep(5)

Best Practices

Environment Strategy

Environment Per Stage: Create separate environments for dev, staging, and production
Isolated Resources: Use environments to isolate different workloads
Naming Convention: Use clear, consistent naming (e.g., “production-us-east”)
Default Environment: Reserve the default environment for testing and development

Worker Management

Dedicated Workers: Deploy dedicated workers for each environment
Worker Scaling: Scale workers based on environment workload
Secure Tokens: Treat worker tokens as secrets; rotate regularly
Health Monitoring: Monitor worker health and connectivity

Configuration

Resource Limits: Set appropriate max_workers and timeout values
Metadata: Use metadata to track ownership, cost centers, and stages
Status Management: Use status to control environment availability
Regular Reviews: Review environment configurations periodically

Security

Token Rotation: Regularly rotate worker tokens
Access Control: Limit who can create/modify environments
Network Isolation: Use network policies to isolate environment workers
Audit Logging: Monitor environment changes and access

Use Cases

Multi-Stage Deployment

# Separate environments for deployment stages
dev = client.environments.create({"name": "dev", "status": "active"})
staging = client.environments.create({"name": "staging", "status": "active"})
prod = client.environments.create({"name": "production", "status": "active"})

Multi-Region Setup

# Environments for different regions
us_east = client.environments.create({
    "name": "production-us-east-1",
    "metadata": {"region": "us-east-1"}
})

eu_west = client.environments.create({
    "name": "production-eu-west-1",
    "metadata": {"region": "eu-west-1"}
})

Team Isolation

# Separate environments for different teams
platform_env = client.environments.create({
    "name": "platform-team",
    "metadata": {"team": "platform"}
})

data_env = client.environments.create({
    "name": "data-team",
    "metadata": {"team": "data"}
})

Testing Environments

# Ephemeral environment for testing
test_env = client.environments.create({
    "name": f"test-{timestamp}",
    "description": "Temporary test environment",
    "configuration": {
        "max_workers": 1,
        "timeout_seconds": 600
    }
})

# Clean up after tests
client.environments.delete(environment_id=test_env['id'])

API Reference

Method	Endpoint Pattern	Description
`list()`	GET /environments	List all environments
`get()`	GET /environments/	Get environment by ID
`create()`	POST /environments	Create new environment
`update()`	PATCH /environments/	Update environment
`delete()`	DELETE /environments/	Delete environment
`get_worker_command()`	GET /environments//worker-command	Get worker registration command

Next Steps

Workers Service

Monitor and manage workers

Agents Service

Deploy agents to environments

Teams Service

Run teams in specific environments

Jobs Service

Schedule jobs to run in environments

Getting Started

Control Plane API

Control Plane Client

Context Graph & Memory

Client SDK

Workflow DSL

Advanced

Resources

​Quick Start

​Features

Environment Management

Worker Registration

Auto-Provisioning

Status Tracking

​List Environments

​Get Environment

​Create Environment

​Update Environment

​Delete Environment

​Worker Registration

​Get Worker Command

​Worker Registration Example

​Complete Example

​Error Handling

​Environment Provisioning

​Understanding Environment States

​Monitoring Provisioning

​Best Practices

​Environment Strategy

​Worker Management

​Configuration

​Security

​Use Cases

​Multi-Stage Deployment

​Multi-Region Setup

​Team Isolation

​Testing Environments

​API Reference

​Next Steps

Workers Service

Agents Service

Teams Service

Jobs Service

Quick Start

Features

List Environments

Get Environment

Create Environment

Update Environment

Delete Environment

Worker Registration

Get Worker Command

Worker Registration Example

Complete Example

Error Handling

Environment Provisioning

Understanding Environment States

Monitoring Provisioning

Best Practices

Environment Strategy

Worker Management

Configuration

Security

Use Cases

Multi-Stage Deployment

Multi-Region Setup

Team Isolation

Testing Environments

API Reference

Next Steps