Integrations Service

The Integrations Service provides programmatic access to third-party service integrations configured in Kubiya, enabling agents to interact with external systems like GitHub, Slack, AWS, and more.

Use the Integrations Service whenever you need to understand which external systems are connected to Kubiya, verify that required integrations are ready before running workflows, or retrieve credentials just-in-time so agents can call out to third-party APIs safely.

Overview

Integrations in Kubiya connect the platform to external services and APIs. The Integrations Service allows you to:

List Integrations: View all configured integrations and their connection status
Get Integration Details: Access specific integration configurations and metadata
Retrieve Credentials: Securely access integration credentials for API calls

From the SDK you typically use these operations to build health dashboards, pre-flight checks for critical vendors (GitHub, Slack, AWS, etc.), and helper utilities that fetch and clean up credentials around specific API calls.

Integration configuration (adding, updating, deleting integrations) is performed through the Kubiya Dashboard. The SDK provides read-only access to integration information.

Quick Start

from kubiya import ControlPlaneClient

# Initialize the client
client = ControlPlaneClient(api_key="your-api-key")

# List all integrations
integrations = client.integrations.list()
for integration in integrations:
    print(f"Integration: {integration['name']} ({integration['vendor']}) - Connected: {integration['connected']}")

# Get specific integration details
github_integration = client.integrations.get(integration_id="github-prod")
print(f"GitHub: {github_integration['config']}")

# Get integration credentials
credentials = client.integrations.get_integration_credentials(
    vendor="github",
    id="github-prod"
)

Core Concepts

Integration Types

Kubiya supports various integration categories:

Version Control: GitHub, GitLab, Bitbucket
Communication: Slack, Microsoft Teams, Discord
Cloud Providers: AWS, Azure, GCP
Monitoring: Datadog, New Relic, PagerDuty
Ticketing: Jira, ServiceNow, Linear
CI/CD: Jenkins, CircleCI, GitHub Actions
Databases: PostgreSQL, MySQL, MongoDB

Integration States

Connected: Integration is configured and credentials are valid
Disconnected: Integration is configured but credentials are invalid or expired
Not Configured: Integration type is available but not set up

Special Integrations

Some integrations (Jira, GitHub App) have special handling:

Always use integration ID "0"
Managed differently by the platform
May have organization-wide configuration

Basic Usage

List All Integrations

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

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

print(f"Total integrations: {len(integrations)}")
for integration in integrations:
    status = "✅" if integration['connected'] else "❌"
    print(f"{status} {integration['name']} ({integration['vendor']})")
    print(f"   Type: {integration['type']}")
    print(f"   Connected: {integration['connected']}")
    if integration.get('last_sync'):
        print(f"   Last Sync: {integration['last_sync']}")
    print("---")

Example Response

[
  {
    "id": "github-prod",
    "name": "GitHub Production",
    "vendor": "github",
    "type": "version_control",
    "connected": true,
    "config": {
      "organization": "acme-corp",
      "repos": ["backend", "frontend", "infrastructure"]
    },
    "created_at": "2025-10-15T09:00:00Z",
    "last_sync": "2025-12-11T10:25:00Z"
  },
  {
    "id": "slack-main",
    "name": "Slack Workspace",
    "vendor": "slack",
    "type": "communication",
    "connected": true,
    "config": {
      "workspace_id": "T01234567",
      "channels": ["#engineering", "#ops"]
    },
    "created_at": "2025-09-20T14:30:00Z",
    "last_sync": "2025-12-11T10:30:00Z"
  },
  {
    "id": "aws-prod",
    "name": "AWS Production Account",
    "vendor": "aws",
    "type": "cloud_provider",
    "connected": true,
    "config": {
      "account_id": "123456789012",
      "regions": ["us-east-1", "us-west-2"]
    },
    "created_at": "2025-11-01T08:15:00Z",
    "last_sync": "2025-12-11T10:20:00Z"
  },
  {
    "id": "jira-main",
    "name": "Jira Cloud",
    "vendor": "jira",
    "type": "ticketing",
    "connected": false,
    "config": {
      "domain": "acme.atlassian.net"
    },
    "created_at": "2025-08-10T11:00:00Z",
    "last_sync": null
  }
]

