Access and manage secrets for agents and integrations in the Control Plane
The Secrets Service provides secure access to secrets stored in the Kubiya Control Plane, enabling agents and integrations to retrieve credentials, API keys, and other sensitive data at runtime.
Use the Secrets Service whenever your agents or automation need to read sensitive credentials at runtime without hard-coding them into configuration files. It lets you centralize where secrets live, enforce access controls from the Control Plane, and build safety checks or audits around which secrets exist and how they are used.
Secrets in Kubiya are securely stored credentials, tokens, API keys, and other sensitive data used by agents and integrations. The Secrets Service allows you to:
List Secrets: View secret metadata (names, types, scopes) without exposing values
Retrieve Values: Securely access secret values when needed by agents or workflows
From the SDK you typically combine these operations to implement workflows like pre-flight checks (“are all required secrets present?”), inventory reports for security teams, and tightly scoped retrieval of individual values just before they are used.
The Secrets Service provides read-only access via the SDK. Secret creation, updates, and deletion must be performed through the Kubiya Dashboard or CLI to ensure proper security controls.
from kubiya import ControlPlaneClient# Initialize the clientclient = ControlPlaneClient(api_key="your-api-key")# List all secrets (metadata only)secrets = client.secrets.list()for secret in secrets: print(f"Secret: {secret['name']} - Type: {secret['type']}")# Get specific secret value (use with caution)secret_value = client.secrets.get_value(name="github-api-token")print(f"Token: {secret_value['value']}") # Handle securely!
from kubiya import ControlPlaneClientclient = ControlPlaneClient(api_key="your-api-key")# Retrieve actual secret valuesecret = client.secrets.get_value(name="github-api-token")# Access the secret valuetoken = secret['value']# Use the secret (example: API call)import requestsheaders = {"Authorization": f"Bearer {token}"}response = requests.get("https://api.github.com/user", headers=headers)# Clear sensitive data from memory after usetoken = Nonesecret = None
Security Warning: Always clear secret values from memory after use and never log, print, or store them insecurely.
The following examples show how to use the Secrets Service for real-world scenarios, such as auditing secret inventory, securely retrieving secret values, and handling sensitive data. Each example includes a short explanation of when and why you might use it.
Use this pattern to generate an inventory for security reviews or cleanup projects, so you know which secrets exist, where they are scoped, and which integrations they are tied to.List all secrets with categorization:
Copy
Ask AI
from kubiya import ControlPlaneClientfrom collections import defaultdictdef generate_secret_inventory(client: ControlPlaneClient): """Generate comprehensive secret inventory.""" secrets = client.secrets.list() # Categorize by type by_type = defaultdict(list) by_scope = defaultdict(list) by_integration = defaultdict(list) for secret in secrets: by_type[secret['type']].append(secret) by_scope[secret['scope']].append(secret) if secret.get('integration'): by_integration[secret['integration']].append(secret) print("Secret Inventory Report") print("=" * 60) print(f"Total Secrets: {len(secrets)}\n") print("By Type:") for secret_type, items in by_type.items(): print(f" {secret_type}: {len(items)}") print("\nBy Scope:") for scope, items in by_scope.items(): print(f" {scope}: {len(items)}") print("\nBy Integration:") for integration, items in by_integration.items(): print(f" {integration}: {len(items)}") return { "total": len(secrets), "by_type": dict(by_type), "by_scope": dict(by_scope), "by_integration": dict(by_integration) }# Usageclient = ControlPlaneClient(api_key="your-api-key")inventory = generate_secret_inventory(client)
Use this validator as a pre-flight check for environments, CI pipelines, or new projects to ensure all required secrets have been created before anything runs.Check if required secrets exist:
Copy
Ask AI
from kubiya import ControlPlaneClientfrom kubiya.resources.exceptions import ControlPlaneErrorfrom typing import Listdef validate_required_secrets( client: ControlPlaneClient, required_secrets: List[str]) -> dict: """Validate that all required secrets are available.""" try: # Get all available secrets available_secrets = client.secrets.list() available_names = {s['name'] for s in available_secrets} # Check for missing secrets missing = [] present = [] for secret_name in required_secrets: if secret_name in available_names: present.append(secret_name) else: missing.append(secret_name) # Report results if missing: print(f"❌ Missing {len(missing)} required secrets:") for name in missing: print(f" - {name}") else: print(f"✅ All {len(required_secrets)} required secrets are available") return { "valid": len(missing) == 0, "present": present, "missing": missing, "total_required": len(required_secrets) } except ControlPlaneError as e: print(f"Failed to validate secrets: {e}") return None# Usageclient = ControlPlaneClient(api_key="your-api-key")required = [ "github-api-token", "aws-credentials", "slack-webhook-url", "datadog-api-key"]validation = validate_required_secrets(client, required)if not validation['valid']: print(f"\nPlease create the missing secrets in Kubiya Dashboard")
Use this context manager in any code path that must touch raw secret values, so you minimize how long secrets stay in memory and centralize cleanup behavior.Safely retrieve and use secrets with automatic cleanup:
Copy
Ask AI
from kubiya import ControlPlaneClientfrom kubiya.resources.exceptions import ControlPlaneErrorfrom contextlib import contextmanagerimport gc@contextmanagerdef secure_secret(client: ControlPlaneClient, secret_name: str): """Context manager for secure secret access with automatic cleanup.""" secret_value = None try: # Retrieve secret secret = client.secrets.get_value(name=secret_name) secret_value = secret['value'] yield secret_value except ControlPlaneError as e: print(f"Failed to access secret '{secret_name}': {e}") raise finally: # Clear secret from memory if secret_value: secret_value = None gc.collect() # Force garbage collection# Usageclient = ControlPlaneClient(api_key="your-api-key")with secure_secret(client, "github-api-token") as token: # Use the token import requests headers = {"Authorization": f"Bearer {token}"} response = requests.get("https://api.github.com/user", headers=headers) print(f"GitHub user: {response.json()['login']}")# Token is automatically cleared from memory here
Use this helper to identify long-lived secrets that may be overdue for rotation, and to generate reports you can share with security or platform teams.Identify secrets that may need rotation:
Copy
Ask AI
from kubiya import ControlPlaneClientfrom datetime import datetime, timedeltadef check_secret_rotation_needed(client: ControlPlaneClient, days_threshold: int = 90): """Identify secrets that may need rotation based on age.""" secrets = client.secrets.list() now = datetime.utcnow() needs_rotation = [] recently_rotated = [] for secret in secrets: created_at = datetime.fromisoformat(secret['created_at'].replace('Z', '+00:00')) age_days = (now - created_at).days if age_days > days_threshold: needs_rotation.append({ "name": secret['name'], "age_days": age_days, "created_at": secret['created_at'], "type": secret['type'] }) elif age_days < 30: recently_rotated.append(secret['name']) print(f"Secret Rotation Analysis (threshold: {days_threshold} days)") print("=" * 60) if needs_rotation: print(f"\n⚠️ {len(needs_rotation)} secrets may need rotation:") for item in sorted(needs_rotation, key=lambda x: x['age_days'], reverse=True): print(f" - {item['name']}: {item['age_days']} days old ({item['type']})") else: print("\n✅ No secrets exceed rotation threshold") if recently_rotated: print(f"\n✅ {len(recently_rotated)} secrets rotated recently:") for name in recently_rotated: print(f" - {name}") return { "needs_rotation": needs_rotation, "recently_rotated": recently_rotated, "threshold_days": days_threshold }# Usageclient = ControlPlaneClient(api_key="your-api-key")rotation_report = check_secret_rotation_needed(client, days_threshold=90)
Use this checker to verify that each integration (GitHub, AWS, Slack, etc.) has all of the secrets it needs before enabling agents or workflows that depend on it.Verify secrets for specific integrations:
Copy
Ask AI
from kubiya import ControlPlaneClientfrom typing import Dict, Listdef check_integration_secrets( client: ControlPlaneClient, integration_requirements: Dict[str, List[str]]) -> dict: """Check if required secrets exist for each integration.""" secrets = client.secrets.list() # Build secret name to integration mapping available_by_integration = {} for secret in secrets: integration = secret.get('integration') if integration: if integration not in available_by_integration: available_by_integration[integration] = [] available_by_integration[integration].append(secret['name']) # Check requirements results = {} for integration, required_secrets in integration_requirements.items(): available = available_by_integration.get(integration, []) missing = [s for s in required_secrets if s not in available] results[integration] = { "required": required_secrets, "available": available, "missing": missing, "complete": len(missing) == 0 } # Print report print("Integration Secrets Report") print("=" * 60) for integration, result in results.items(): status = "✅" if result['complete'] else "❌" print(f"\n{status} {integration}:") print(f" Required: {len(result['required'])}") print(f" Available: {len(result['available'])}") if result['missing']: print(f" Missing: {', '.join(result['missing'])}") return results# Usageclient = ControlPlaneClient(api_key="your-api-key")requirements = { "github": ["github-api-token", "github-webhook-secret"], "aws": ["aws-credentials", "aws-session-token"], "slack": ["slack-webhook-url", "slack-bot-token"], "datadog": ["datadog-api-key", "datadog-app-key"]}integration_report = check_integration_secrets(client, requirements)
Secret operations can fail because a secret does not exist, the caller lacks permission, or network and authentication problems prevent the Control Plane from returning data. The following patterns show how to handle ControlPlaneError alongside common SDK errors, and how to distinguish between not-found and permission-denied cases.
Copy
Ask AI
from kubiya import ControlPlaneClientfrom kubiya.resources.exceptions import ControlPlaneErrorfrom kubiya.core.exceptions import ( APIError as KubiyaAPIError, AuthenticationError as KubiyaAuthenticationError, TimeoutError as KubiyaTimeoutError)client = ControlPlaneClient(api_key="your-api-key")# Handle secret access errorstry: secrets = client.secrets.list()except ControlPlaneError as e: print(f"Failed to list secrets: {e}")except KubiyaAuthenticationError as e: print(f"Authentication failed: {e}")except KubiyaAPIError as e: print(f"API error: {e.status_code} - {e.message}")# Handle secret not foundtry: secret = client.secrets.get_value(name="non-existent-secret")except ControlPlaneError as e: if "not found" in str(e).lower(): print("Secret not found") elif "permission denied" in str(e).lower(): print("Insufficient permissions to access secret") else: print(f"Error accessing secret: {e}")# Handle permission errorstry: secret = client.secrets.get_value(name="admin-only-secret")except ControlPlaneError as e: print(f"Access denied: {e}") # Log audit trail or alert security team
These patterns help you use secrets safely from code: avoid leaking values, keep access tightly scoped, validate requirements up front, and only cache when you have a clear operational need.
Never print or log raw secret values. If you need to log something, prefer high-level metadata like secret names or counts instead of the underlying data.
Copy
Ask AI
from kubiya import ControlPlaneClientimport loggingclient = ControlPlaneClient(api_key="your-api-key")# ❌ BAD - Never do thissecret = client.secrets.get_value(name="api-token")logging.info(f"Token: {secret['value']}") # NEVER LOG SECRET VALUES!# ✅ GOOD - Log metadata onlysecrets = client.secrets.list()logging.info(f"Found {len(secrets)} secrets")logging.info(f"Secret names: {[s['name'] for s in secrets]}")
Wrap secret access in context managers so you have a single, well-reviewed place where values are loaded and then cleared from memory once the work is done.
Copy
Ask AI
from contextlib import contextmanagerfrom kubiya import ControlPlaneClient@contextmanagerdef get_secret_safely(client: ControlPlaneClient, name: str): """Safely retrieve and cleanup secrets.""" secret_value = None try: secret = client.secrets.get_value(name=name) yield secret['value'] finally: secret_value = None # Clear from memory# Usageclient = ControlPlaneClient(api_key="your-api-key")with get_secret_safely(client, "api-token") as token: # Use token here pass# Token is cleared automatically
Validate that required secrets exist before starting jobs or deployments so failures happen early and clearly instead of deep inside business logic.
Copy
Ask AI
from kubiya import ControlPlaneClientdef ensure_secrets_available(client: ControlPlaneClient, required: list) -> bool: """Check secrets exist before attempting to use them.""" available = {s['name'] for s in client.secrets.list()} missing = [name for name in required if name not in available] if missing: raise ValueError(f"Missing required secrets: {', '.join(missing)}") return True# Usageclient = ControlPlaneClient(api_key="your-api-key")ensure_secrets_available(client, ["github-token", "aws-key"])
If you must cache secrets for performance reasons, keep the cache in memory only, use short TTLs, and provide an explicit way to clear cached values.
Copy
Ask AI
from kubiya import ControlPlaneClientfrom datetime import datetime, timedeltafrom typing import Optionalclass SecretCache: """In-memory secret cache with TTL (use with caution!).""" def __init__(self, client: ControlPlaneClient, ttl_seconds: int = 300): self.client = client self.ttl_seconds = ttl_seconds self._cache = {} def get(self, name: str) -> str: """Get secret with caching.""" now = datetime.utcnow() # Check cache if name in self._cache: value, expiry = self._cache[name] if now < expiry: return value # Fetch and cache secret = self.client.secrets.get_value(name=name) value = secret['value'] expiry = now + timedelta(seconds=self.ttl_seconds) self._cache[name] = (value, expiry) return value def clear(self): """Clear all cached secrets.""" self._cache = {}# Usage (use only when absolutely necessary)cache = SecretCache(client, ttl_seconds=300)token = cache.get("api-token")
Security Note: Only cache secrets when absolutely necessary and always use short TTLs. Cached secrets in memory are a security risk if the process is compromised.