List Connected Integrations Only

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Filter to only connected integrations
connected_integrations = client.integrations.list(connected_only=True)

print(f"Connected integrations: {len(connected_integrations)}")
for integration in connected_integrations:
    print(f"✅ {integration['name']} ({integration['vendor']})")

Get Integration Details

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Get specific integration by ID
integration = client.integrations.get(integration_id="github-prod")

print(f"Integration: {integration['name']}")
print(f"Vendor: {integration['vendor']}")
print(f"Type: {integration['type']}")
print(f"Connected: {integration['connected']}")
print(f"Configuration: {integration['config']}")

Get Integration Credentials

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Get credentials for GitHub integration
credentials = client.integrations.get_integration_credentials(
    vendor="github",
    id="github-prod"
)

# Use credentials (example: make API call)
token = credentials['token']
import requests
headers = {"Authorization": f"Bearer {token}"}
response = requests.get("https://api.github.com/user/repos", headers=headers)

# Clear sensitive data
token = None
credentials = None

Credentials contain sensitive authentication data. Always handle them securely and clear from memory after use.

Get Special Integration Credentials

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Special integrations (Jira, GitHub App) automatically use ID "0"
jira_credentials = client.integrations.get_integration_credentials(
    vendor="jira",
    id="any-id"  # Will be automatically converted to "0"
)

github_app_credentials = client.integrations.get_integration_credentials(
    vendor="github_app",
    id="any-id"  # Will be automatically converted to "0"
)

Special integrations (Jira, GitHub App) always use ID "0" internally, regardless of the ID you provide. This is handled automatically by the SDK.

Practical Examples

The following examples show how to use the Integrations Service for real-world scenarios, such as listing integrations, retrieving credentials, and handling integration states. Each example includes a short explanation of when and why you might use it.

1. Integration Health Dashboard

Use this pattern to build a simple dashboard or scheduled job that reports which integrations are connected, disconnected, or stale so you can react before they break workflows. Monitor all integrations and report connection status:

from kubiya import ControlPlaneClient
from datetime import datetime, timedelta
from typing import Dict, Any

def generate_integration_health_report(client: ControlPlaneClient) -> Dict[str, Any]:
    """Generate comprehensive integration health report."""
    integrations = client.integrations.list()

    connected = []
    disconnected = []
    stale = []
    by_vendor = {}
    by_type = {}

    now = datetime.utcnow()

    for integration in integrations:
        # Track by connection status
        if integration['connected']:
            connected.append(integration)

            # Check for stale sync (no sync in last 24 hours)
            if integration.get('last_sync'):
                last_sync = datetime.fromisoformat(integration['last_sync'].replace('Z', '+00:00'))
                if now - last_sync > timedelta(hours=24):
                    stale.append(integration)
        else:
            disconnected.append(integration)

        # Track by vendor
        vendor = integration['vendor']
        if vendor not in by_vendor:
            by_vendor[vendor] = []
        by_vendor[vendor].append(integration)

        # Track by type
        int_type = integration['type']
        if int_type not in by_type:
            by_type[int_type] = []
        by_type[int_type].append(integration)

    # Print report
    print("Integration Health Report")
    print("=" * 60)
    print(f"Total Integrations: {len(integrations)}")
    print(f"✅ Connected: {len(connected)}")
    print(f"❌ Disconnected: {len(disconnected)}")
    print(f"⚠️  Stale (no sync in 24h): {len(stale)}")

    if disconnected:
        print("\nDisconnected Integrations:")
        for integration in disconnected:
            print(f"  ❌ {integration['name']} ({integration['vendor']})")

    if stale:
        print("\nStale Integrations:")
        for integration in stale:
            hours_since = (now - datetime.fromisoformat(integration['last_sync'].replace('Z', '+00:00'))).seconds // 3600
            print(f"  ⚠️  {integration['name']}: {hours_since}h since last sync")

    print("\nBy Vendor:")
    for vendor, items in by_vendor.items():
        print(f"  {vendor}: {len(items)}")

    print("\nBy Type:")
    for int_type, items in by_type.items():
        print(f"  {int_type}: {len(items)}")

    return {
        "total": len(integrations),
        "connected": len(connected),
        "disconnected": len(disconnected),
        "stale": len(stale),
        "by_vendor": {v: len(items) for v, items in by_vendor.items()},
        "by_type": {t: len(items) for t, items in by_type.items()}
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
health_report = generate_integration_health_report(client)

2. Integration Configuration Validator

Use this validator as a pre-flight check to ensure that all vendors your agents depend on are configured and connected before deploying environments or running scheduled jobs. Validate required integrations are configured and connected:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from typing import List, Dict

def validate_required_integrations(
    client: ControlPlaneClient,
    required: List[str]
) -> Dict[str, Any]:
    """Validate that required integrations are connected."""
    try:
        integrations = client.integrations.list(connected_only=False)

        # Build vendor to integrations mapping
        by_vendor = {}
        for integration in integrations:
            vendor = integration['vendor']
            if vendor not in by_vendor:
                by_vendor[vendor] = []
            by_vendor[vendor].append(integration)

        # Check requirements
        results = {
            "valid": True,
            "present": [],
            "missing": [],
            "disconnected": []
        }

        for vendor in required:
            if vendor in by_vendor:
                # Check if any integration for this vendor is connected
                connected = any(i['connected'] for i in by_vendor[vendor])

                if connected:
                    results["present"].append(vendor)
                else:
                    results["disconnected"].append(vendor)
                    results["valid"] = False
            else:
                results["missing"].append(vendor)
                results["valid"] = False

        # Print report
        if results["valid"]:
            print(f"✅ All {len(required)} required integrations are connected")
        else:
            if results["missing"]:
                print(f"❌ Missing {len(results['missing'])} integrations:")
                for vendor in results["missing"]:
                    print(f"  - {vendor}")

            if results["disconnected"]:
                print(f"⚠️  {len(results['disconnected'])} integrations disconnected:")
                for vendor in results["disconnected"]:
                    print(f"  - {vendor}")

        return results

    except IntegrationError as e:
        print(f"Failed to validate integrations: {e}")
        return None

# Usage
client = ControlPlaneClient(api_key="your-api-key")

required_integrations = ["github", "slack", "aws", "datadog"]
validation = validate_required_integrations(client, required_integrations)

if not validation['valid']:
    print("\nPlease configure missing integrations in Kubiya Dashboard")

3. Integration Credentials Manager

Use this context manager whenever you need to work with raw integration credentials so you load them only for the duration of an operation and then clear them from memory. Safely retrieve and use integration credentials:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from contextlib import contextmanager
import gc

@contextmanager
def integration_credentials(client: ControlPlaneClient, vendor: str, integration_id: str):
    """Context manager for safe credential access."""
    credentials = None
    try:
        credentials = client.integrations.get_integration_credentials(
            vendor=vendor,
            id=integration_id
        )
        yield credentials

    except IntegrationError as e:
        print(f"Failed to get credentials for {vendor}/{integration_id}: {e}")
        raise

    finally:
        if credentials:
            credentials = None
        gc.collect()

# Usage
client = ControlPlaneClient(api_key="your-api-key")

with integration_credentials(client, "github", "github-prod") as creds:
    token = creds['token']

    # Use credentials
    import requests
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get("https://api.github.com/user/repos", headers=headers)

    print(f"Found {len(response.json())} repositories")

# Credentials automatically cleared from memory

4. Multi-Integration Query

Use this helper to fan out read-only queries across multiple integrations in parallel, for example to build aggregated views across GitHub, AWS, and Slack. Query data from multiple integrations simultaneously:

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from typing import Dict, Any, List
import concurrent.futures

def query_multiple_integrations(
    client: ControlPlaneClient,
    queries: Dict[str, callable]
) -> Dict[str, Any]:
    """Execute queries across multiple integrations in parallel."""

    results = {}

    def execute_query(vendor_id: str, query_func: callable):
        """Execute single integration query."""
        try:
            return vendor_id, query_func(client)
        except IntegrationError as e:
            return vendor_id, {"error": str(e)}

    # Execute queries in parallel
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(execute_query, vendor_id, query_func)
            for vendor_id, query_func in queries.items()
        ]

        for future in concurrent.futures.as_completed(futures):
            vendor_id, result = future.result()
            results[vendor_id] = result

    return results

# Define integration-specific queries
def query_github(client: ControlPlaneClient) -> Dict:
    """Query GitHub integration."""
    creds = client.integrations.get_integration_credentials(vendor="github", id="github-prod")
    # Make API call using credentials
    return {"repos": 42, "status": "active"}

def query_aws(client: ControlPlaneClient) -> Dict:
    """Query AWS integration."""
    creds = client.integrations.get_integration_credentials(vendor="aws", id="aws-prod")
    # Make API call using credentials
    return {"instances": 15, "status": "active"}

def query_slack(client: ControlPlaneClient) -> Dict:
    """Query Slack integration."""
    creds = client.integrations.get_integration_credentials(vendor="slack", id="slack-main")
    # Make API call using credentials
    return {"channels": 23, "status": "active"}

# Usage
client = ControlPlaneClient(api_key="your-api-key")

queries = {
    "github": query_github,
    "aws": query_aws,
    "slack": query_slack
}

results = query_multiple_integrations(client, queries)

for vendor, result in results.items():
    if "error" in result:
        print(f"❌ {vendor}: {result['error']}")
    else:
        print(f"✅ {vendor}: {result}")

5. Integration Usage Analytics

Use this analytics helper to understand which vendors and integration types are most common in your organization and how broadly integrations are connected. Analyze integration usage patterns:

from kubiya import ControlPlaneClient
from datetime import datetime
from collections import Counter

def analyze_integration_usage(client: ControlPlaneClient) -> dict:
    """Analyze integration usage patterns."""
    integrations = client.integrations.list()

    # Categorize integrations
    vendor_counts = Counter(i['vendor'] for i in integrations)
    type_counts = Counter(i['type'] for i in integrations)
    connection_rate = sum(1 for i in integrations if i['connected']) / len(integrations)

    # Find most/least used types
    most_common_vendor = vendor_counts.most_common(1)[0] if vendor_counts else None
    most_common_type = type_counts.most_common(1)[0] if type_counts else None

    print("Integration Usage Analytics")
    print("=" * 60)
    print(f"Total Integrations: {len(integrations)}")
    print(f"Connection Rate: {connection_rate * 100:.1f}%")

    if most_common_vendor:
        print(f"Most Used Vendor: {most_common_vendor[0]} ({most_common_vendor[1]} integrations)")

    if most_common_type:
        print(f"Most Used Type: {most_common_type[0]} ({most_common_type[1]} integrations)")

    print("\nVendor Distribution:")
    for vendor, count in vendor_counts.most_common():
        print(f"  {vendor}: {count}")

    print("\nType Distribution:")
    for int_type, count in type_counts.most_common():
        print(f"  {int_type}: {count}")

    return {
        "total": len(integrations),
        "connection_rate": connection_rate,
        "vendor_counts": dict(vendor_counts),
        "type_counts": dict(type_counts)
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
usage_report = analyze_integration_usage(client)

Error Handling

Integration operations can fail if an integration is missing, disconnected, misconfigured, or if credentials cannot be fetched due to permission or network issues. The following patterns show how to catch IntegrationError along with common SDK errors, and how to distinguish not-found and permission problems when accessing credentials.

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from kubiya.core.exceptions import (
    APIError as KubiyaAPIError,
    AuthenticationError as KubiyaAuthenticationError,
    TimeoutError as KubiyaTimeoutError
)

client = ControlPlaneClient(api_key="your-api-key")

# Handle integration errors
try:
    integrations = client.integrations.list()
except IntegrationError as e:
    print(f"Integration operation failed: {e}")
except KubiyaAuthenticationError as e:
    print(f"Authentication failed: {e}")
except KubiyaAPIError as e:
    print(f"API error: {e.status_code} - {e.message}")

# Handle integration not found
try:
    integration = client.integrations.get(integration_id="non-existent")
except IntegrationError as e:
    if "not found" in str(e).lower():
        print("Integration not found")
    else:
        print(f"Error getting integration: {e}")

# Handle credential access errors
try:
    credentials = client.integrations.get_integration_credentials(
        vendor="github",
        id="invalid-id"
    )
except IntegrationError as e:
    if "not found" in str(e).lower():
        print("Integration credentials not found")
    elif "permission denied" in str(e).lower():
        print("Insufficient permissions")
    else:
        print(f"Failed to get credentials: {e}")

Best Practices

These patterns help you treat integrations as a reliable, observable dependency: always verify connectivity, avoid unnecessary API calls, handle special vendor behavior, and monitor health for critical systems.

1. Check Integration Connection Before Use

Always verify that an integration is connected before relying on it in a workflow, so failures are caught early rather than surfacing as opaque API errors.

from kubiya import ControlPlaneClient

def get_integration_safely(client: ControlPlaneClient, integration_id: str) -> dict:
    """Get integration and verify it's connected."""
    integration = client.integrations.get(integration_id=integration_id)

    if not integration['connected']:
        raise ValueError(f"Integration {integration_id} is not connected")

    return integration

# Usage
client = ControlPlaneClient(api_key="your-api-key")
try:
    github = get_integration_safely(client, "github-prod")
    # Safe to use
except ValueError as e:
    print(f"Cannot use integration: {e}")

2. Cache Integration List for Performance

Cache the integration list when you need to query it frequently (for dashboards or schedulers) so you reduce API calls while still refreshing periodically.

from kubiya import ControlPlaneClient
from datetime import datetime, timedelta

class IntegrationCache:
    """Cache integration list to reduce API calls."""

    def __init__(self, client: ControlPlaneClient, ttl_seconds: int = 300):
        self.client = client
        self.ttl_seconds = ttl_seconds
        self._cache = None
        self._expiry = None

    def get_integrations(self) -> list:
        """Get integrations with caching."""
        now = datetime.utcnow()

        # Check cache
        if self._cache and self._expiry and now < self._expiry:
            return self._cache

        # Fetch and cache
        self._cache = self.client.integrations.list()
        self._expiry = now + timedelta(seconds=self.ttl_seconds)

        return self._cache

    def invalidate(self):
        """Clear cache."""
        self._cache = None
        self._expiry = None

# Usage
cache = IntegrationCache(client, ttl_seconds=300)
integrations = cache.get_integrations()

3. Handle Special Integrations

Some vendors (like Jira and GitHub App) have special handling around IDs; centralizing that logic makes your code easier to reason about and less error-prone.

from kubiya import ControlPlaneClient

# Special integrations that always use ID "0"
SPECIAL_INTEGRATIONS = {"jira", "github_app"}

def get_credentials_safely(
    client: ControlPlaneClient,
    vendor: str,
    integration_id: str
):
    """Get credentials with special integration handling."""
    # SDK handles this automatically, but good to be aware
    if vendor.lower() in SPECIAL_INTEGRATIONS:
        print(f"Note: {vendor} is a special integration using ID '0'")

    return client.integrations.get_integration_credentials(
        vendor=vendor,
        id=integration_id
    )

# Usage
client = ControlPlaneClient(api_key="your-api-key")
jira_creds = get_credentials_safely(client, "jira", "any-id")

4. Monitor Integration Health

Set up lightweight, recurring checks for your most critical integrations so you can alert quickly when something becomes disconnected.

import schedule
from kubiya import ControlPlaneClient

def check_critical_integrations(client: ControlPlaneClient):
    """Monitor critical integrations."""
    critical = ["github", "slack", "aws"]

    connected = client.integrations.list(connected_only=True)
    connected_vendors = {i['vendor'] for i in connected}

    for vendor in critical:
        if vendor not in connected_vendors:
            # Alert or take action
            print(f"ALERT: Critical integration {vendor} is not connected!")

# Schedule checks
client = ControlPlaneClient(api_key="your-api-key")
schedule.every(15).minutes.do(lambda: check_critical_integrations(client))

API Reference

Methods

Method	Description	Parameters	Returns
`list(connected_only)`	List all integrations	`connected_only: Optional[bool]`	`List[Dict[str, Any]]`
`get(integration_id)`	Get integration details	`integration_id: str`	`Dict[str, Any]`
`get_integration_credentials(vendor, id)`	Get integration credentials	`vendor: str`, `id: str`	`Dict[str, Any]`

Integration Object Structure

{
    "id": "github-prod",
    "name": "GitHub Production",
    "vendor": "github",
    "type": "version_control",  # version_control, communication, cloud_provider, ticketing, etc.
    "connected": true,
    "config": {
        "organization": "acme-corp",
        "repos": ["backend", "frontend"]
    },
    "created_at": "2025-10-15T09:00:00Z",
    "last_sync": "2025-12-11T10:25:00Z"
}

Supported Integration Vendors

Common integration vendors include:

Version Control: github, gitlab, bitbucket
Communication: slack, teams, discord
Cloud: aws, azure, gcp
Monitoring: datadog, newrelic, pagerduty
Ticketing: jira, servicenow, linear
CI/CD: jenkins, circleci, github_actions

Next Steps

Secrets Service

Manage secrets and credentials

Context Graph

Query the knowledge graph

Error Handling

Handle SDK exceptions

Best Practices

SDK best practices guide

Getting Started

Control Plane Client

Context Graph & Memory

Advanced

Resources

Integrations Service

Overview

Quick Start

Core Concepts

Integration Types

Integration States

Special Integrations

Basic Usage

List All Integrations

List Connected Integrations Only

Get Integration Details

Get Integration Credentials

Get Special Integration Credentials

Practical Examples

1. Integration Health Dashboard

2. Integration Configuration Validator

3. Integration Credentials Manager

4. Multi-Integration Query

5. Integration Usage Analytics

Error Handling

Best Practices

1. Check Integration Connection Before Use

2. Cache Integration List for Performance

3. Handle Special Integrations

4. Monitor Integration Health

API Reference

Methods

Integration Object Structure

Supported Integration Vendors

Next Steps

Secrets Service

Context Graph

Error Handling

Best Practices

​Overview

​Quick Start

​Core Concepts

​Integration Types

​Integration States

​Special Integrations

​Basic Usage

​List All Integrations

​List Connected Integrations Only

​Get Integration Details

​Get Integration Credentials

​Get Special Integration Credentials

​Practical Examples

​1. Integration Health Dashboard

​2. Integration Configuration Validator

​3. Integration Credentials Manager

​4. Multi-Integration Query

​5. Integration Usage Analytics

​Error Handling

​Best Practices

​1. Check Integration Connection Before Use

​2. Cache Integration List for Performance

​3. Handle Special Integrations

​4. Monitor Integration Health

​API Reference

​Methods

​Integration Object Structure

​Supported Integration Vendors

​Next Steps

Secrets Service

Context Graph

Error Handling

Best Practices

Overview

Quick Start

Core Concepts

Integration Types

Integration States

Special Integrations

Basic Usage

List All Integrations

List Connected Integrations Only

Get Integration Details

Get Integration Credentials

Get Special Integration Credentials

Practical Examples

1. Integration Health Dashboard

2. Integration Configuration Validator

3. Integration Credentials Manager

4. Multi-Integration Query

5. Integration Usage Analytics

Error Handling

Best Practices

1. Check Integration Connection Before Use

2. Cache Integration List for Performance

3. Handle Special Integrations

4. Monitor Integration Health

API Reference

Methods

Integration Object Structure

Supported Integration Vendors

Next Steps